jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Feature OpenLookeng ON YARN 

Signed-off-by: fmdewal <fmdewal89@gmail.com>
This commit is contained in:
Faras Mohan Dewal 2021-08-07 05:21:29 -06:00 committed by fmdewal
parent 3b76c590f8
commit d0513cc6ce
26 changed files with 931 additions and 192 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
hetu-metastore/src/test/java/io/hetu/core/metastore/TestHetuMetastoreGlobalCache.java
View File

@ -40,6 +40,7 @@ import io.prestosql.spi.metastore.model.TableEntity;
import io.prestosql.spi.metastore.model.TableEntityType;
import io.prestosql.spi.seedstore.Seed;
import io.prestosql.spi.seedstore.SeedStore;
import io.prestosql.spi.seedstore.SeedStoreSubType;
import io.prestosql.spi.statestore.StateStore;
import io.prestosql.spi.statestore.StateStoreBootstrapper;
import io.prestosql.spi.statestore.StateStoreFactory;
@ -134,7 +135,7 @@ public class TestHetuMetastoreGlobalCache
seeds.add(mockSeed);
SeedStoreManager mockSeedStoreManager = mock(SeedStoreManager.class);
when(mockSeedStoreManager.getSeedStore()).thenReturn(mockSeedStore);
when(mockSeedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST)).thenReturn(mockSeedStore);
when(mockSeed.getLocation()).thenReturn(LOCALHOST + ":" + PORT3);
when(mockSeedStore.get()).thenReturn(seeds);

8
hetu-seed-store/pom.xml
View File

@ -42,6 +42,14 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.airlift</groupId>
<artifactId>json</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-annotations</artifactId>
</dependency>
</dependencies>
<build>
<plugins>

71
hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeed.java
View File

@ -15,6 +15,8 @@
package io.hetu.core.seedstore.filebased;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import io.hetu.core.common.util.SecureObjectInputStream;
import io.prestosql.spi.seedstore.Seed;
@ -37,10 +39,13 @@ public class FileBasedSeed
{
private static final long serialVersionUID = 4L;
private final String location;
private final long timestamp;
private String location;
private long timestamp;
private FileBasedSeed(String location, long timestamp)
@JsonCreator
public FileBasedSeed(
@JsonProperty("location") String location,
@JsonProperty("timestamp") long timestamp)
{
this.location = location;
this.timestamp = timestamp;
@ -66,17 +71,33 @@ public class FileBasedSeed
}
@Override
@JsonProperty
public String getLocation()
{
return location;
}
@Override
@JsonProperty
public void setLocation(String location)
{
this.location = location;
}
@Override
@JsonProperty
public long getTimestamp()
{
return timestamp;
}
@Override
@JsonProperty
public void setTimestamp(long timestamp)
{
this.timestamp = timestamp;
}
@Override
public String serialize()
throws IOException
@ -114,48 +135,4 @@ public class FileBasedSeed
+ "location='" + location + '\''
+ ", timestamp=" + timestamp + '}';
}
/**
* FileBasedSeedBuilder
*
* @since 2020-03-08
*/
public static class FileBasedSeedBuilder
{
private String location;
private long timestamp;
/**
* constructor of FileBasedSeedBuilder
*
* @param location location(eg ip) of seed
*/
public FileBasedSeedBuilder(String location)
{
this.location = location;
}
/**
* Set timestamp for FileBasedSeedBuilder
*
* @param timestamp timestamp of seed
* @return FileBasedSeedBuilder
*/
public FileBasedSeedBuilder setTimestamp(long timestamp)
{
this.timestamp = timestamp;
return this;
}
/**
* build FileBasedSeed Object
*
* @return FileBasedSeed
*/
public FileBasedSeed build()
{
return new FileBasedSeed(location, timestamp);
}
}
}

5
hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedConstants.java
View File

@ -22,9 +22,12 @@ package io.hetu.core.seedstore.filebased;
*/
public class FileBasedSeedConstants
{
// seed file name
// Hazelcast seed file name
static final String SEED_FILE_NAME = "seeds.txt";
// ON-YARN seed file name
static final String ON_YARN_SEED_FILE_NAME = "seeds.json";
// dir config properties name
static final String SEED_STORE_FILESYSTEM_DIR = "seed-store.filesystem.seed-dir";

2
hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStore.java
View File

@ -157,7 +157,7 @@ public class FileBasedSeedStore
throw new NullPointerException("Cannot create filebased seed since location is null or empty");
}
return new FileBasedSeed.FileBasedSeedBuilder(location).setTimestamp(Long.parseLong(timestamp)).build();
return new FileBasedSeed(location, Long.parseLong(timestamp));
}
@Override

10
hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStoreFactory.java
View File

@ -18,6 +18,7 @@ package io.hetu.core.seedstore.filebased;
import io.prestosql.spi.filesystem.HetuFileSystemClient;
import io.prestosql.spi.seedstore.SeedStore;
import io.prestosql.spi.seedstore.SeedStoreFactory;
import io.prestosql.spi.seedstore.SeedStoreSubType;
import java.util.Map;
@ -32,9 +33,14 @@ public class FileBasedSeedStoreFactory
private static final String FACTORY_TYPE = "filebased";
@Override
public SeedStore create(String name, HetuFileSystemClient fs, Map<String, String> config)
public SeedStore create(String name, SeedStoreSubType subType, HetuFileSystemClient fs, Map<String, String> config)
{
return new FileBasedSeedStore(name, fs, config);
if (subType == SeedStoreSubType.ON_YARN) {
return new FileBasedSeedStoreOnYarn(name, fs, config);
}
else {
return new FileBasedSeedStore(name, fs, config);
}
}
@Override

217
hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStoreOnYarn.java Normal file
View File

