diff --git a/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastClusterMembershipListener.java b/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastClusterMembershipListener.java index a29a90982..b2a23a23f 100644 --- a/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastClusterMembershipListener.java +++ b/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastClusterMembershipListener.java @@ -56,6 +56,9 @@ public class HazelcastClusterMembershipListener membershipEvent.getMember().getAddress().getHost(), membershipEvent.getMember().getAddress().getPort()); - memberRemovedHandler.accept(membershipEvent.getMember().getAddress().getHost()); + memberRemovedHandler.accept( + membershipEvent.getMember().getAddress().getHost() + + ":" + + membershipEvent.getMember().getAddress().getPort()); } } diff --git a/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastConstants.java b/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastConstants.java index 7c68f7eb5..fef03c846 100644 --- a/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastConstants.java +++ b/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastConstants.java @@ -27,9 +27,14 @@ public final class HazelcastConstants public static final String DISCOVERY_MODE_CONFIG_NAME = "hazelcast.discovery.mode"; /** - * Hazelcast TCP-IP port config name + * Hazelcast discovery port config name */ - public static final String PORT_CONFIG_NAME = "hazelcast.tcp-ip.port"; + public static final String DISCOVERY_PORT_CONFIG_NAME = "hazelcast.discovery.port"; + + /** + * Hazelcast default discovery port + */ + public static final String DEFAULT_DISCOVERY_PORT = "5701"; /** * Hazelcast multicast discovery mode diff --git a/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreBootstrapper.java b/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreBootstrapper.java index f73becca6..6ce4e75de 100644 --- a/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreBootstrapper.java +++ b/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreBootstrapper.java @@ -37,13 +37,14 @@ import java.util.Map; import static io.hetu.core.statestore.Constants.STATE_STORE_CLUSTER_CONFIG_NAME; import static io.hetu.core.statestore.StateStoreUtils.getEncryptionTypeFromConfig; import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DEFAULT_CLUSTER_ID; +import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DEFAULT_DISCOVERY_PORT; import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_ENABLED; import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_MODE_CONFIG_NAME; import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_MODE_MULTICAST; import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_MODE_TCPIP; import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_MULTICAST_STRATEGY_CLASS_NAME; +import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_PORT_CONFIG_NAME; import static io.hetu.core.statestore.hazelcast.HazelcastConstants.MINIMUM_CP_MEMBER_COUNT; -import static io.hetu.core.statestore.hazelcast.HazelcastConstants.PORT_CONFIG_NAME; import static io.prestosql.spi.StandardErrorCode.CONFIGURATION_INVALID; import static io.prestosql.spi.StandardErrorCode.STATE_STORE_FAILURE; @@ -62,7 +63,7 @@ public class HazelcastStateStoreBootstrapper private static final int TIMETOLIVESECONDS = 300; @Override - public HazelcastStateStore bootstrap(Collection seedIps, Map config) + public HazelcastStateStore bootstrap(Collection locations, Map config) { // Initialize the Hazelcast instance and discovery service final String discoveryMode = config.get(DISCOVERY_MODE_CONFIG_NAME); @@ -85,6 +86,9 @@ public class HazelcastStateStoreBootstrapper // Set eviction rules hzConfig = setEvictionConfigs(hzConfig, MERGEMAP); + // Set discovery port + hzConfig = setPortConfigs(config, hzConfig); + // default discovery_mode = multicast if (discoveryMode == null || discoveryMode.equals(DISCOVERY_MODE_MULTICAST)) { hzConfig.setProperty(DISCOVERY_ENABLED, "true"); @@ -97,25 +101,18 @@ public class HazelcastStateStoreBootstrapper hzInstance = Hazelcast.newHazelcastInstance(hzConfig); } else if (discoveryMode.equals(DISCOVERY_MODE_TCPIP)) { - if (seedIps == null || seedIps.isEmpty()) { + if (locations == null || locations.isEmpty()) { throw new PrestoException(STATE_STORE_FAILURE, "Using TCP-IP discovery but no seed ip found." + "Please check whether seed store is enabled"); } // Hardcode seed IP for testing NetworkConfig network = hzConfig.getNetworkConfig(); - // Disable port autoincrement if port is specified - String port = config.get(PORT_CONFIG_NAME); - if (port != null) { - network.setPortAutoIncrement(false); - network.setPort(Integer.parseInt(port)); - } - JoinConfig join = network.getJoin(); join.getAwsConfig().setEnabled(false); join.getMulticastConfig().setEnabled(false); join.getTcpIpConfig().setEnabled(true) - .addMember(String.join(",", seedIps)); + .addMember(String.join(",", locations)); hzInstance = Hazelcast.newHazelcastInstance(hzConfig); } @@ -148,6 +145,21 @@ public class HazelcastStateStoreBootstrapper return hzConfig; } + private Config setPortConfigs(Map properties, Config config) + { + String port = properties.get(DISCOVERY_PORT_CONFIG_NAME); + + if (port == null || port.trim().isEmpty()) { + port = DEFAULT_DISCOVERY_PORT; + } + + // Disable port auto increment + config.getNetworkConfig().setPortAutoIncrement(false); + config.getNetworkConfig().setPort(Integer.parseInt(port)); + + return config; + } + private Config setCpSystemConfigs(Map properties, Config config) { String cpMemberCountValue = properties.get("hazelcast.cp-system.member-count"); diff --git a/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterLifecycleListener.java b/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterLifecycleListener.java index 1cfe3101f..f8251db5a 100644 --- a/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterLifecycleListener.java +++ b/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterLifecycleListener.java @@ -30,6 +30,7 @@ import java.util.Map; import java.util.Set; import java.util.UUID; +import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_PORT_CONFIG_NAME; import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertTrue; @@ -73,7 +74,7 @@ public class TestHazelcastClusterLifecycleListener Map config = new HashMap<>(0); config.put("hazelcast.discovery.mode", "tcp-ip"); config.put("state-store.cluster", TEST_CLUSTER_NAME); - config.put("hazelcast.tcp-ip.port", PORT1); + config.put(DISCOVERY_PORT_CONFIG_NAME, PORT1); MockSeedStore mockSeedStore = new MockSeedStore(); mockSeedStore.add(seeds); @@ -88,7 +89,7 @@ public class TestHazelcastClusterLifecycleListener Map config = new HashMap<>(0); config.put("hazelcast.discovery.mode", "tcp-ip"); config.put("state-store.cluster", TEST_CLUSTER_NAME); - config.put("hazelcast.tcp-ip.port", port); + config.put(DISCOVERY_PORT_CONFIG_NAME, port); StateStoreBootstrapper bootstrapper = new HazelcastStateStoreBootstrapper(); return bootstrapper.bootstrap(ImmutableSet.of(MEMBER_1_ADDRESS, MEMBER_2_ADDRESS), config); diff --git a/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterMembershipListener.java b/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterMembershipListener.java index a73f50d62..3f64ac550 100644 --- a/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterMembershipListener.java +++ b/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterMembershipListener.java @@ -24,6 +24,7 @@ import java.util.Map; import java.util.UUID; import java.util.concurrent.TimeUnit; +import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_PORT_CONFIG_NAME; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertTrue; @@ -56,7 +57,7 @@ public class TestHazelcastClusterMembershipListener final long timeout3000 = 3000L; StateStore member1 = setupHazelcastInstance(PORT1); member1.registerNodeFailureHandler(node -> { - assertEquals((String) node, LOCALHOST); + assertEquals((String) node, MEMBER_2_ADDRESS); isNotified = true; }); StateStore member2 = setupHazelcastInstance(PORT2); @@ -78,7 +79,7 @@ public class TestHazelcastClusterMembershipListener Map config = new HashMap<>(0); config.put("hazelcast.discovery.mode", "tcp-ip"); config.put("state-store.cluster", TEST_CLUSTER_NAME); - config.put("hazelcast.tcp-ip.port", port); + config.put(DISCOVERY_PORT_CONFIG_NAME, port); StateStoreBootstrapper bootstrapper = new HazelcastStateStoreBootstrapper(); return bootstrapper.bootstrap(ImmutableSet.of(MEMBER_1_ADDRESS, MEMBER_2_ADDRESS), config); diff --git a/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateStoreBootstrapper.java b/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateStoreBootstrapper.java index fd119f390..7d2a6ce72 100644 --- a/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateStoreBootstrapper.java +++ b/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateStoreBootstrapper.java @@ -24,6 +24,7 @@ import java.util.Map; import java.util.UUID; import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_MODE_CONFIG_NAME; +import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_PORT_CONFIG_NAME; import static org.testng.Assert.assertTrue; /** @@ -35,6 +36,7 @@ public class TestHazelcastStateStoreBootstrapper { private static final String LOCALHOST = "127.0.0.1"; private static final String PORT = "5707"; + private static final String PORT2 = "5715"; /** * Test Bootstrap @@ -46,7 +48,7 @@ public class TestHazelcastStateStoreBootstrapper config.put(DISCOVERY_MODE_CONFIG_NAME, "tcp-ip"); config.put("state-store.cluster", "cluster-" + UUID.randomUUID()); config.put("hazelcast.cp-system.member-count", "3"); - config.put("hazelcast.tcp-ip.port", PORT); + config.put(DISCOVERY_PORT_CONFIG_NAME, PORT); StateStoreBootstrapper bootstrapper = new HazelcastStateStoreBootstrapper(); StateStore stateStore = bootstrapper.bootstrap(ImmutableSet.of(LOCALHOST), config); @@ -63,6 +65,7 @@ public class TestHazelcastStateStoreBootstrapper Map config = new HashMap<>(0); config.put(DISCOVERY_MODE_CONFIG_NAME, "multicast"); config.put("state-store.cluster", "cluster"); + config.put(DISCOVERY_PORT_CONFIG_NAME, PORT2); StateStoreBootstrapper bootstrapper = new HazelcastStateStoreBootstrapper(); StateStore stateStore = bootstrapper.bootstrap(ImmutableSet.of(LOCALHOST), config); diff --git a/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateStoreFactory.java b/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateStoreFactory.java index 2b59f1ae4..173f43593 100644 --- a/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateStoreFactory.java +++ b/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateStoreFactory.java @@ -35,7 +35,7 @@ import java.util.UUID; import static io.hetu.core.statestore.Constants.STATE_STORE_CLUSTER_CONFIG_NAME; import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_MODE_CONFIG_NAME; import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_MODE_TCPIP; -import static io.hetu.core.statestore.hazelcast.HazelcastConstants.PORT_CONFIG_NAME; +import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_PORT_CONFIG_NAME; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; @@ -130,7 +130,7 @@ public class TestHazelcastStateStoreFactory Map config = new HashMap<>(0); config.put(DISCOVERY_MODE_CONFIG_NAME, DISCOVERY_MODE_TCPIP); config.put(STATE_STORE_CLUSTER_CONFIG_NAME, TEST_CLUSTER_NAME); - config.put(PORT_CONFIG_NAME, PORT); + config.put(DISCOVERY_PORT_CONFIG_NAME, PORT); StateStoreBootstrapper bootstrapper = new HazelcastStateStoreBootstrapper(); bootstrapper.bootstrap(ImmutableSet.of(MEMBER_ADDRESS), config); diff --git a/presto-main/src/main/java/io/prestosql/seedstore/SeedStoreManager.java b/presto-main/src/main/java/io/prestosql/seedstore/SeedStoreManager.java index 1c708be17..47b3cbbde 100644 --- a/presto-main/src/main/java/io/prestosql/seedstore/SeedStoreManager.java +++ b/presto-main/src/main/java/io/prestosql/seedstore/SeedStoreManager.java @@ -33,6 +33,7 @@ import java.util.Collection; import java.util.HashMap; import java.util.HashSet; import java.util.Map; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -40,7 +41,7 @@ import java.util.stream.Collectors; import static com.google.common.base.Preconditions.checkState; import static io.airlift.concurrent.Threads.daemonThreadsNamed; -import static io.prestosql.spi.StandardErrorCode.STATE_STORE_FAILURE; +import static io.prestosql.spi.StandardErrorCode.SEED_STORE_FAILURE; import static io.prestosql.util.PropertiesUtil.loadProperties; import static java.lang.String.format; import static java.util.Objects.requireNonNull; @@ -63,19 +64,20 @@ public class SeedStoreManager // properties default value 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 = "30000"; // 30 seconds + private static final String 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 seedStoreFactories = new ConcurrentHashMap<>(); private final FileSystemClientManager fileSystemClientManager; - private ScheduledExecutorService seedRefreshExecutor; + private ScheduledExecutorService seedRefreshExecutor = newSingleThreadScheduledExecutor(daemonThreadsNamed("SeedRefresher")); private SeedStore seedStore; private String seedStoreType; private String filesystemProfile; private long seedHeartBeat; private long seedHeartBeatTimeout; + private ConcurrentHashMap refreshableSeedsMap = new ConcurrentHashMap<>(); @Inject public SeedStoreManager(FileSystemClientManager fileSystemClientManager) @@ -115,39 +117,59 @@ public class SeedStoreManager fileSystemClientManager.getFileSystemClient(filesystemProfile), ImmutableMap.copyOf(config)); } - + try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(seedStore.getClass().getClassLoader())) { + // start seed refresher + seedRefreshExecutor.scheduleWithFixedDelay(() -> refreshSeeds(), 0, seedHeartBeat, TimeUnit.MILLISECONDS); + } LOG.info("-- Loaded seed store %s --", seedStoreType); } } /** - * add seed to seed store + * Get all seeds from seed store * - * @param seedLocation - * @return a collection of current seeds in the seed store - * @throws Exception + * @return a collection of seeds in the seed store + * @throws IOException */ - public Collection addSeedToSeedStore(String seedLocation) - throws Exception + public Collection getAllSeeds() + throws IOException + { + if (seedStore == null) { + throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null"); + } + + Collection seeds = seedStore.get(); + + return seeds; + } + + /** + * Add seed to seed store. If refreshable is enabled, seed will be refreshed periodically + * + * @param refreshable + * @return a collection of seeds in the seed store + * @throws IOException + */ + + public Collection addSeed(String seedLocation, boolean refreshable) + throws IOException { Collection seeds = new HashSet<>(); if (seedStore == null) { - throw new PrestoException(STATE_STORE_FAILURE, "Seed store is null"); + throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null"); } - if (seedStoreType.equalsIgnoreCase(SEED_STORE_TYPE_DEFAULT_VALUE)) { - try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(seedStore.getClass().getClassLoader())) { - // Clear expired seeds in the seed file - clearExpiredSeed(); - // Create a seed and add seed to seed store - Seed seed = seedStore.create(ImmutableMap.of( - Seed.LOCATION_PROPERTY_NAME, seedLocation, - Seed.TIMESTAMP_PROPERTY_NAME, String.valueOf(System.currentTimeMillis()))); - seeds = addToSeedStore(seed); - //start seed refresher - startSeedRefresh(seed); - } + + Seed seed = seedStore.create(ImmutableMap.of( + Seed.LOCATION_PROPERTY_NAME, seedLocation, + Seed.TIMESTAMP_PROPERTY_NAME, String.valueOf(System.currentTimeMillis()))); + seeds = addSeed(seed); + + if (refreshable) { + refreshableSeedsMap.put(seedLocation, seed); } + + LOG.debug("Seed=%s added to seed store", seedLocation); return seeds; } @@ -155,37 +177,65 @@ public class SeedStoreManager * remove seed from seed store * * @param seedLocation - * @return a collection of current seeds in the seed store - * @throws Exception + * @return a collection of seeds in the seed store + * @throws IOException */ - public Collection removeSeedFromSeedStore(String seedLocation) - throws Exception + public Collection removeSeed(String seedLocation) + throws IOException { Collection seeds = new HashSet<>(); if (seedStore == null) { - throw new PrestoException(STATE_STORE_FAILURE, "Seed store is null"); + throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null"); } - if (seedStoreType.equalsIgnoreCase(SEED_STORE_TYPE_DEFAULT_VALUE)) { - try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(seedStore.getClass().getClassLoader())) { - Seed seed = seedStore.get().stream().filter(s -> s.getLocation().equals(seedLocation)).findFirst().get(); + refreshableSeedsMap.remove(seedLocation); + Optional seedOptional = seedStore.get().stream().filter(s -> s.getLocation().equals(seedLocation)).findFirst(); - seeds.remove(Lists.newArrayList(seed)); - } + if (seedOptional.isPresent()) { + seeds = seedStore.remove(Lists.newArrayList(seedOptional.get())); + LOG.debug("Seed=%s removed from seed store", seedLocation); } + return seeds; } /** - * Add seed to seed store + * Clear expired seed in the seed store * - * @return updated list of seeds - * @throws Exception + * @throws IOException */ - private Collection addToSeedStore(Seed seed) - throws Exception + public void clearExpiredSeeds() + throws IOException + { + if (seedStore == null) { + throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null"); + } + + long retryInterval = 0L; + for (int retryTimes = 0; retryTimes <= SEED_RETRY_TIMES; retryTimes++) { + try { + TimeUnit.MILLISECONDS.sleep(retryInterval); + Collection expiredSeeds = seedStore.get() + .stream() + .filter(s -> (System.currentTimeMillis() - s.getTimestamp() > seedHeartBeatTimeout)) + .collect(Collectors.toList()); + if (expiredSeeds.size() > 0) { + LOG.debug("Expired seeds=%s will be cleared", expiredSeeds); + seedStore.remove(expiredSeeds); + } + break; + } + catch (InterruptedException | RuntimeException e) { + LOG.warn("clearExpiredSeed failed: %s, will retry at times: %s", e.getMessage(), ++retryTimes); + retryInterval += SEED_RETRY_INTERVAL; + } + } + } + + private Collection addSeed(Seed seed) + throws IOException { int retryTimes = 0; long retryInterval = 0L; @@ -197,7 +247,7 @@ public class SeedStoreManager seeds = seedStore.add(Lists.newArrayList(seed)); } catch (InterruptedException | RuntimeException e) { - LOG.warn("addSeedToSeedStore failed: %s, will retry at times: %s", e.getMessage(), retryTimes); + LOG.warn("add seed=%s failed: %s, will retry at times: %s", seed, e.getMessage(), retryTimes); } finally { retryTimes++; @@ -207,7 +257,8 @@ public class SeedStoreManager while (retryTimes <= SEED_RETRY_TIMES && (seeds == null || seeds.size() == 0)); if (seeds == null || seeds.size() == 0) { - throw new PrestoException(STATE_STORE_FAILURE, "addSeedToSeedStore failed after retry:" + SEED_RETRY_TIMES); + throw new PrestoException(SEED_STORE_FAILURE, String.format("add seed=%s to seed store failed after retry:%d", + seed.getLocation(), SEED_RETRY_TIMES)); } return seeds; @@ -234,54 +285,22 @@ public class SeedStoreManager return seedStore; } - /** - * Start background task to keep refreshing seed's timestamp to SeedStore - * - * @param seed Seed node to refresh - */ - private void startSeedRefresh(Seed seed) + private void refreshSeeds() { - if (seedRefreshExecutor == null) { - seedRefreshExecutor = newSingleThreadScheduledExecutor(daemonThreadsNamed("SeedRefresher")); - seedRefreshExecutor.scheduleWithFixedDelay(() -> refreshSeed(seed), 0, seedHeartBeat, TimeUnit.MILLISECONDS); - } - else { - LOG.debug("Seed Refresh has been started, ignored"); - } - } - - private void refreshSeed(Seed seed) - { - try { + for (Map.Entry entry : refreshableSeedsMap.entrySet()) { + long newTime = System.currentTimeMillis(); + LOG.debug("seed=%s refresh with oldTimestamp=%s and newTimestamp=%s", entry.getKey(), entry.getValue().getTimestamp(), newTime); Seed newSeed = seedStore.create(ImmutableMap.of( - Seed.LOCATION_PROPERTY_NAME, seed.getLocation(), - Seed.TIMESTAMP_PROPERTY_NAME, String.valueOf(System.currentTimeMillis()))); - - seedStore.add(Lists.newArrayList(newSeed)); - } - catch (Exception e) { - LOG.debug("Error refreshSeed: %s, will refresh in next %s milliseconds" + e.getMessage(), seedHeartBeat); - } - } - - private void clearExpiredSeed() - throws IOException - { - long retryInterval = 0L; - for (int retryTimes = 0; retryTimes <= SEED_RETRY_TIMES; retryTimes++) { + Seed.LOCATION_PROPERTY_NAME, entry.getKey(), + Seed.TIMESTAMP_PROPERTY_NAME, String.valueOf(newTime))); try { - TimeUnit.MILLISECONDS.sleep(retryInterval); - Collection expiredSeeds = seedStore.get().stream() - .filter(s -> (System.currentTimeMillis() - s.getTimestamp() > seedHeartBeatTimeout)) - .collect(Collectors.toList()); - if (expiredSeeds.size() > 0) { - seedStore.remove(expiredSeeds); - } - break; + seedStore.add(Lists.newArrayList(newSeed)); + entry.setValue(newSeed); } - catch (InterruptedException | RuntimeException e) { - LOG.debug("clearExpiredSeed failed: %s, will retry at times: %s", e.getMessage(), ++retryTimes); - retryInterval += SEED_RETRY_INTERVAL; + catch (IOException | RuntimeException e) { + LOG.warn("Error refresh seed=%s with error message: %s, will refresh in next %s milliseconds", + entry.getKey(), e.getMessage(), seedHeartBeat); + continue; } } } diff --git a/presto-main/src/main/java/io/prestosql/statestore/EmbeddedStateStoreLauncher.java b/presto-main/src/main/java/io/prestosql/statestore/EmbeddedStateStoreLauncher.java index c0fec8e1d..4bd0701c1 100644 --- a/presto-main/src/main/java/io/prestosql/statestore/EmbeddedStateStoreLauncher.java +++ b/presto-main/src/main/java/io/prestosql/statestore/EmbeddedStateStoreLauncher.java @@ -40,8 +40,12 @@ import java.util.concurrent.locks.Lock; import java.util.stream.Collectors; import static io.prestosql.spi.StandardErrorCode.STATE_STORE_FAILURE; +import static io.prestosql.statestore.StateStoreConstants.DEFAULT_HAZELCAST_DISCOVERY_PORT; +import static io.prestosql.statestore.StateStoreConstants.HAZELCAST; +import static io.prestosql.statestore.StateStoreConstants.HAZELCAST_DISCOVERY_PORT_PROPERTY_NAME; import static io.prestosql.statestore.StateStoreConstants.STATE_STORE_CLUSTER_PROPERTY_NAME; import static io.prestosql.statestore.StateStoreConstants.STATE_STORE_CONFIGURATION_PATH; +import static io.prestosql.statestore.StateStoreConstants.STATE_STORE_TYPE_PROPERTY_NAME; import static io.prestosql.util.PropertiesUtil.loadProperties; import static java.util.Objects.requireNonNull; @@ -98,16 +102,27 @@ public class EmbeddedStateStoreLauncher if (STATE_STORE_LAUNCHER_CONFIGURATION.exists()) { Map properties = new HashMap<>(loadProperties(STATE_STORE_LAUNCHER_CONFIGURATION)); - Set ips = new HashSet<>(); - if (seedStoreManager.getSeedStore() != null) { // Set seed store name seedStoreManager.getSeedStore().setName(properties.get(STATE_STORE_CLUSTER_PROPERTY_NAME)); - // Add seed to seed store - ips = seedStoreManager.addSeedToSeedStore(getNodeUri().getHost()).stream().map(x -> x.getLocation()).collect(Collectors.toSet()); + // Clear expired seeds + seedStoreManager.clearExpiredSeeds(); + // Get all seeds + Set locations = seedStoreManager.getAllSeeds().stream().map(x -> x.getLocation()).collect(Collectors.toSet()); + String launcherPort = getStateStoreLauncherPort(properties); + requireNonNull(launcherPort, "The launcher port is null"); + // 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); + } + } + else { + launchStateStore(new HashSet<>(), properties); } - launchStateStore(ips, properties); if (stateStore == null) { throw new PrestoException(STATE_STORE_FAILURE, "Unable to launch state store, please check your configuration"); } @@ -229,22 +244,20 @@ public class EmbeddedStateStoreLauncher return registered; } - private void handleNodeFailure(Object failureNodeHost) + private void handleNodeFailure(Object failureMember) { - // Only retry if the failed node is the current discovery node - // and when there are multiple coordinators if (hetuConfig.isMultipleCoordinatorEnabled()) { - StateMap discoveryServiceMap = - (StateMap) stateStore.getStateCollection(DISCOVERY_SERVICE); - registerDiscoveryService((String) failureNodeHost); + // failureMember format host:port + String failureMemberHost = ((String) failureMember).split(":")[0]; + registerDiscoveryService(failureMemberHost); } if (seedStoreManager.getSeedStore() != null) { try { - seedStoreManager.removeSeedFromSeedStore((String) failureNodeHost); + seedStoreManager.removeSeed((String) failureMember); } catch (Exception e) { - LOG.error("Cannot remove failure node %s from seed store: %s", failureNodeHost, e.getMessage()); + LOG.error("Cannot remove failure node %s from seed store: %s", failureMember, e.getMessage()); } } } @@ -258,4 +271,19 @@ public class EmbeddedStateStoreLauncher return httpServerInfo.getHttpUri(); } } + + private String getStateStoreLauncherPort(Map properties) + { + String port = null; + String stateStoreType = properties.get(STATE_STORE_TYPE_PROPERTY_NAME); + if (stateStoreType != null) { + if (stateStoreType.trim().equals(HAZELCAST)) { + port = properties.get(HAZELCAST_DISCOVERY_PORT_PROPERTY_NAME); + if (port == null || port.trim().isEmpty()) { + port = DEFAULT_HAZELCAST_DISCOVERY_PORT; + } + } + } + return port; + } } diff --git a/presto-main/src/main/java/io/prestosql/statestore/StateStoreConstants.java b/presto-main/src/main/java/io/prestosql/statestore/StateStoreConstants.java index febb205ad..38aade19c 100644 --- a/presto-main/src/main/java/io/prestosql/statestore/StateStoreConstants.java +++ b/presto-main/src/main/java/io/prestosql/statestore/StateStoreConstants.java @@ -105,6 +105,21 @@ public class StateStoreConstants */ public static final long DEFAULT_ACQUIRED_LOCK_TIME_MS = 1000L; + /** + * Hazelcast + */ + public static final String HAZELCAST = "hazelcast"; + + /** + * Hazelcast discovery port property name + */ + public static final String HAZELCAST_DISCOVERY_PORT_PROPERTY_NAME = "hazelcast.discovery.port"; + + /** + * Hazelcast default discovery port + */ + public static final String DEFAULT_HAZELCAST_DISCOVERY_PORT = "5701"; + /** * Hazelcast discovery mode property name */ diff --git a/presto-main/src/test/java/io/prestosql/operator/TestDynamicFilterSourceOperator.java b/presto-main/src/test/java/io/prestosql/operator/TestDynamicFilterSourceOperator.java index 2b74f9576..d74d0b500 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestDynamicFilterSourceOperator.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestDynamicFilterSourceOperator.java @@ -60,6 +60,7 @@ import java.util.concurrent.ScheduledExecutorService; import static com.google.common.base.Strings.repeat; import static io.airlift.concurrent.Threads.daemonThreadsNamed; import static io.airlift.slice.Slices.utf8Slice; +import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_PORT_CONFIG_NAME; import static io.prestosql.SequencePageBuilder.createSequencePage; import static io.prestosql.SessionTestUtils.TEST_SESSION; import static io.prestosql.SystemSessionProperties.getDynamicFilteringMaxPerDriverSize; @@ -113,7 +114,7 @@ public class TestDynamicFilterSourceOperator "state-store.name=test\n" + "state-store.cluster=test-cluster\n" + "hazelcast.discovery.mode=tcp-ip\n" + - "hazelcast.tcp-ip.port=7980\n"); + "hazelcast.discovery.port=7980\n"); configWriter.close(); Set seeds = new HashSet<>(); SeedStore mockSeedStore = mock(SeedStore.class); @@ -150,7 +151,7 @@ public class TestDynamicFilterSourceOperator Map config = new HashMap<>(); config.put("hazelcast.discovery.mode", "tcp-ip"); config.put("state-store.cluster", "test-cluster"); - config.put("hazelcast.tcp-ip.port", port); + config.put(DISCOVERY_PORT_CONFIG_NAME, port); StateStoreBootstrapper bootstrapper = new HazelcastStateStoreBootstrapper(); return bootstrapper.bootstrap(ImmutableSet.of("127.0.0.1:" + port), config); diff --git a/presto-main/src/test/java/io/prestosql/seedstore/TestSeedStoreManager.java b/presto-main/src/test/java/io/prestosql/seedstore/TestSeedStoreManager.java index 1f38a5238..9f9d5991e 100644 --- a/presto-main/src/test/java/io/prestosql/seedstore/TestSeedStoreManager.java +++ b/presto-main/src/test/java/io/prestosql/seedstore/TestSeedStoreManager.java @@ -19,6 +19,7 @@ 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.testing.assertions.Assert; import org.testng.annotations.BeforeMethod; import org.testng.annotations.BeforeTest; import org.testng.annotations.Test; @@ -29,6 +30,7 @@ import java.io.IOException; import java.util.Collection; import java.util.HashSet; import java.util.Map; +import java.util.Objects; import java.util.Set; import static org.mockito.Matchers.any; @@ -60,9 +62,13 @@ public class TestSeedStoreManager seedStoreConfigFile.createNewFile(); } FileWriter configWritter = new FileWriter("etc/seed-store.properties"); - configWritter.write("seed-store.type=hdfs\n"); + configWritter.write("seed-store.type=filebased\n"); // Any profile, must be provided but will not be used - configWritter.write("seed-store.filesystem.profile=etc/filesystem/hdfs-config-default.properties"); + configWritter.write("seed-store.filesystem.profile=etc/filesystem/hdfs-config-default.properties\n"); + // Set heartbeat to 1 seconds + configWritter.write("seed-store.seed.heartbeat=1000\n"); + // Set heartbeat timeout to 3 seconds + configWritter.write("seed-store.seed.heartbeat.timeout=3000"); configWritter.close(); } @@ -75,12 +81,9 @@ public class TestSeedStoreManager seedStoreManager = new SeedStoreManager(mockFileSystemClientManager); SeedStore mockSeedStore = new MockSeedStore(); - HashSet seeds = new HashSet<>(); - seeds.add(new MockSeed("location1")); - seeds.add(new MockSeed("location2")); - mockSeedStore.add(seeds); + mockSeedStore.add(new HashSet<>()); SeedStoreFactory mockSeedStoreFactory = mock(SeedStoreFactory.class); - when(mockSeedStoreFactory.getName()).thenReturn("hdfs"); + when(mockSeedStoreFactory.getName()).thenReturn("filebased"); when(mockSeedStoreFactory.create(any(String.class), any(HetuFileSystemClient.class), any(Map.class))).thenReturn(mockSeedStore); @@ -99,23 +102,68 @@ public class TestSeedStoreManager public void testAddToSeedStore() throws Exception { + String location1 = "location1"; + String location2 = "location2"; seedStoreManager.loadSeedStore(); - seedStoreManager.addSeedToSeedStore("location3"); + seedStoreManager.addSeed(location1, false); + seedStoreManager.addSeed(location2, true); + Collection result = seedStoreManager.getAllSeeds(); + Assert.assertEquals(result.size(), 2); + Assert.assertTrue(result.stream().filter(s -> s.getLocation().equals(location1)).findAny().isPresent()); + Assert.assertTrue(result.stream().filter(s -> s.getLocation().equals(location2)).findAny().isPresent()); + + long timestamp1Old = result.stream().filter(s -> s.getLocation().equals(location1)).findAny().get().getTimestamp(); + 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(); + 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); + Assert.assertTrue(timestamp2Old < timestamp2New); } @Test void testRemoveFromSeedStore() throws Exception { + String location1 = "location1"; + String location2 = "location2"; + String location3 = "location3"; seedStoreManager.loadSeedStore(); - seedStoreManager.removeSeedFromSeedStore("location1"); + seedStoreManager.addSeed(location1, false); + seedStoreManager.addSeed(location2, true); + Assert.assertEquals(seedStoreManager.getAllSeeds().size(), 2); + seedStoreManager.removeSeed(location1); + seedStoreManager.removeSeed(location3); + Collection result = seedStoreManager.getAllSeeds(); + Assert.assertEquals(result.size(), 1); + Assert.assertFalse(result.stream().filter(s -> s.getLocation().equals(location1)).findAny().isPresent()); + } + + @Test + void testClearExpiredSeed() + throws Exception + { + String location1 = "location1"; + String location2 = "location2"; + seedStoreManager.loadSeedStore(); + seedStoreManager.addSeed(location1, false); + seedStoreManager.addSeed(location2, true); + Assert.assertEquals(seedStoreManager.getAllSeeds().size(), 2); + //wait 5 seconds, location1 expired since refreshable is not enabled + Thread.sleep(5000); + seedStoreManager.clearExpiredSeeds(); + Collection result = seedStoreManager.getAllSeeds(); + Assert.assertEquals(result.size(), 1); + Assert.assertFalse(result.stream().filter(s -> s.getLocation().equals(location1)).findAny().isPresent()); } @Test(expectedExceptions = IllegalArgumentException.class) void testDupAddFactory() { SeedStoreFactory mockSeedStoreFactory2 = mock(SeedStoreFactory.class); - when(mockSeedStoreFactory2.getName()).thenReturn("hdfs"); + when(mockSeedStoreFactory2.getName()).thenReturn("filebased"); seedStoreManager.addSeedStoreFactory(mockSeedStoreFactory2); } @@ -125,17 +173,17 @@ public class TestSeedStoreManager private static final long serialVersionUID = 4L; String location; - long timeStamp; + long timestamp; /** * Constructor for the mock seed * * @param location host location of this seed */ - public MockSeed(String location) + public MockSeed(String location, long timestamp) { this.location = location; - timeStamp = 0L; + this.timestamp = timestamp; } @Override @@ -147,7 +195,7 @@ public class TestSeedStoreManager @Override public long getTimestamp() { - return 0L; + return timestamp; } @Override @@ -157,16 +205,43 @@ public class TestSeedStoreManager return "MOCK SEED. SHOULD NOT SERIALIZE."; } - public void setTimeStamp(long timeStamp) + public void setTimestamp(long timestamp) { - this.timeStamp = timeStamp; + this.timestamp = timestamp; + } + + @Override + public boolean equals(Object obj) + { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + MockSeed mockSeed = (MockSeed) obj; + return location.equals(mockSeed.location); + } + + @Override + public int hashCode() + { + return Objects.hash(location); + } + + @Override + public String toString() + { + return "MockSeed{" + + "location='" + location + '\'' + + ", timestamp=" + timestamp + '}'; } } class MockSeedStore implements SeedStore { - private static final int INITIAL_SIZE = 2; + private static final int INITIAL_SIZE = 0; private Set seeds; /** @@ -180,6 +255,8 @@ public class TestSeedStoreManager @Override public Collection add(Collection seedsToAdd) { + // overwrite all seeds + this.seeds.removeAll(seedsToAdd); this.seeds.addAll(seedsToAdd); return this.seeds; } @@ -201,7 +278,8 @@ public class TestSeedStoreManager public Seed create(Map properties) { String location = properties.get(Seed.LOCATION_PROPERTY_NAME); - return new MockSeed(location); + long timestamp = Long.parseLong(properties.get(Seed.TIMESTAMP_PROPERTY_NAME)); + return new MockSeed(location, timestamp); } @Override diff --git a/presto-main/src/test/java/io/prestosql/statestore/TestStateStoreLauncherAndProvider.java b/presto-main/src/test/java/io/prestosql/statestore/TestStateStoreLauncherAndProvider.java index ed14a564f..219f2d269 100644 --- a/presto-main/src/test/java/io/prestosql/statestore/TestStateStoreLauncherAndProvider.java +++ b/presto-main/src/test/java/io/prestosql/statestore/TestStateStoreLauncherAndProvider.java @@ -45,6 +45,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_PORT_CONFIG_NAME; import static io.prestosql.statestore.StateStoreConstants.STATE_STORE_CONFIGURATION_PATH; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -85,7 +86,7 @@ public class TestStateStoreLauncherAndProvider "state-store.name=test\n" + "state-store.cluster=test-cluster\n" + "hazelcast.discovery.mode=tcp-ip\n" + - "hazelcast.tcp-ip.port=7980\n"); + "hazelcast.discovery.port=7980\n"); configWriter.close(); } @@ -128,7 +129,7 @@ public class TestStateStoreLauncherAndProvider SeedStoreManager mockSeedStoreManager = mock(SeedStoreManager.class); when(mockSeedStoreManager.getSeedStore()).thenReturn(mockSeedStore); - when(mockSeedStoreManager.addSeedToSeedStore("localhost")).thenReturn(seeds); + when(mockSeedStoreManager.addSeed("localhost", true)).thenReturn(seeds); InternalCommunicationConfig mockInternalCommunicationConfig = mock(InternalCommunicationConfig.class); HttpServerInfo mockHttpServerInfo = mock(HttpServerInfo.class); @@ -147,7 +148,7 @@ public class TestStateStoreLauncherAndProvider // mock "remove" second instance from cluster (delete from seed store) seeds.remove(mockSeed2); when(mockSeed1.getLocation()).thenReturn("127.0.0.1:7980"); - when(mockSeedStoreManager.addSeedToSeedStore("localhost")).thenReturn(seeds); + when(mockSeedStoreManager.addSeed("localhost", true)).thenReturn(seeds); ((HazelcastStateStore) second).shutdown(); // Allow the first node to handle failure @@ -161,7 +162,7 @@ public class TestStateStoreLauncherAndProvider Map config = new HashMap<>(); config.put("hazelcast.discovery.mode", "tcp-ip"); config.put("state-store.cluster", "test-cluster"); - config.put("hazelcast.tcp-ip.port", "7981"); + config.put(DISCOVERY_PORT_CONFIG_NAME, "7981"); StateStoreBootstrapper bootstrapper = new HazelcastStateStoreBootstrapper(); return bootstrapper.bootstrap(ImmutableSet.of("127.0.0.1:7980", "127.0.0.1:7981"), config); @@ -261,7 +262,7 @@ public class TestStateStoreLauncherAndProvider Map config = new HashMap<>(); config.put("hazelcast.discovery.mode", "tcp-ip"); config.put("state-store.cluster", "test-cluster"); - config.put("hazelcast.tcp-ip.port", port); + config.put(DISCOVERY_PORT_CONFIG_NAME, port); StateStoreBootstrapper bootstrapper = new HazelcastStateStoreBootstrapper(); return bootstrapper.bootstrap(ImmutableSet.of("127.0.0.1:" + port), config); diff --git a/presto-spi/src/main/java/io/prestosql/spi/StandardErrorCode.java b/presto-spi/src/main/java/io/prestosql/spi/StandardErrorCode.java index 7ff862595..b9544e984 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/StandardErrorCode.java +++ b/presto-spi/src/main/java/io/prestosql/spi/StandardErrorCode.java @@ -91,6 +91,8 @@ public enum StandardErrorCode CONFIGURATION_INVALID(0x0001_0017, INTERNAL_ERROR), CONFIGURATION_UNAVAILABLE(0x0001_0018, INTERNAL_ERROR), INVALID_RESOURCE_GROUP(0x0001_0019, INTERNAL_ERROR), + STATE_STORE_FAILURE(0x0001_001A, INTERNAL_ERROR), + SEED_STORE_FAILURE(0x0001_001B, INTERNAL_ERROR), GENERIC_INSUFFICIENT_RESOURCES(0x0002_0000, INSUFFICIENT_RESOURCES), EXCEEDED_GLOBAL_MEMORY_LIMIT(0x0002_0001, INSUFFICIENT_RESOURCES), @@ -100,9 +102,8 @@ public enum StandardErrorCode EXCEEDED_CPU_LIMIT(0x0002_0005, INSUFFICIENT_RESOURCES), EXCEEDED_SPILL_LIMIT(0x0002_0006, INSUFFICIENT_RESOURCES), EXCEEDED_LOCAL_MEMORY_LIMIT(0x0002_0007, INSUFFICIENT_RESOURCES), - ADMINISTRATIVELY_PREEMPTED(0x0002_0008, INSUFFICIENT_RESOURCES), + ADMINISTRATIVELY_PREEMPTED(0x0002_0008, INSUFFICIENT_RESOURCES) - STATE_STORE_FAILURE(0x0002_0009, INTERNAL_ERROR), /**/; // Connectors can use error codes starting at the range 0x0100_0000 diff --git a/presto-spi/src/main/java/io/prestosql/spi/statestore/StateStoreBootstrapper.java b/presto-spi/src/main/java/io/prestosql/spi/statestore/StateStoreBootstrapper.java index ab0fdfa7f..310eb681e 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/statestore/StateStoreBootstrapper.java +++ b/presto-spi/src/main/java/io/prestosql/spi/statestore/StateStoreBootstrapper.java @@ -27,9 +27,9 @@ public interface StateStoreBootstrapper /** * Bootstraps a state store and initializations * - * @param ips seed ips to bootstrap the state store + * @param locations locations a collection of host:port to bootstrap the state store * @param config the state store configs * @return bootstrapped StateStore instance */ - StateStore bootstrap(Collection ips, Map config); + StateStore bootstrap(Collection locations, Map config); } diff --git a/presto-tests/src/main/java/io/prestosql/tests/DistributedQueryRunnerWithStateStore.java b/presto-tests/src/main/java/io/prestosql/tests/DistributedQueryRunnerWithStateStore.java index 185af8d43..b999786a7 100644 --- a/presto-tests/src/main/java/io/prestosql/tests/DistributedQueryRunnerWithStateStore.java +++ b/presto-tests/src/main/java/io/prestosql/tests/DistributedQueryRunnerWithStateStore.java @@ -48,7 +48,7 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; -import static io.hetu.core.statestore.hazelcast.HazelcastConstants.PORT_CONFIG_NAME; +import static io.hetu.core.statestore.hazelcast.HazelcastConstants.DISCOVERY_PORT_CONFIG_NAME; import static java.lang.String.format; import static java.util.Objects.requireNonNull; @@ -109,7 +109,7 @@ public class DistributedQueryRunnerWithStateStore while (!availablePort(port)) { port = nextPort.getAndIncrement(); } - stateStoreProperties.put(PORT_CONFIG_NAME, port + ""); + stateStoreProperties.put(DISCOVERY_PORT_CONFIG_NAME, port + ""); stateStoreProperties.putIfAbsent(HazelcastConstants.DISCOVERY_MODE_CONFIG_NAME, HazelcastConstants.DISCOVERY_MODE_TCPIP); this.stateStores.add(((EmbeddedStateStoreLauncher) launcher).launchStateStore(ips, stateStoreProperties)); } @@ -161,7 +161,7 @@ public class DistributedQueryRunnerWithStateStore if (provider instanceof LocalStateStoreProvider) { Map stateStoreProperties = new HashMap<>(); stateStoreProperties.putIfAbsent(HazelcastConstants.DISCOVERY_MODE_CONFIG_NAME, HazelcastConstants.DISCOVERY_MODE_TCPIP); - stateStoreProperties.put(PORT_CONFIG_NAME, port + ""); + stateStoreProperties.put(DISCOVERY_PORT_CONFIG_NAME, port + ""); ((LocalStateStoreProvider) provider).setStateStore("hazelcast", stateStoreProperties); ((LocalStateStoreProvider) provider).createStateCollections(); }