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

Replace ip with ip:port in the seed 

Disable hazelcast port auto increment
Replace hazelcast config property hazelcast.tcp_ip.port with
hazelcast.discovery.port
Refactor SeedStoreManager and optimize its UT
This commit is contained in:
Jack Guo 2020-07-14 23:54:17 -04:00
parent d26b748c20
commit 49765d9fa9
16 changed files with 318 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastClusterMembershipListener.java
View File

@ -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());
}
}

9
hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastConstants.java
View File

@ -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

34
hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreBootstrapper.java
View File

@ -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<String> seedIps, Map<String, String> config)
public HazelcastStateStore bootstrap(Collection<String> locations, Map<String, String> 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<String, String> 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<String, String> properties, Config config)
{
String cpMemberCountValue = properties.get("hazelcast.cp-system.member-count");

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

@ -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<String, String> 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<String, String> 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);

5
hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterMembershipListener.java
View File

@ -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<String, String> 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);

5
hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateStoreBootstrapper.java
View File

@ -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<String, String> 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);

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

@ -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<String, String> 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);

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

@ -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<String, SeedStoreFactory> 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<String, Seed> 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<Seed> addSeedToSeedStore(String seedLocation)
throws Exception
public Collection<Seed> getAllSeeds()
throws IOException
{
if (seedStore == null) {
throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null");
}
Collection<Seed> 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<Seed> addSeed(String seedLocation, boolean refreshable)
throws IOException
{
Collection<Seed> 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<Seed> removeSeed(String seedLocation)
throws IOException
{
Collection<Seed> 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<Seed> 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<Seed> 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<Seed> 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<Seed> 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<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);
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<Seed> 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;
}
}
}

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

@ -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<String, String> properties = new HashMap<>(loadProperties(STATE_STORE_LAUNCHER_CONFIGURATION));
Set<String> 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<String> 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<String, String> discoveryServiceMap =
(StateMap<String, String>) 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<String, String> 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;
}
}

15
presto-main/src/main/java/io/prestosql/statestore/StateStoreConstants.java
View File

@ -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
*/

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

@ -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<Seed> seeds = new HashSet<>();
SeedStore mockSeedStore = mock(SeedStore.class);
@ -150,7 +151,7 @@ public class TestDynamicFilterSourceOperator
Map<String, String> 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);

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

@ -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<Seed> 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<Seed> 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<Seed> 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<Seed> 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<Seed> seeds;
/**
@ -180,6 +255,8 @@ public class TestSeedStoreManager
@Override
public Collection<Seed> add(Collection<Seed> 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<String, String> 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

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

@ -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<String, String> 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<String, String> 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);

5
presto-spi/src/main/java/io/prestosql/spi/StandardErrorCode.java
View File

@ -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

4
presto-spi/src/main/java/io/prestosql/spi/statestore/StateStoreBootstrapper.java
View File

@ -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<String> ips, Map<String, String> config);
StateStore bootstrap(Collection<String> locations, Map<String, String> config);
}

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

@ -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<String, String> 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();
}