@ -0,0 +1,217 @@
/*
* Copyright (C) 2018-2020. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.hetu.core.seedstore.filebased;
import com.google.common.collect.ImmutableList;
import io.airlift.json.JsonCodec;
import io.airlift.log.Logger;
import io.hetu.core.common.util.SecurePathWhiteList;
import io.prestosql.spi.filesystem.FileBasedLock;
import io.prestosql.spi.filesystem.HetuFileSystemClient;
import io.prestosql.spi.seedstore.Seed;
import io.prestosql.spi.seedstore.SeedStore;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.stream.Collectors;
import static com.google.common.base.Preconditions.checkArgument;
import static io.airlift.json.JsonCodec.listJsonCodec;
import static java.nio.file.StandardOpenOption.CREATE_NEW;
/**
* FileBasedOnYarnSeedStore
*
* @since 2021-06-23
*/
public class FileBasedSeedStoreOnYarn
implements SeedStore
{
private static final Logger LOG = Logger.get(FileBasedSeedStoreOnYarn.class);
private static final JsonCodec<List<FileBasedSeed>> LIST_FILE_BASED_SEED_CODEC = listJsonCodec(FileBasedSeed.class);
private HetuFileSystemClient fs;
private Map<String, String> config;
private String name; // seed store name represents the sub-folder where a seed file is located
private Path seedFileFolder; // seed file folder
private Path seedFilePath; // seed file full path, = <seedFileFolder>/<name>/seeds.json
/**
* Constructor for file based seed store - on YARN
*
* @param name seed store name
* @param config seed store config
*/
public FileBasedSeedStoreOnYarn(String name, HetuFileSystemClient fs, Map<String, String> config)
{
this.name = name;
this.fs = fs;
this.config = config;
seedFileFolder = Paths.get(config.getOrDefault(FileBasedSeedConstants.SEED_STORE_FILESYSTEM_DIR, FileBasedSeedConstants.SEED_STORE_FILESYSTEM_DIR_DEFAULT_VALUE).trim());
try {
checkArgument(!seedFileFolder.toString().contains("../"),
"SeedStore directory path must be absolute and at user workspace " + SecurePathWhiteList.getSecurePathWhiteList().toString());
checkArgument(SecurePathWhiteList.isSecurePath(seedFileFolder.toString()),
"SeedStore directory path must be at user workspace " + SecurePathWhiteList.getSecurePathWhiteList().toString());
}
catch (IOException e) {
throw new IllegalArgumentException("Failed to get secure path list.", e);
}
seedFilePath = seedFileFolder.resolve(name).resolve(FileBasedSeedConstants.ON_YARN_SEED_FILE_NAME);
}
@Override
public Set<Seed> add(Collection<Seed> seeds)
throws IOException
{
LOG.debug("FileBasedOnYarnSeedStore::add() invoked.");
Lock lock = new FileBasedLock(fs, seedFileFolder.resolve(name));
try {
lock.lock();
Set<FileBasedSeed> latestSeeds = new HashSet<>();
// add all new seeds
latestSeeds.addAll(
seeds.stream()
.filter(s -> (s instanceof FileBasedSeed))
.map(s -> (FileBasedSeed) s)
.collect(Collectors.toList()));
// load existing seeds and filter out repeated seed compared to new seeds
if (fs.exists(seedFilePath)) {
String json = loadFromFile(seedFilePath);
List<FileBasedSeed> existingSeeds = LIST_FILE_BASED_SEED_CODEC.fromJson(json);
latestSeeds.addAll(
existingSeeds.stream()
.filter(s -> !latestSeeds.contains(s))
.collect(Collectors.toList()));
}
String output = LIST_FILE_BASED_SEED_CODEC.toJson(ImmutableList.copyOf(latestSeeds));
writeToFile(seedFilePath, output, true);
return new HashSet<>(latestSeeds);
}
catch (UncheckedIOException e) {
throw new IOException(e);
}
finally {
lock.unlock();
}
}
@Override
public Set<Seed> get()
throws IOException
{
LOG.debug("FileBasedOnYarnSeedStore::get() invoked.");
try {
Set<Seed> outputs = new HashSet<>();
if (fs.exists(seedFilePath)) {
String json = loadFromFile(seedFilePath);
outputs.addAll(LIST_FILE_BASED_SEED_CODEC.fromJson(json));
}
return outputs;
}
catch (UncheckedIOException e) {
throw new IOException(e);
}
}
@Override
public Set<Seed> remove(Collection<Seed> seeds)
throws IOException
{
LOG.debug("FileBasedOnYarnSeedStore::remove() invoked.");
Lock lock = new FileBasedLock(fs, seedFileFolder.resolve(name));
try {
lock.lock();
Set<Seed> outputs = new HashSet<>();
if (fs.exists(seedFilePath)) {
String json = loadFromFile(seedFilePath);
List<FileBasedSeed> existingSeeds = LIST_FILE_BASED_SEED_CODEC.fromJson(json);
Set<FileBasedSeed> latestSeeds = new HashSet<>(existingSeeds);
latestSeeds.removeAll(
seeds.stream()
.filter(s -> (s instanceof FileBasedSeed))
.map(s -> (FileBasedSeed) s)
.collect(Collectors.toList()));
String output = LIST_FILE_BASED_SEED_CODEC.toJson(ImmutableList.copyOf(latestSeeds));
writeToFile(seedFilePath, output, true);
outputs.addAll(latestSeeds);
}
return outputs;
}
catch (UncheckedIOException e) {
throw new IOException(e);
}
finally {
lock.unlock();
}
}
@Override
public Seed create(Map<String, String> properties)
{
String location = properties.get(Seed.LOCATION_PROPERTY_NAME);
String timestamp = properties.get(Seed.TIMESTAMP_PROPERTY_NAME);
// add more properties if interfaces change
if (location == null || location.isEmpty()) {
throw new NullPointerException("Cannot create file-based seed since location is null or empty");
}
return new FileBasedSeed(location, Long.parseLong(timestamp));
}
@Override
public String getName()
{
return name;
}
@Override
public void setName(String name)
{
checkArgument(name.matches("[\\p{Alnum}_\\-]+"), "Invalid cluster name");
this.name = name;
this.seedFilePath = seedFileFolder.resolve(name).resolve(FileBasedSeedConstants.SEED_FILE_NAME);
}
private void writeToFile(Path file, String content, boolean overwrite)
throws IOException
{
try (OutputStream os = (overwrite) ? fs.newOutputStream(file) : fs.newOutputStream(file, CREATE_NEW)) {
os.write(content.getBytes());
}
}
private String loadFromFile(Path file)
throws IOException
{
StringBuilder content = new StringBuilder(0);
try (BufferedReader br = new BufferedReader(new InputStreamReader(fs.newInputStream(file)))) {
br.lines().forEach(content::append);
}
return content.toString();
}
}

9
hetu-seed-store/src/test/java/io/hetu/core/seedstore/filebased/TestFileBasedSeedStore.java
View File

@ -18,6 +18,7 @@ package io.hetu.core.seedstore.filebased;
import io.hetu.core.filesystem.HetuLocalFileSystemClient;
import io.hetu.core.filesystem.LocalConfig;
import io.prestosql.spi.seedstore.Seed;
import io.prestosql.spi.seedstore.SeedStoreSubType;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@ -56,7 +57,7 @@ public class TestFileBasedSeedStore
Map<String, String> config = new HashMap<>(0);
config.put(FileBasedSeedConstants.SEED_STORE_FILESYSTEM_DIR, rootDir);
filebasedSeedStoreFactory = new FileBasedSeedStoreFactory();
seedStore = (FileBasedSeedStore) filebasedSeedStoreFactory.create("filebased",
seedStore = (FileBasedSeedStore) filebasedSeedStoreFactory.create("filebased", SeedStoreSubType.HAZELCAST,
new HetuLocalFileSystemClient(new LocalConfig(new Properties()), Paths.get(rootDir)), config);
seedStore.setName(clusterName);
}
@ -85,8 +86,8 @@ public class TestFileBasedSeedStore
String ip2 = "10.0.0.2";
final long timestamp2 = 2000L;
final int resultSize = 2;
Seed seed1 = new FileBasedSeed.FileBasedSeedBuilder(ip1).setTimestamp(timestamp1).build();
Seed seed2 = new FileBasedSeed.FileBasedSeedBuilder(ip2).setTimestamp(timestamp2).build();
Seed seed1 = new FileBasedSeed(ip1, timestamp1);
Seed seed2 = new FileBasedSeed(ip2, timestamp2);
// add operation
seedStore.add(Lists.newArrayList(seed1, seed2));
@ -110,7 +111,7 @@ public class TestFileBasedSeedStore
// overwrite operation
final long updateTimestamp2 = 3000L;
Seed seed2Update = new FileBasedSeed.FileBasedSeedBuilder(ip2).setTimestamp(updateTimestamp2).build();
Seed seed2Update = new FileBasedSeed(ip2, updateTimestamp2);
seedStore.add(Lists.newArrayList(seed2Update));
results = seedStore.get();
assertEquals(results.size(), 1);

6
hetu-state-store/pom.xml
View File

@ -77,6 +77,10 @@
<groupId>com.fasterxml.jackson.datatype</groupId>
<artifactId>jackson-datatype-jdk8</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-annotations</artifactId>
</dependency>
</dependencies>
<build>
@ -113,4 +117,4 @@
</plugin>
</plugins>
</build>
</project>
</project>

25
hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterLifecycleListener.java
View File

@ -14,6 +14,7 @@
*/
package io.hetu.core.statestore.hazelcast;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.ImmutableSet;
import io.prestosql.spi.seedstore.Seed;
import io.prestosql.spi.seedstore.SeedStore;
@ -106,7 +107,7 @@ public class TestHazelcastClusterLifecycleListener
private static final long serialVersionUID = 4L;
String location;
long timeStamp;
long timestamp;
/**
* Constructor for the mock seed
@ -116,21 +117,37 @@ public class TestHazelcastClusterLifecycleListener
MockSeed(String location)
{
this.location = location;
timeStamp = 0L;
timestamp = 0L;
}
@Override
@JsonProperty
public String getLocation()
{
return location;
}
@Override
@JsonProperty
public void setLocation(String location)
{
this.location = location;
}
@Override
@JsonProperty
public long getTimestamp()
{
return 0L;
}
@Override
@JsonProperty
public void setTimestamp(long timestamp)
{
this.timestamp = timestamp;
}
@Override
public String serialize()
throws IOException
@ -138,9 +155,9 @@ public class TestHazelcastClusterLifecycleListener
return "MOCK SEED. SHOULD NOT SERIALIZE.";
}
public void setTimeStamp(long timeStamp)
public void setTimeStamp(long timestamp)
{
this.timeStamp = timeStamp;
this.timestamp = timestamp;
}
}

2
hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateStoreFactory.java
View File

@ -78,7 +78,7 @@ public class TestHazelcastStateStoreFactory
properties.put(DISCOVERY_MODE_CONFIG_NAME, DISCOVERY_MODE_TCPIP);
SeedStore seedStore = mock(SeedStore.class);
Seed seed = new FileBasedSeed.FileBasedSeedBuilder(MEMBER_ADDRESS).build();
Seed seed = new FileBasedSeed(MEMBER_ADDRESS, 0);
when(seedStore.get()).thenReturn(ImmutableList.of(seed));
setupHazelcastInstance();

67
presto-main/src/main/java/io/prestosql/discovery/HetuServiceInventory.java
View File

@ -26,8 +26,10 @@ import io.airlift.http.client.HttpClient;
import io.airlift.json.JsonCodec;
import io.airlift.log.Logger;
import io.airlift.node.NodeInfo;
import io.prestosql.seedstore.SeedStoreManager;
import io.prestosql.server.InternalCommunicationConfig;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.seedstore.SeedStoreSubType;
import io.prestosql.spi.statestore.StateMap;
import io.prestosql.spi.statestore.StateStore;
import io.prestosql.statestore.StateStoreConstants;
@ -49,11 +51,13 @@ public class HetuServiceInventory
private final HetuConfig hetuConfig;
private final String nodeId;
private final StateStoreProvider stateStoreProvider;
private final SeedStoreManager seedStoreManager;
private final InternalCommunicationConfig internalCommunicationConfig;
private AtomicBoolean serverUp = new AtomicBoolean(true);
@Inject
public HetuServiceInventory(HetuConfig hetuConfig,
SeedStoreManager seedStoreManager,
StateStoreProvider stateStoreProvider,
InternalCommunicationConfig internalCommunicationConfig,
ServiceInventoryConfig config,
@ -64,6 +68,7 @@ public class HetuServiceInventory
super(config, nodeInfo, serviceDescriptorsCodec, httpClient);
this.nodeId = nodeInfo.getNodeId();
this.hetuConfig = hetuConfig;
this.seedStoreManager = seedStoreManager;
this.stateStoreProvider = stateStoreProvider;
this.internalCommunicationConfig = internalCommunicationConfig;
}
@ -71,28 +76,58 @@ public class HetuServiceInventory
@Override
public Iterable<ServiceDescriptor> getServiceDescriptors(String type)
{
if ("discovery".equals(type) && hetuConfig.isMultipleCoordinatorEnabled()) {
try {
StateStore stateStore = stateStoreProvider.getStateStore();
if ("discovery".equals(type)) {
if (hetuConfig.isMultipleCoordinatorEnabled()) {
try {
StateStore stateStore = stateStoreProvider.getStateStore();
if (stateStore == null) {
throw new PrestoException(GENERIC_INTERNAL_ERROR, "State store has not been loaded yet");
}
if (stateStore == null) {
throw new PrestoException(GENERIC_INTERNAL_ERROR, "State store has not been loaded yet");
}
Map.Entry<String, String> entry = ((StateMap<String, String>) stateStore.getStateCollection(StateStoreConstants.DISCOVERY_SERVICE_COLLECTION_NAME))
.getAll().entrySet().stream().findFirst().get();
Map.Entry<String, String> entry = ((StateMap<String, String>) stateStore.getStateCollection(StateStoreConstants.DISCOVERY_SERVICE_COLLECTION_NAME))
.getAll().entrySet().stream().findFirst().get();
ImmutableMap.Builder properties = new ImmutableMap.Builder();
if (internalCommunicationConfig != null && internalCommunicationConfig.isHttpsRequired()) {
properties.put("https", "https://" + entry.getKey() + ":" + entry.getValue());
ImmutableMap.Builder properties = new ImmutableMap.Builder();
if (internalCommunicationConfig != null && internalCommunicationConfig.isHttpsRequired()) {
properties.put("https", "https://" + entry.getKey() + ":" + entry.getValue());
}
else {
properties.put("http", "http://" + entry.getKey() + ":" + entry.getValue());
}
return ImmutableList.of(new ServiceDescriptor(UUID.randomUUID(), nodeId, "discovery", null, null, ServiceState.RUNNING, properties.build()));
}
else {
properties.put("http", "http://" + entry.getKey() + ":" + entry.getValue());
catch (Exception e) {
logServerError("Select service from state store failed: " + e);
// do not return the service descriptors loaded from the configuration file, it might register this node to other cluster.
// return an empty list here.
return ImmutableList.of();
}
return ImmutableList.of(new ServiceDescriptor(UUID.randomUUID(), nodeId, "discovery", null, null, ServiceState.RUNNING, properties.build()));
}
catch (Exception e) {
logServerError("Select service from state store failed:" + e);
else if (seedStoreManager != null && seedStoreManager.isSeedStoreOnYarnEnabled()) {
try {
boolean httpsRequired = (internalCommunicationConfig != null && internalCommunicationConfig.isHttpsRequired());
String location = seedStoreManager.getLatestSeedLocation(SeedStoreSubType.ON_YARN, httpsRequired);
if (location == null || location.isEmpty()) {
throw new PrestoException(GENERIC_INTERNAL_ERROR, "Seed store has not been initialized yet");
}
ImmutableMap.Builder properties = new ImmutableMap.Builder();
if (httpsRequired) {
properties.put("https", location);
}
else {
properties.put("http", location);
}
return ImmutableList.of(new ServiceDescriptor(UUID.randomUUID(), nodeId, "discovery", null, null, ServiceState.RUNNING, properties.build()));
}
catch (Exception e) {
logServerError("Select service from seed store failed: " + e);
// do not return the service descriptors loaded from the configuration file, it might register this node to other cluster.
// return an empty list here.
return ImmutableList.of();
}
}
}
return super.getServiceDescriptors(type);

222
presto-main/src/main/java/io/prestosql/httpserver/HetuHttpServerInfo.java Normal file
View File

@ -0,0 +1,222 @@
/*
* Copyright (C) 2018-2020. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.prestosql.httpserver;
import com.google.common.annotations.VisibleForTesting;
import io.airlift.http.server.HttpServerConfig;
import io.airlift.http.server.HttpServerInfo;
import io.airlift.log.Logger;
import io.airlift.node.NodeInfo;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import javax.inject.Inject;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.lang.reflect.Field;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.channels.ServerSocketChannel;
public class HetuHttpServerInfo
extends HttpServerInfo
{
private static final Logger LOG = Logger.get(HetuHttpServerInfo.class);
private URI httpUri;
private URI httpExternalUri;
private URI httpsUri;
private URI httpsExternalUri;
private URI adminUri;
private URI adminExternalUri;
private ServerSocketChannel httpChannel;
private ServerSocketChannel httpsChannel;
private ServerSocketChannel adminChannel;
private static final int MIN_PORT_NUMBER = 1100;
private static final int MAX_PORT_NUMBER = 65535;
@Inject
public HetuHttpServerInfo(HttpServerConfig config, NodeInfo nodeInfo)
{
super(new HttpServerConfig().setHttpEnabled(false).setHttpsEnabled(false).setAdminEnabled(false), nodeInfo);
// http channel
if (config.isHttpEnabled()) {
this.httpChannel = createChannel(config.getHttpPort(), config.getHttpAcceptQueueSize());
if (config.getHttpPort() != port(this.httpChannel)) {
config.setHttpPort(port(this.httpChannel));
}
this.httpUri = buildUri("http", nodeInfo.getInternalAddress(), port(this.httpChannel));
this.httpExternalUri = buildUri("http", nodeInfo.getExternalAddress(), this.httpUri.getPort());
LOG.info(String.format("HetuHttpServerInfo http channel bind to port %s", port(this.httpChannel)));
}
else {
httpChannel = null;
httpUri = null;
httpExternalUri = null;
}
setBaseChannel(HttpServerInfo.class, "httpChannel", httpChannel);
// https channel
if (config.isHttpsEnabled()) {
this.httpsChannel = createChannel(config.getHttpsPort(), config.getHttpAcceptQueueSize());
if (config.getHttpsPort() != port(this.httpsChannel)) {
config.setHttpsPort(port(this.httpsChannel));
}
this.httpsUri = buildUri("https", nodeInfo.getInternalAddress(), port(this.httpsChannel));
this.httpsExternalUri = buildUri("https", nodeInfo.getExternalAddress(), this.httpsUri.getPort());
LOG.info(String.format("HetuHttpServerInfo https channel bind to port %s", port(this.httpChannel)));
}
else {
httpsChannel = null;
httpsUri = null;
httpsExternalUri = null;
}
setBaseChannel(HttpServerInfo.class, "httpsChannel", httpsChannel);
// admin channel
if (config.isAdminEnabled()) {
this.adminChannel = createChannel(config.getAdminPort(), config.getHttpAcceptQueueSize());
if (config.isHttpsEnabled()) {
this.adminUri = buildUri("https", nodeInfo.getInternalAddress(), port(this.adminChannel));
this.adminExternalUri = buildUri("https", nodeInfo.getExternalAddress(), this.adminUri.getPort());
}
else {
this.adminUri = buildUri("http", nodeInfo.getInternalAddress(), port(this.adminChannel));
this.adminExternalUri = buildUri("http", nodeInfo.getExternalAddress(), this.adminUri.getPort());
}
if (config.getAdminPort() != port(this.adminChannel)) {
config.setAdminPort(port(this.adminChannel));
}
}
else {
adminChannel = null;
adminUri = null;
adminExternalUri = null;
}
setBaseChannel(HttpServerInfo.class, "adminChannel", adminChannel);
}
private void setBaseChannel(Class<HttpServerInfo> clazz, String name, Object value)
{
Field channel = null;
try {
channel = clazz.getDeclaredField(name);
}
catch (NoSuchFieldException e) {
throw new PrestoException(StandardErrorCode.GENERIC_INTERNAL_ERROR, "No such channel found");
}
boolean oldVal = channel.isAccessible();
try {
channel.setAccessible(true);
channel.set(this, value);
}
catch (IllegalAccessException e) {
throw new PrestoException(StandardErrorCode.GENERIC_INTERNAL_ERROR, "Channel not accessible");
}
finally {
channel.setAccessible(oldVal);
}
}
public URI getHttpUri()
{
return this.httpUri;
}
public URI getHttpExternalUri()
{
return this.httpExternalUri;
}
public URI getHttpsUri()
{
return this.httpsUri;
}
public URI getHttpsExternalUri()
{
return this.httpsExternalUri;
}
public URI getAdminUri()
{
return this.adminUri;
}
public URI getAdminExternalUri()
{
return this.adminExternalUri;
}
ServerSocketChannel getHttpChannel()
{
return this.httpChannel;
}
ServerSocketChannel getHttpsChannel()
{
return this.httpsChannel;
}
ServerSocketChannel getAdminChannel()
{
return this.adminChannel;
}
private static URI buildUri(String scheme, String host, int port)
{
try {
return new URI(scheme, (String) null, host, port, (String) null, (String) null, (String) null);
}
catch (URISyntaxException e) {
throw new IllegalArgumentException(e);
}
}
@VisibleForTesting
static int port(ServerSocketChannel channel)
{
try {
return ((InetSocketAddress) channel.getLocalAddress()).getPort();
}
catch (IOException e) {
throw new UncheckedIOException(e);
}
}
private static ServerSocketChannel createChannel(int port, int acceptQueueSize)
{
ServerSocketChannel channel = null;
int nextPort = port;
while (true) {
try {
channel = ServerSocketChannel.open();
channel.socket().setReuseAddress(true);
channel.socket().bind(new InetSocketAddress(nextPort), acceptQueueSize);
return channel;
}
catch (IOException e) {
// cannot bind to this port
nextPort++;
if (nextPort < MIN_PORT_NUMBER || nextPort > MAX_PORT_NUMBER) {
LOG.warn("Failed to bind port withing acceptable limit");
nextPort = MIN_PORT_NUMBER;
}
}
}
}
}

31
presto-main/src/main/java/io/prestosql/httpserver/HetuHttpServerModule.java Normal file
View File

@ -0,0 +1,31 @@
/*
* Copyright (C) 2018-2020. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.prestosql.httpserver;
import com.google.inject.Binder;
import com.google.inject.Module;
import com.google.inject.Scopes;
import io.airlift.http.server.HttpServerInfo;
public class HetuHttpServerModule
implements Module
{
@Override
public void configure(Binder binder)
{
binder.bind(HttpServerInfo.class).to(HetuHttpServerInfo.class).in(Scopes.SINGLETON);
}
}

244
presto-main/src/main/java/io/prestosql/seedstore/SeedStoreManager.java
View File

@ -26,13 +26,16 @@ import io.prestosql.spi.filesystem.HetuFileSystemClient;
import io.prestosql.spi.seedstore.Seed;
import io.prestosql.spi.seedstore.SeedStore;
import io.prestosql.spi.seedstore.SeedStoreFactory;
import io.prestosql.spi.seedstore.SeedStoreSubType;
import io.prestosql.statestore.StateStoreConstants;
import java.io.File;
import java.io.IOException;
import java.nio.file.Paths;
import java.security.InvalidParameterException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
@ -49,6 +52,7 @@ import static io.prestosql.spi.StandardErrorCode.SEED_STORE_FAILURE;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
import static java.util.stream.Collectors.toCollection;
/**
* SeedStoreManager manages the lifecycle of SeedStores
@ -58,31 +62,43 @@ import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
public class SeedStoreManager
{
private static final Logger LOG = Logger.get(SeedStoreManager.class);
// properties name
// configuration files
private static final File SEED_STORE_CONFIGURATION = new File("etc/seed-store.properties");
private static final File STATE_STORE_CONFIGURATION = new File("etc/state-store.properties");
// properties name
private static final String SEED_STORE_ON_YARN_PROPERTY_NAME = "seed-store.on-yarn";
private static final String SEED_STORE_TYPE_PROPERTY_NAME = "seed-store.type";
private static final String SEED_STORE_CLUSTER_PROPERTY_NAME = "seed-store.cluster";
private static final String SEED_STORE_SEED_HEARTBEAT_PROPERTY_NAME = "seed-store.seed.heartbeat";
private static final String SEED_STORE_SEED_HEARTBEAT_TIMEOUT_PROPERTY_NAME = "seed-store.seed.heartbeat.timeout";
private static final String SEED_STORE_FILESYSTEM_PROFILE = "seed-store.filesystem.profile";
// properties default value
private static final boolean SEED_STORE_ON_YARN_ENABLED_DEFAULT_VALUE = false;
private static final String SEED_STORE_TYPE_DEFAULT_VALUE = "filebased";
private static final String SEED_STORE_SEED_HEARTBEAT_DEFAULT_VALUE = "10000"; // 10 seconds
private static final String SEED_STORE_SEED_HEARTBEAT_TIMEOUT_DEFAULT_VALUE = "60000"; // 60 seconds
private static final String SEED_STORE_CLUSTER_DEFAULT_VALUE = "olk_default";
private static final long SEED_STORE_SEED_HEARTBEAT_DEFAULT_VALUE = 10000; // 10 seconds
private static final long SEED_STORE_SEED_HEARTBEAT_TIMEOUT_DEFAULT_VALUE = 60000; // 60 seconds
private static final String SEED_STORE_FILESYSTEM_PROFILE_DEFAULT_VALUE = "hdfs-config-default";
private static final int SEED_RETRY_TIMES = 5;
private static final long SEED_RETRY_INTERVAL = 500L;
private final Map<String, SeedStoreFactory> seedStoreFactories = new ConcurrentHashMap<>();
private final FileSystemClientManager fileSystemClientManager;
private ScheduledExecutorService seedRefreshExecutor = newSingleThreadScheduledExecutor(daemonThreadsNamed("SeedRefresher"));
private SeedStore seedStore;
private String seedStoreType;
private String filesystemProfile;
private HetuFileSystemClient fileSystemClient;
private ScheduledExecutorService seedRefreshExecutor = newSingleThreadScheduledExecutor(daemonThreadsNamed("SeedRefresher"));
private String seedStoreType;
private String clusterName;
private String filesystemProfile;
private boolean isSeedStoreOnYarnEnabled;
private long seedHeartBeat;
private long seedHeartBeatTimeout;
private boolean isSeedStoreEnabled;
private boolean isSeedStoreHazelcastEnabled;
private SeedStore seedStoreOnYarn;
private SeedStore seedStoreHazelcast;
private ConcurrentHashMap<String, Seed> refreshableSeedsMap = new ConcurrentHashMap<>();
@Inject
@ -111,26 +127,90 @@ public class SeedStoreManager
public void loadSeedStore()
throws IOException
{
// load configuration
Map<String, String> config = loadConfiguration(STATE_STORE_CONFIGURATION, SEED_STORE_CONFIGURATION);
// initialize variables
isSeedStoreOnYarnEnabled = SEED_STORE_ON_YARN_ENABLED_DEFAULT_VALUE;
seedStoreType = SEED_STORE_TYPE_DEFAULT_VALUE;
clusterName = SEED_STORE_CLUSTER_DEFAULT_VALUE;
filesystemProfile = SEED_STORE_FILESYSTEM_PROFILE_DEFAULT_VALUE;
seedHeartBeat = SEED_STORE_SEED_HEARTBEAT_DEFAULT_VALUE;
seedHeartBeatTimeout = SEED_STORE_SEED_HEARTBEAT_TIMEOUT_DEFAULT_VALUE;
if (isSeedStoreEnabled) {
LOG.info("-- Loading seed store --");
// create seed store
SeedStoreFactory seedStoreFactory = seedStoreFactories.get(seedStoreType);
checkState(seedStoreFactory != null, "SeedStoreFactory %s is not registered", seedStoreFactory);
try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(seedStoreFactory.getClass().getClassLoader())) {
fileSystemClient = fileSystemClientManager.getFileSystemClient(filesystemProfile, Paths.get("/"));
seedStore = seedStoreFactory.create(seedStoreType,
fileSystemClient,
ImmutableMap.copyOf(config));
Map<String, String> config = new HashMap<>();
// load seed store configuration
if (SEED_STORE_CONFIGURATION.exists()) {
Map<String, String> seedStoreProperties = new HashMap<>(loadPropertiesFrom(SEED_STORE_CONFIGURATION.getPath()));
String propertyValue = seedStoreProperties.get(SEED_STORE_ON_YARN_PROPERTY_NAME);
if (propertyValue != null) {
isSeedStoreOnYarnEnabled = Boolean.parseBoolean(propertyValue);
}
try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(seedStore.getClass().getClassLoader())) {
// start seed refresher
seedRefreshExecutor.scheduleWithFixedDelay(() -> refreshSeeds(), 0, seedHeartBeat, TimeUnit.MILLISECONDS);
seedStoreType = seedStoreProperties.getOrDefault(SEED_STORE_TYPE_PROPERTY_NAME, seedStoreType);
clusterName = seedStoreProperties.getOrDefault(SEED_STORE_CLUSTER_PROPERTY_NAME, clusterName);
filesystemProfile = seedStoreProperties.getOrDefault(SEED_STORE_FILESYSTEM_PROFILE, filesystemProfile);
propertyValue = seedStoreProperties.get(SEED_STORE_SEED_HEARTBEAT_PROPERTY_NAME);
if (propertyValue != null) {
seedHeartBeat = Long.parseLong(propertyValue);
}
propertyValue = seedStoreProperties.get(SEED_STORE_SEED_HEARTBEAT_TIMEOUT_PROPERTY_NAME);
if (propertyValue != null) {
seedHeartBeatTimeout = Long.parseLong(propertyValue);
}
if (seedHeartBeat > seedHeartBeatTimeout) {
throw new InvalidParameterException(format("The value of %s cannot be greater than the value of %s in the property file",
SEED_STORE_SEED_HEARTBEAT_PROPERTY_NAME, SEED_STORE_SEED_HEARTBEAT_TIMEOUT_PROPERTY_NAME));
}
config.putAll(seedStoreProperties);
if (isSeedStoreOnYarnEnabled) {
// create seed store
SeedStoreFactory seedStoreFactory = seedStoreFactories.get(seedStoreType);
checkState(seedStoreFactory != null, "SeedStoreFactory %s is not registered", seedStoreFactory);
if (fileSystemClient == null) {
fileSystemClient = fileSystemClientManager.getFileSystemClient(filesystemProfile, Paths.get("/"));
}
LOG.info("-- Loading seed store on-yarn --");
try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(seedStoreFactory.getClass().getClassLoader())) {
seedStoreOnYarn = seedStoreFactory.create(clusterName,
SeedStoreSubType.ON_YARN,
fileSystemClient,
ImmutableMap.copyOf(config));
}
}
LOG.info("-- Loaded seed store %s --", seedStoreType);
}
// load state store config if exist
if (STATE_STORE_CONFIGURATION.exists()) {
Map<String, String> stateStoreProperties = new HashMap<>(loadPropertiesFrom(STATE_STORE_CONFIGURATION.getPath()));
// for now, seed store is started only if tcp-ip mode enabled and tcp-ip.seeds is not set
String discoveryMode = stateStoreProperties.get(StateStoreConstants.DISCOVERY_MODE_PROPERTY_NAME);
isSeedStoreHazelcastEnabled = discoveryMode != null && discoveryMode.equals(StateStoreConstants.DISCOVERY_MODE_TCPIP)
&& stateStoreProperties.get(StateStoreConstants.HAZELCAST_DISCOVERY_TCPIP_SEEDS) == null;
if (isSeedStoreHazelcastEnabled) {
// create seed store
SeedStoreFactory seedStoreFactory = seedStoreFactories.get(seedStoreType);
checkState(seedStoreFactory != null, "SeedStoreFactory %s is not registered", seedStoreFactory);
if (fileSystemClient == null) {
String stateStoreFilesystemProfile = stateStoreProperties.getOrDefault(StateStoreConstants.HAZELCAST_DISCOVERY_TCPIP_PROFILE, filesystemProfile);
fileSystemClient = fileSystemClientManager.getFileSystemClient(stateStoreFilesystemProfile, Paths.get("/"));
}
LOG.info("-- Loading seed store hazelcast --");
try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(seedStoreFactory.getClass().getClassLoader())) {
seedStoreHazelcast = seedStoreFactory.create(clusterName,
SeedStoreSubType.HAZELCAST,
fileSystemClient,
ImmutableMap.copyOf(config));
}
try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(seedStoreHazelcast.getClass().getClassLoader())) {
// start seed refresher
seedRefreshExecutor.scheduleWithFixedDelay(() -> refreshSeeds(SeedStoreSubType.HAZELCAST), 0, seedHeartBeat, TimeUnit.MILLISECONDS);
}
}
}
}
/**
* Check if seed-store on-YARN is enabled
*/
public boolean isSeedStoreOnYarnEnabled()
{
return isSeedStoreOnYarnEnabled;
}
/**
@ -148,9 +228,10 @@ public class SeedStoreManager
* @return a collection of seeds in the seed store
* @throws IOException
*/
public Collection<Seed> getAllSeeds()
public Collection<Seed> getAllSeeds(SeedStoreSubType subType)
throws IOException
{
SeedStore seedStore = getSeedStore(subType);
if (seedStore == null) {
throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null");
}
@ -160,6 +241,43 @@ public class SeedStoreManager
return seeds;
}
/**
* Get all seeds from seed store
*
* @return the latest seed location in the seed store - https or http
* @throws IOException
*/
public String getLatestSeedLocation(SeedStoreSubType subType, boolean httpsRequired)
throws IOException
{
SeedStore seedStore = getSeedStore(subType);
if (seedStore == null) {
throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null");
}
Collection<Seed> seeds = seedStore.get();
if (seeds.isEmpty()) {
return null;
}
ArrayList<Seed> list = seeds.stream()
.filter(seed -> {
if (httpsRequired) {
return seed.getLocation().contains("https:");
}
else {
return seed.getLocation().contains("http:");
}
})
.sorted(Comparator.comparing(Seed::getTimestamp))
.collect(toCollection(ArrayList::new));
if (list.isEmpty()) {
return null;
}
else {
return list.get(list.size() - 1).getLocation();
}
}
/**
* Add seed to seed store. If refreshable is enabled, seed will be refreshed periodically
*
@ -168,11 +286,12 @@ public class SeedStoreManager
* @throws IOException
*/
public Collection<Seed> addSeed(String seedLocation, boolean refreshable)
public Collection<Seed> addSeed(SeedStoreSubType subType, String seedLocation, boolean refreshable)
throws IOException
{
Collection<Seed> seeds = new HashSet<>();
SeedStore seedStore = getSeedStore(subType);
if (seedStore == null) {
throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null");
}
@ -180,7 +299,7 @@ public class SeedStoreManager
Seed seed = seedStore.create(ImmutableMap.of(
Seed.LOCATION_PROPERTY_NAME, seedLocation,
Seed.TIMESTAMP_PROPERTY_NAME, String.valueOf(System.currentTimeMillis())));
seeds = addSeed(seed);
seeds = addSeed(subType, seed);
if (refreshable) {
refreshableSeedsMap.put(seedLocation, seed);
@ -198,11 +317,12 @@ public class SeedStoreManager
* @throws IOException
*/
public Collection<Seed> removeSeed(String seedLocation)
public Collection<Seed> removeSeed(SeedStoreSubType subType, String seedLocation)
throws IOException
{
Collection<Seed> seeds = new HashSet<>();
SeedStore seedStore = getSeedStore(subType);
if (seedStore == null) {
throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null");
}
@ -223,9 +343,10 @@ public class SeedStoreManager
*
* @throws IOException
*/
public void clearExpiredSeeds()
public void clearExpiredSeeds(SeedStoreSubType subType)
throws IOException
{
SeedStore seedStore = getSeedStore(subType);
if (seedStore == null) {
throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null");
}
@ -244,13 +365,17 @@ public class SeedStoreManager
}
}
private Collection<Seed> addSeed(Seed seed)
private Collection<Seed> addSeed(SeedStoreSubType subType, Seed seed)
throws IOException
{
int retryTimes = 0;
long retryInterval = 0L;
Collection<Seed> seeds = null;
SeedStore seedStore = getSeedStore(subType);
if (seedStore == null) {
throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null");
}
do {
try {
TimeUnit.MILLISECONDS.sleep(retryInterval);
@ -280,9 +405,14 @@ public class SeedStoreManager
*
* Set the local SeedStore
* */
public void setSeedStore(SeedStore seedStore)
public void setSeedStore(SeedStoreSubType subType, SeedStore seedStore)
{
this.seedStore = seedStore;
if (subType == SeedStoreSubType.ON_YARN) {
seedStoreOnYarn = seedStore;
}
else {
seedStoreHazelcast = seedStore;
}
}
/**
@ -290,13 +420,18 @@ public class SeedStoreManager
*
* @return loaded SeedStore or null if it's not loaded
*/
public SeedStore getSeedStore()
public SeedStore getSeedStore(SeedStoreSubType subType)
{
SeedStore seedStore = (subType == SeedStoreSubType.ON_YARN) ? seedStoreOnYarn : seedStoreHazelcast;
return seedStore;
}
private void refreshSeeds()
private void refreshSeeds(SeedStoreSubType subType)
{
SeedStore seedStore = getSeedStore(subType);
if (seedStore == null) {
throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null");
}
for (Map.Entry<String, Seed> entry : refreshableSeedsMap.entrySet()) {
long newTime = System.currentTimeMillis();
LOG.debug("seed=%s refresh with oldTimestamp=%s and newTimestamp=%s", entry.getKey(), entry.getValue().getTimestamp(), newTime);
@ -314,45 +449,4 @@ public class SeedStoreManager
}
}
}
private Map<String, String> loadConfiguration(File stateStoreConfig, File seedStoreConfig)
throws IOException
{
Map<String, String> properties = new HashMap<>();
// initialize variables
seedStoreType = SEED_STORE_TYPE_DEFAULT_VALUE;
filesystemProfile = SEED_STORE_FILESYSTEM_PROFILE_DEFAULT_VALUE;
seedHeartBeat = Long.parseLong(SEED_STORE_SEED_HEARTBEAT_DEFAULT_VALUE);
seedHeartBeatTimeout = Long.parseLong(SEED_STORE_SEED_HEARTBEAT_TIMEOUT_DEFAULT_VALUE);
// load state store config if exist
if (stateStoreConfig.exists()) {
Map<String, String> stateStoreProperties = new HashMap<>(loadPropertiesFrom(stateStoreConfig.getPath()));
filesystemProfile = stateStoreProperties.getOrDefault(StateStoreConstants.HAZELCAST_DISCOVERY_TCPIP_PROFILE, filesystemProfile);
// for now, seed store is started only if tcp-ip mode enabled and tcp-ip.seeds is not set
String discoveryMode = stateStoreProperties.get(StateStoreConstants.DISCOVERY_MODE_PROPERTY_NAME);
isSeedStoreEnabled = discoveryMode != null && discoveryMode.equals(StateStoreConstants.DISCOVERY_MODE_TCPIP)
&& stateStoreProperties.get(StateStoreConstants.HAZELCAST_DISCOVERY_TCPIP_SEEDS) == null;
}
// load seed store config if exist
if (seedStoreConfig.exists()) {
Map<String, String> seedStoreProperties = new HashMap<>(loadPropertiesFrom(seedStoreConfig.getPath()));
properties.putAll(seedStoreProperties);
seedStoreType = properties.getOrDefault(SEED_STORE_TYPE_PROPERTY_NAME, seedStoreType);
filesystemProfile = properties.getOrDefault(SEED_STORE_FILESYSTEM_PROFILE, filesystemProfile);
if (properties.get(SEED_STORE_SEED_HEARTBEAT_PROPERTY_NAME) != null) {
seedHeartBeat = Long.parseLong(properties.get(SEED_STORE_SEED_HEARTBEAT_PROPERTY_NAME));
}
if (properties.get(SEED_STORE_SEED_HEARTBEAT_TIMEOUT_PROPERTY_NAME) != null) {
seedHeartBeatTimeout = Long.parseLong(properties.get(SEED_STORE_SEED_HEARTBEAT_TIMEOUT_PROPERTY_NAME));
}
if (seedHeartBeat > seedHeartBeatTimeout) {
throw new InvalidParameterException(format("The value of %s cannot be greater than the value of %s in the property file",
SEED_STORE_SEED_HEARTBEAT_PROPERTY_NAME, SEED_STORE_SEED_HEARTBEAT_TIMEOUT_PROPERTY_NAME));
}
}
return properties;
}
}

41
presto-main/src/main/java/io/prestosql/server/PrestoServer.java
View File

@ -22,6 +22,7 @@ import io.airlift.discovery.client.Announcer;
import io.airlift.discovery.client.DiscoveryModule;
import io.airlift.event.client.JsonEventModule;
import io.airlift.event.client.http.HttpEventModule;
import io.airlift.http.server.HttpServerInfo;
import io.airlift.http.server.HttpServerModule;
import io.airlift.jaxrs.JaxrsModule;
import io.airlift.jmx.JmxHttpModule;
@ -43,6 +44,8 @@ import io.prestosql.execution.scheduler.NodeSchedulerConfig;
import io.prestosql.execution.warnings.WarningCollectorModule;
import io.prestosql.filesystem.FileSystemClientManager;
import io.prestosql.heuristicindex.HeuristicIndexerManager;
import io.prestosql.httpserver.HetuHttpServerInfo;
import io.prestosql.httpserver.HetuHttpServerModule;
import io.prestosql.jmx.HetuJmxModule;
import io.prestosql.metadata.StaticCatalogStore;
import io.prestosql.metadata.StaticFunctionNamespaceStore;
@ -55,6 +58,7 @@ import io.prestosql.seedstore.SeedStoreManager;
import io.prestosql.server.security.PasswordAuthenticatorManager;
import io.prestosql.server.security.ServerSecurityModule;
import io.prestosql.snapshot.SnapshotUtils;
import io.prestosql.spi.seedstore.SeedStoreSubType;
import io.prestosql.sql.parser.SqlParserOptions;
import io.prestosql.statestore.StateStoreLauncher;
import io.prestosql.statestore.StateStoreProvider;
@ -106,7 +110,7 @@ public class PrestoServer
modules.add(
new NodeModule(),
Modules.override(new DiscoveryModule()).with(new HetuDiscoveryModule()),
new HttpServerModule(),
Modules.override(new HttpServerModule()).with(new HetuHttpServerModule()),
new JsonModule(),
new SmileModule(),
new JaxrsModule(),
@ -156,6 +160,13 @@ public class PrestoServer
injector.getInstance(PasswordAuthenticatorManager.class).loadPasswordAuthenticator();
injector.getInstance(EventListenerManager.class).loadConfiguredEventListener();
if (injector.getInstance(SeedStoreManager.class).isSeedStoreOnYarnEnabled()) {
addSeedOnYarnInformation(
injector.getInstance(ServerConfig.class),
injector.getInstance(SeedStoreManager.class),
(HetuHttpServerInfo) injector.getInstance(HttpServerInfo.class));
}
// preload index (on coordinator only)
if (injector.getInstance(ServerConfig.class).isCoordinator()) {
injector.getInstance(HeuristicIndexerManager.class).preloadIndex();
@ -187,6 +198,34 @@ public class PrestoServer
return ImmutableList.of();
}
private static void addSeedOnYarnInformation(ServerConfig serverConfig,
SeedStoreManager seedStoreManager,
HetuHttpServerInfo httpServerInfo)
{
if (serverConfig == null || seedStoreManager == null || httpServerInfo == null) {
return;
}
if (!serverConfig.isCoordinator()) {
return;
}
String httpUri;
if (httpServerInfo.getHttpExternalUri() != null) {
httpUri = httpServerInfo.getHttpExternalUri().toString();
}
else if (httpServerInfo.getHttpsExternalUri() != null) {
httpUri = httpServerInfo.getHttpsExternalUri().toString();
}
else {
return;
}
try {
seedStoreManager.addSeed(SeedStoreSubType.ON_YARN, httpUri, false);
}
catch (IOException e) {
return;
}
}
private static void launchEmbeddedStateStore(HetuConfig config, StateStoreLauncher launcher)
throws Exception
{

43
presto-main/src/main/java/io/prestosql/statestore/EmbeddedStateStoreLauncher.java
View File

@ -22,12 +22,17 @@ import io.prestosql.seedstore.SeedStoreManager;
import io.prestosql.server.InternalCommunicationConfig;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.filesystem.FileBasedLock;
import io.prestosql.spi.seedstore.SeedStoreSubType;
import io.prestosql.spi.statestore.StateCollection;
import io.prestosql.spi.statestore.StateMap;
import io.prestosql.spi.statestore.StateStore;
import io.prestosql.spi.statestore.StateStoreBootstrapper;
import io.prestosql.utils.HetuConfig;
import javax.net.ServerSocketFactory;
import javax.net.ssl.SSLServerSocket;
import javax.net.ssl.SSLServerSocketFactory;
import java.io.File;
import java.io.IOException;
import java.net.URI;
@ -111,11 +116,11 @@ public class EmbeddedStateStoreLauncher
if (staticSeeds.size() > 0) {
launchStateStore(staticSeeds, properties);
}
else if (seedStoreManager.getSeedStore() != null) {
else if (seedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST) != null) {
// Set seed store name
seedStoreManager.getSeedStore().setName(properties.get(STATE_STORE_CLUSTER_PROPERTY_NAME));
seedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST).setName(properties.get(STATE_STORE_CLUSTER_PROPERTY_NAME));
// Clear expired seeds
seedStoreManager.clearExpiredSeeds();
seedStoreManager.clearExpiredSeeds(SeedStoreSubType.HAZELCAST);
// Use lock to control synchronization of state store launch among all coordinators
Lock launcherLock = new FileBasedLock(seedStoreManager.getFileSystemClient(), Paths.get(LAUNCHER_LOCK_FILE_PATH));
try {
@ -142,18 +147,42 @@ public class EmbeddedStateStoreLauncher
private void launchStateStoreFromSeedStore(Map<String, String> properties) throws IOException
{
// Get all seeds
Set<String> locations = seedStoreManager.getAllSeeds()
Set<String> locations = seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST)
.stream()
.map(x -> x.getLocation())
.collect(Collectors.toSet());
String launcherPort = getStateStoreLauncherPort(properties);
requireNonNull(launcherPort, "The launcher port is null");
// Detect port conflict and get the next usable one
launcherPort = checkAndGetAvailablePort(launcherPort);
properties.put(HAZELCAST_DISCOVERY_PORT_PROPERTY_NAME, launcherPort);
// Launch state store
String currentLocation = getNodeUri().getHost() + ":" + launcherPort;
locations.add(currentLocation);
if (launchStateStore(locations, properties) != null) {
// Add seed to seed store if and only if state store launched successfully
seedStoreManager.addSeed(currentLocation, true);
seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, currentLocation, true);
}
}
private String checkAndGetAvailablePort(String port)
{
int nextPort = Integer.parseInt(port);
while (!isPortAvailable(nextPort)) {
nextPort++;
}
return String.valueOf(nextPort);
}
private static boolean isPortAvailable(int port)
{
ServerSocketFactory sslServerSocketFactory = SSLServerSocketFactory.getDefault();
try (SSLServerSocket socket = (SSLServerSocket) sslServerSocketFactory.createServerSocket(port)) {
return true;
}
catch (IOException e) {
return false;
}
}
@ -277,9 +306,9 @@ public class EmbeddedStateStoreLauncher
registerDiscoveryService(failureMemberHost);
}
if (seedStoreManager.getSeedStore() != null) {
if (seedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST) != null) {
try {
seedStoreManager.removeSeed((String) failureMember);
seedStoreManager.removeSeed(SeedStoreSubType.HAZELCAST, (String) failureMember);
}
catch (Exception e) {
LOG.error("Cannot remove failure node %s from seed store: %s", failureMember, e.getMessage());

3
presto-main/src/main/java/io/prestosql/statestore/LocalStateStoreProvider.java
View File

@ -21,6 +21,7 @@ import io.prestosql.metastore.MetaStoreConstants;
import io.prestosql.seedstore.SeedStoreManager;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.classloader.ThreadContextClassLoader;
import io.prestosql.spi.seedstore.SeedStoreSubType;
import io.prestosql.spi.statestore.StateCollection;
import io.prestosql.spi.statestore.StateStore;
import io.prestosql.spi.statestore.StateStoreFactory;
@ -100,7 +101,7 @@ public class LocalStateStoreProvider
stateStoreName = DEFAULT_STATE_STORE_NAME;
}
// Create state stores defined in config
stateStore = stateStoreFactory.create(stateStoreName, seedStoreManager.getSeedStore(), ImmutableMap.copyOf(properties));
stateStore = stateStoreFactory.create(stateStoreName, seedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST), ImmutableMap.copyOf(properties));
stateStore.registerClusterFailureHandler(this::handleClusterDisconnection);
stateStore.init();
}

2
presto-main/src/test/java/io/prestosql/discovery/TestHetuServiceInventory.java
View File

@ -59,6 +59,7 @@ public class TestHetuServiceInventory
hetuConfig.setMultipleCoordinatorEnabled(true);
HetuServiceInventory inventory = new HetuServiceInventory(hetuConfig,
null,
createMockStateStoreProvider(),
internalCommunicationConfig,
new ServiceInventoryConfig(),
@ -86,6 +87,7 @@ public class TestHetuServiceInventory
hetuConfig.setMultipleCoordinatorEnabled(true);
HetuServiceInventory inventory = new HetuServiceInventory(hetuConfig,
null,
createMockStateStoreProvider(),
internalCommunicationConfig,
new ServiceInventoryConfig(),

3
presto-main/src/test/java/io/prestosql/operator/TestDynamicFilterSourceOperator.java
View File

@ -35,6 +35,7 @@ import io.prestosql.spi.plan.PlanNodeId;
import io.prestosql.spi.plan.Symbol;
import io.prestosql.spi.seedstore.Seed;
import io.prestosql.spi.seedstore.SeedStore;
import io.prestosql.spi.seedstore.SeedStoreSubType;
import io.prestosql.spi.statestore.StateSet;
import io.prestosql.spi.statestore.StateStore;
import io.prestosql.spi.statestore.StateStoreBootstrapper;
@ -140,7 +141,7 @@ public class TestDynamicFilterSourceOperator
seeds.add(mockSeed);
SeedStoreManager mockSeedStoreManager = mock(SeedStoreManager.class);
when(mockSeedStoreManager.getSeedStore()).thenReturn(mockSeedStore);
when(mockSeedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST)).thenReturn(mockSeedStore);
when(mockSeed.getLocation()).thenReturn("127.0.0.1:6991");
when(mockSeedStore.get()).thenReturn(seeds);

58
presto-main/src/test/java/io/prestosql/seedstore/TestSeedStoreManager.java
View File

@ -14,12 +14,14 @@
*/
package io.prestosql.seedstore;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.ImmutableSet;
import io.prestosql.filesystem.FileSystemClientManager;
import io.prestosql.spi.filesystem.HetuFileSystemClient;
import io.prestosql.spi.seedstore.Seed;
import io.prestosql.spi.seedstore.SeedStore;
import io.prestosql.spi.seedstore.SeedStoreFactory;
import io.prestosql.spi.seedstore.SeedStoreSubType;
import io.prestosql.testing.assertions.Assert;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.BeforeTest;
@ -90,6 +92,7 @@ public class TestSeedStoreManager
SeedStoreFactory mockSeedStoreFactory = mock(SeedStoreFactory.class);
when(mockSeedStoreFactory.getName()).thenReturn("filebased");
when(mockSeedStoreFactory.create(any(String.class),
SeedStoreSubType.HAZELCAST,
any(HetuFileSystemClient.class),
any(Map.class))).thenReturn(mockSeedStore);
seedStoreManager.addSeedStoreFactory(mockSeedStoreFactory);
@ -100,7 +103,7 @@ public class TestSeedStoreManager
throws IOException
{
seedStoreManager.loadSeedStore();
SeedStore returned = seedStoreManager.getSeedStore();
SeedStore returned = seedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST);
}
@Test
@ -110,9 +113,9 @@ public class TestSeedStoreManager
String location1 = "location1";
String location2 = "location2";
seedStoreManager.loadSeedStore();
seedStoreManager.addSeed(location1, false);
seedStoreManager.addSeed(location2, true);
Collection<Seed> result = seedStoreManager.getAllSeeds();
seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, location1, false);
seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, location2, true);
Collection<Seed> result = seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST);
Assert.assertEquals(result.size(), 2);
Assert.assertTrue(result.stream().anyMatch(s -> s.getLocation().equals(location1)));
Assert.assertTrue(result.stream().anyMatch(s -> s.getLocation().equals(location2)));
@ -121,7 +124,7 @@ public class TestSeedStoreManager
long timestamp2Old = result.stream().filter(s -> s.getLocation().equals(location2)).findAny().get().getTimestamp();
//wait 2 seconds, seed2 will be updated with new timestamp
Thread.sleep(2000);
result = seedStoreManager.getAllSeeds();
result = seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST);
long timestamp1New = result.stream().filter(s -> s.getLocation().equals(location1)).findAny().get().getTimestamp();
long timestamp2New = result.stream().filter(s -> s.getLocation().equals(location2)).findAny().get().getTimestamp();
Assert.assertTrue(timestamp1Old == timestamp1New);
@ -136,12 +139,12 @@ public class TestSeedStoreManager
String location2 = "location2";
String location3 = "location3";
seedStoreManager.loadSeedStore();
seedStoreManager.addSeed(location1, false);
seedStoreManager.addSeed(location2, true);
Assert.assertEquals(seedStoreManager.getAllSeeds().size(), 2);
seedStoreManager.removeSeed(location1);
seedStoreManager.removeSeed(location3);
Collection<Seed> result = seedStoreManager.getAllSeeds();
seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, location1, false);
seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, location2, true);
Assert.assertEquals(seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST).size(), 2);
seedStoreManager.removeSeed(SeedStoreSubType.HAZELCAST, location1);
seedStoreManager.removeSeed(SeedStoreSubType.HAZELCAST, location3);
Collection<Seed> result = seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST);
Assert.assertEquals(result.size(), 1);
Assert.assertFalse(result.stream().filter(s -> s.getLocation().equals(location1)).findAny().isPresent());
}
@ -153,13 +156,13 @@ public class TestSeedStoreManager
String location1 = "location1";
String location2 = "location2";
seedStoreManager.loadSeedStore();
seedStoreManager.addSeed(location1, false);
seedStoreManager.addSeed(location2, true);
Assert.assertEquals(seedStoreManager.getAllSeeds().size(), 2);
seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, location1, false);
seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, location2, true);
Assert.assertEquals(seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST).size(), 2);
//wait 5 seconds, location1 expired since refreshable is not enabled
Thread.sleep(5000);
seedStoreManager.clearExpiredSeeds();
Collection<Seed> result = seedStoreManager.getAllSeeds();
seedStoreManager.clearExpiredSeeds(SeedStoreSubType.HAZELCAST);
Collection<Seed> result = seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST);
Assert.assertEquals(result.size(), 1);
Assert.assertFalse(result.stream().filter(s -> s.getLocation().equals(location1)).findAny().isPresent());
}
@ -188,7 +191,7 @@ public class TestSeedStoreManager
}};
createConfigFile("etc/state-store.properties", stateStoreConfigs);
seedStoreManager.loadSeedStore();
Assert.assertNull(seedStoreManager.getSeedStore());
Assert.assertNull(seedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST));
// reset config files
prepareConfigFiles();
}
@ -231,17 +234,33 @@ public class TestSeedStoreManager
}
@Override
@JsonProperty
public String getLocation()
{
return location;
}
@Override
@JsonProperty
public void setLocation(String location)
{
this.location = location;
}
@Override
@JsonProperty
public long getTimestamp()
{
return timestamp;
}
@Override
@JsonProperty
public void setTimestamp(long timestamp)
{
this.timestamp = timestamp;
}
@Override
public String serialize()
throws IOException
@ -249,11 +268,6 @@ public class TestSeedStoreManager
return "MOCK SEED. SHOULD NOT SERIALIZE.";
}
public void setTimestamp(long timestamp)
{
this.timestamp = timestamp;
}
@Override
public boolean equals(Object obj)
{

9
presto-main/src/test/java/io/prestosql/statestore/TestStateStoreLauncherAndProvider.java
View File

@ -26,6 +26,7 @@ import io.prestosql.seedstore.SeedStoreManager;
import io.prestosql.server.InternalCommunicationConfig;
import io.prestosql.spi.seedstore.Seed;
import io.prestosql.spi.seedstore.SeedStore;
import io.prestosql.spi.seedstore.SeedStoreSubType;
import io.prestosql.spi.statestore.StateStore;
import io.prestosql.spi.statestore.StateStoreBootstrapper;
import io.prestosql.spi.statestore.StateStoreFactory;
@ -109,7 +110,7 @@ public class TestStateStoreLauncherAndProvider
seeds.add(mockSeed);
SeedStoreManager mockSeedStoreManager = mock(SeedStoreManager.class);
when(mockSeedStoreManager.getSeedStore()).thenReturn(mockSeedStore);
when(mockSeedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST)).thenReturn(mockSeedStore);
when(mockSeed.getLocation()).thenReturn(LOCALHOST + ":" + PORT3);
when(mockSeedStore.get()).thenReturn(seeds);
@ -136,8 +137,8 @@ public class TestStateStoreLauncherAndProvider
when(mockSeedStore.get()).thenReturn(seeds);
SeedStoreManager mockSeedStoreManager = mock(SeedStoreManager.class);
when(mockSeedStoreManager.getSeedStore()).thenReturn(mockSeedStore);
when(mockSeedStoreManager.addSeed(LOCALHOST, true)).thenReturn(seeds);
when(mockSeedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST)).thenReturn(mockSeedStore);
when(mockSeedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, LOCALHOST, true)).thenReturn(seeds);
when(mockSeedStoreManager.getFileSystemClient()).thenReturn(new HetuLocalFileSystemClient(new LocalConfig(new Properties()), Paths.get("/")));
InternalCommunicationConfig mockInternalCommunicationConfig = mock(InternalCommunicationConfig.class);
@ -157,7 +158,7 @@ public class TestStateStoreLauncherAndProvider
// mock "remove" second instance from cluster (delete from seed store)
seeds.remove(mockSeed2);
when(mockSeed1.getLocation()).thenReturn(LOCALHOST + ":" + PORT1);
when(mockSeedStoreManager.addSeed(LOCALHOST, true)).thenReturn(seeds);
when(mockSeedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, LOCALHOST, true)).thenReturn(seeds);
((HazelcastStateStore) second).shutdown();
// Allow the first node to handle failure

14
presto-spi/src/main/java/io/prestosql/spi/seedstore/Seed.java
View File

@ -42,6 +42,13 @@ public interface Seed
*/
String getLocation();
/**
* St location of seed
*
* @return void
*/
void setLocation(String location);
/**
* Get timestamp of seed
*
@ -49,6 +56,13 @@ public interface Seed
*/
long getTimestamp();
/**
* Set timestamp of seed
*
* @return void
*/
void setTimestamp(long timestamp);
/**
* Serialize seed object to string and return
*

2
presto-spi/src/main/java/io/prestosql/spi/seedstore/SeedStoreFactory.java
View File

@ -40,5 +40,5 @@ public interface SeedStoreFactory
* @param config seed store configurations
* @return created seed store
*/
SeedStore create(String name, HetuFileSystemClient fs, Map<String, String> config);
SeedStore create(String name, SeedStoreSubType subType, HetuFileSystemClient fs, Map<String, String> config);
}

21
presto-spi/src/main/java/io/prestosql/spi/seedstore/SeedStoreSubType.java Normal file
View File

@ -0,0 +1,21 @@
/*
* Copyright (C) 2018-2020. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.prestosql.spi.seedstore;
public enum SeedStoreSubType {
DEFAULT,
ON_YARN,
HAZELCAST
}

5
presto-tests/src/main/java/io/prestosql/tests/DistributedQueryRunnerWithStateStore.java
View File

@ -28,6 +28,7 @@ import io.prestosql.seedstore.SeedStoreManager;
import io.prestosql.server.testing.TestingPrestoServer;
import io.prestosql.spi.seedstore.Seed;
import io.prestosql.spi.seedstore.SeedStore;
import io.prestosql.spi.seedstore.SeedStoreSubType;
import io.prestosql.spi.statestore.StateStore;
import io.prestosql.sql.parser.SqlParserOptions;
import io.prestosql.statestore.EmbeddedStateStoreLauncher;
@ -107,7 +108,7 @@ public class DistributedQueryRunnerWithStateStore
launcher.launchStateStore();
StateStoreProvider provider = server.getInstance(Key.get(StateStoreProvider.class));
Seed seed = new FileBasedSeed.FileBasedSeedBuilder("127.0.0.1:" + port).build();
Seed seed = new FileBasedSeed("127.0.0.1:" + port, 0);
SeedStore seedStore = new SeedStore()
{
@Override
@ -148,7 +149,7 @@ public class DistributedQueryRunnerWithStateStore
{
}
};
server.getInstance(Key.get(SeedStoreManager.class)).setSeedStore(seedStore);
server.getInstance(Key.get(SeedStoreManager.class)).setSeedStore(SeedStoreSubType.HAZELCAST, seedStore);
if (provider instanceof LocalStateStoreProvider) {
Map<String, String> stateStoreProperties = new HashMap<>();
stateStoreProperties.putIfAbsent(HazelcastConstants.DISCOVERY_MODE_CONFIG_NAME, HazelcastConstants.DISCOVERY_MODE_TCPIP);