Initial Merge for periodic sync
This commit is contained in:
Chunxue Yang
commit
6c4297f5b3
7 changed files with 562 additions and 12 deletions
|
|
@ -0,0 +1,152 @@
|
|||
/*
|
||||
* Copyright 2020 Amazon.com, Inc. or its affiliates.
|
||||
* 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 software.amazon.kinesis.coordinator;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import software.amazon.awssdk.utils.CollectionUtils;
|
||||
import software.amazon.kinesis.leases.Lease;
|
||||
import software.amazon.kinesis.leases.LeaseRefresher;
|
||||
import software.amazon.kinesis.leases.exceptions.DependencyException;
|
||||
import software.amazon.kinesis.leases.exceptions.InvalidStateException;
|
||||
import software.amazon.kinesis.leases.exceptions.ProvisionedThroughputException;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.Collections;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Random;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.locks.ReadWriteLock;
|
||||
import java.util.concurrent.locks.ReentrantReadWriteLock;
|
||||
import java.util.function.BooleanSupplier;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
* An implementation of the {@code LeaderDecider} to elect leader(s) based on workerId.
|
||||
* Leases are shuffled using a predetermined constant seed so that lease ordering is
|
||||
* preserved across workers.
|
||||
* This reduces the probability of choosing the leader workers co-located on the same
|
||||
* host in case workerId starts with a common string (e.g. IP Address).
|
||||
* Hence if a host has 3 workers, IPADDRESS_Worker1, IPADDRESS_Worker2, and IPADDRESS_Worker3,
|
||||
* we don't end up choosing all 3 for shard sync as a result of natural ordering of Strings.
|
||||
* This ensures redundancy for shard-sync during host failures.
|
||||
*/
|
||||
@Slf4j
|
||||
class DeterministicShuffleShardSyncLeaderDecider
|
||||
implements LeaderDecider {
|
||||
// Fixed seed so that the shuffle order is preserved across workers
|
||||
static final int DETERMINISTIC_SHUFFLE_SEED = 1947;
|
||||
|
||||
private static final long ELECTION_INITIAL_DELAY_MILLIS = 60 * 1000;
|
||||
private static final long ELECTION_SCHEDULING_INTERVAL_MILLIS = 5 * 60 * 1000;
|
||||
private static final int AWAIT_TERMINATION_MILLIS = 5000;
|
||||
|
||||
private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
|
||||
|
||||
private final LeaseRefresher leaseRefresher;
|
||||
private final int numPeriodicShardSyncWorkers;
|
||||
private final ScheduledExecutorService leaderElectionThreadPool;
|
||||
|
||||
private volatile Set<String> leaders;
|
||||
|
||||
/**
|
||||
* @param leaseRefresher LeaseManager instance used to fetch leases.
|
||||
* @param leaderElectionThreadPool Thread-pool to be used for leaderElection.
|
||||
* @param numPeriodicShardSyncWorkers Number of leaders that will be elected to perform periodic shard syncs.
|
||||
*/
|
||||
DeterministicShuffleShardSyncLeaderDecider(LeaseRefresher leaseRefresher, ScheduledExecutorService leaderElectionThreadPool,
|
||||
int numPeriodicShardSyncWorkers) {
|
||||
this.leaseRefresher = leaseRefresher;
|
||||
this.leaderElectionThreadPool = leaderElectionThreadPool;
|
||||
this.numPeriodicShardSyncWorkers = numPeriodicShardSyncWorkers;
|
||||
}
|
||||
|
||||
/*
|
||||
* Shuffles the leases deterministically and elects numPeriodicShardSyncWorkers number of workers
|
||||
* as leaders (workers that will perform shard sync).
|
||||
*/
|
||||
private void electLeaders() {
|
||||
try {
|
||||
log.debug("Started leader election at: " + Instant.now());
|
||||
List<Lease> leases = leaseRefresher.listLeases();
|
||||
List<String> uniqueHosts = leases.stream().map(Lease::leaseOwner)
|
||||
.filter(owner -> owner != null).distinct().sorted().collect(Collectors.toList());
|
||||
|
||||
Collections.shuffle(uniqueHosts, new Random(DETERMINISTIC_SHUFFLE_SEED));
|
||||
int numShardSyncWorkers = Math.min(uniqueHosts.size(), numPeriodicShardSyncWorkers);
|
||||
// In case value is currently being read, we wait for reading to complete before updating the variable.
|
||||
// This is to prevent any ConcurrentModificationException exceptions.
|
||||
readWriteLock.writeLock().lock();
|
||||
leaders = new HashSet<>(uniqueHosts.subList(0, numShardSyncWorkers));
|
||||
log.info("Elected leaders: " + String.join(", ", leaders));
|
||||
log.debug("Completed leader election at: " + Instant.now());
|
||||
} catch (DependencyException | InvalidStateException | ProvisionedThroughputException e) {
|
||||
log.error("Exception occurred while trying to fetch all leases for leader election", e);
|
||||
} catch (Throwable t) {
|
||||
log.error("Unknown exception during leader election.", t);
|
||||
} finally {
|
||||
readWriteLock.writeLock().unlock();
|
||||
}
|
||||
}
|
||||
|
||||
private boolean isWorkerLeaderForShardSync(String workerId) {
|
||||
return CollectionUtils.isNullOrEmpty(leaders) || leaders.contains(workerId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized Boolean isLeader(String workerId) {
|
||||
// if no leaders yet, synchronously get leaders. This will happen at first Shard Sync.
|
||||
if (executeConditionCheckWithReadLock(() -> CollectionUtils.isNullOrEmpty(leaders))) {
|
||||
electLeaders();
|
||||
// start a scheduled executor that will periodically update leaders.
|
||||
// The first run will be after a minute.
|
||||
// We don't need jitter since it is scheduled with a fixed delay and time taken to scan leases
|
||||
// will be different at different times and on different hosts/workers.
|
||||
leaderElectionThreadPool.scheduleWithFixedDelay(this::electLeaders, ELECTION_INITIAL_DELAY_MILLIS,
|
||||
ELECTION_SCHEDULING_INTERVAL_MILLIS, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
return executeConditionCheckWithReadLock(() -> isWorkerLeaderForShardSync(workerId));
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void shutdown() {
|
||||
try {
|
||||
leaderElectionThreadPool.shutdown();
|
||||
if (leaderElectionThreadPool.awaitTermination(AWAIT_TERMINATION_MILLIS, TimeUnit.MILLISECONDS)) {
|
||||
log.info("Successfully stopped leader election on the worker");
|
||||
} else {
|
||||
leaderElectionThreadPool.shutdownNow();
|
||||
log.info(String.format("Stopped leader election thread after awaiting termination for %d milliseconds",
|
||||
AWAIT_TERMINATION_MILLIS));
|
||||
}
|
||||
|
||||
} catch (InterruptedException e) {
|
||||
log.debug("Encountered InterruptedException while awaiting leader election threadPool termination");
|
||||
}
|
||||
}
|
||||
|
||||
// Execute condition checks using shared variables under a read-write lock.
|
||||
private boolean executeConditionCheckWithReadLock(BooleanSupplier action) {
|
||||
try {
|
||||
readWriteLock.readLock().lock();
|
||||
return action.getAsBoolean();
|
||||
} finally {
|
||||
readWriteLock.readLock().unlock();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,39 @@
|
|||
/*
|
||||
* Copyright 2020 Amazon.com, Inc. or its affiliates.
|
||||
* 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 software.amazon.kinesis.coordinator;
|
||||
|
||||
/**
|
||||
* Used in conjunction with periodic shard sync.
|
||||
* Implement this interface to allow KCL to decide if the current worker should execute shard sync.
|
||||
* When periodic shard sync is enabled, PeriodicShardSyncManager periodically checks if the current
|
||||
* worker is one of the leaders designated to execute shard-sync and then acts accordingly.
|
||||
*/
|
||||
public interface LeaderDecider {
|
||||
|
||||
/**
|
||||
* Method invoked to check the given workerId corresponds to one of the workers
|
||||
* designated to execute shard-syncs periodically.
|
||||
*
|
||||
* @param workerId ID of the worker
|
||||
* @return True if the worker with ID workerId can execute shard-sync. False otherwise.
|
||||
*/
|
||||
Boolean isLeader(String workerId);
|
||||
|
||||
/**
|
||||
* Can be invoked, if needed, to shutdown any clients/thread-pools
|
||||
* being used in the LeaderDecider implementation.
|
||||
*/
|
||||
void shutdown();
|
||||
}
|
||||
|
|
@ -0,0 +1,128 @@
|
|||
/*
|
||||
* Copyright 2019 Amazon.com, Inc. or its affiliates.
|
||||
* 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 software.amazon.kinesis.coordinator;
|
||||
|
||||
import lombok.EqualsAndHashCode;
|
||||
import lombok.Getter;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.lang3.Validate;
|
||||
import software.amazon.kinesis.exceptions.internal.KinesisClientLibIOException;
|
||||
import software.amazon.kinesis.leases.ShardSyncTask;
|
||||
import software.amazon.kinesis.lifecycle.ConsumerTask;
|
||||
import software.amazon.kinesis.lifecycle.TaskResult;
|
||||
import software.amazon.kinesis.metrics.MetricsCollectingTaskDecorator;
|
||||
import software.amazon.kinesis.metrics.MetricsFactory;
|
||||
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* The top level orchestrator for coordinating the periodic shard sync related
|
||||
* activities.
|
||||
*/
|
||||
@Getter
|
||||
@EqualsAndHashCode
|
||||
@Slf4j
|
||||
class PeriodicShardSyncManager {
|
||||
private static final long INITIAL_DELAY = 60 * 1000L;
|
||||
private static final long PERIODIC_SHARD_SYNC_INTERVAL_MILLIS = 5 * 60 * 1000L;
|
||||
|
||||
private final String workerId;
|
||||
private final LeaderDecider leaderDecider;
|
||||
private final ConsumerTask metricsEmittingShardSyncTask;
|
||||
private final ScheduledExecutorService shardSyncThreadPool;
|
||||
private boolean isRunning;
|
||||
|
||||
PeriodicShardSyncManager(String workerId, LeaderDecider leaderDecider, ShardSyncTask shardSyncTask, MetricsFactory metricsFactory) {
|
||||
this(workerId, leaderDecider, shardSyncTask, Executors.newSingleThreadScheduledExecutor(), metricsFactory);
|
||||
}
|
||||
|
||||
PeriodicShardSyncManager(String workerId, LeaderDecider leaderDecider, ShardSyncTask shardSyncTask, ScheduledExecutorService shardSyncThreadPool, MetricsFactory metricsFactory) {
|
||||
Validate.notBlank(workerId, "WorkerID is required to initialize PeriodicShardSyncManager.");
|
||||
Validate.notNull(leaderDecider, "LeaderDecider is required to initialize PeriodicShardSyncManager.");
|
||||
Validate.notNull(shardSyncTask, "ShardSyncTask is required to initialize PeriodicShardSyncManager.");
|
||||
this.workerId = workerId;
|
||||
this.leaderDecider = leaderDecider;
|
||||
this.metricsEmittingShardSyncTask = new MetricsCollectingTaskDecorator(shardSyncTask, metricsFactory);
|
||||
this.shardSyncThreadPool = shardSyncThreadPool;
|
||||
}
|
||||
|
||||
public synchronized TaskResult start() {
|
||||
if (!isRunning) {
|
||||
final Runnable periodicShardSyncer = () -> {
|
||||
try {
|
||||
runShardSync();
|
||||
} catch (Throwable t) {
|
||||
log.error("Error during runShardSync.", t);
|
||||
}
|
||||
};
|
||||
shardSyncThreadPool.scheduleWithFixedDelay(periodicShardSyncer, INITIAL_DELAY, PERIODIC_SHARD_SYNC_INTERVAL_MILLIS,
|
||||
TimeUnit.MILLISECONDS);
|
||||
isRunning = true;
|
||||
|
||||
}
|
||||
return new TaskResult(null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs shardSync once
|
||||
* Does not schedule periodic shardSync
|
||||
* @return the result of the task
|
||||
*/
|
||||
public synchronized TaskResult syncShardsOnce() {
|
||||
|
||||
Exception lastException = null;
|
||||
try {
|
||||
if (!isRunning) {
|
||||
runShardSync();
|
||||
}
|
||||
} catch (Exception e) {
|
||||
lastException = e;
|
||||
}
|
||||
return new TaskResult(lastException);
|
||||
}
|
||||
|
||||
public void stop() {
|
||||
if (isRunning) {
|
||||
log.info(String.format("Shutting down leader decider on worker %s", workerId));
|
||||
leaderDecider.shutdown();
|
||||
log.info(String.format("Shutting down periodic shard sync task scheduler on worker %s", workerId));
|
||||
shardSyncThreadPool.shutdown();
|
||||
isRunning = false;
|
||||
}
|
||||
}
|
||||
|
||||
private void runShardSync() {
|
||||
if (leaderDecider.isLeader(workerId)) {
|
||||
log.info(String.format("WorkerId %s is a leader, running the shard sync task", workerId));
|
||||
final TaskResult taskResult = metricsEmittingShardSyncTask.call();
|
||||
if (taskResult != null && taskResult.getException() != null) {
|
||||
throw new KinesisClientLibIOException("Failed to sync shards", taskResult.getException());
|
||||
}
|
||||
} else {
|
||||
log.debug(String.format("WorkerId %s is not a leader, not running the shard sync task", workerId));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if the entire hash range is covered
|
||||
* @return true if covered, false otherwise
|
||||
*/
|
||||
public boolean hashRangeCovered() {
|
||||
// TODO: Implement method
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
|
@ -34,7 +34,6 @@ import java.util.function.Function;
|
|||
import java.util.stream.Collectors;
|
||||
|
||||
import com.google.common.annotations.VisibleForTesting;
|
||||
|
||||
import io.reactivex.plugins.RxJavaPlugins;
|
||||
import lombok.AccessLevel;
|
||||
import lombok.Getter;
|
||||
|
|
@ -45,6 +44,8 @@ import lombok.extern.slf4j.Slf4j;
|
|||
import software.amazon.awssdk.utils.Validate;
|
||||
import software.amazon.kinesis.checkpoint.CheckpointConfig;
|
||||
import software.amazon.kinesis.checkpoint.ShardRecordProcessorCheckpointer;
|
||||
import software.amazon.kinesis.common.InitialPositionInStreamExtended;
|
||||
import software.amazon.kinesis.leases.HierarchicalShardSyncer;
|
||||
import software.amazon.kinesis.common.StreamConfig;
|
||||
import software.amazon.kinesis.common.StreamIdentifier;
|
||||
import software.amazon.kinesis.leases.Lease;
|
||||
|
|
@ -61,7 +62,10 @@ import software.amazon.kinesis.leases.HierarchicalShardSyncer;
|
|||
import software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseCoordinator;
|
||||
import software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseSerializer;
|
||||
import software.amazon.kinesis.leases.dynamodb.DynamoDBMultiStreamLeaseSerializer;
|
||||
import software.amazon.kinesis.leases.exceptions.DependencyException;
|
||||
import software.amazon.kinesis.leases.exceptions.InvalidStateException;
|
||||
import software.amazon.kinesis.leases.exceptions.LeasingException;
|
||||
import software.amazon.kinesis.leases.exceptions.ProvisionedThroughputException;
|
||||
import software.amazon.kinesis.lifecycle.LifecycleConfig;
|
||||
import software.amazon.kinesis.lifecycle.ShardConsumer;
|
||||
import software.amazon.kinesis.lifecycle.ShardConsumerArgument;
|
||||
|
|
@ -81,6 +85,20 @@ import software.amazon.kinesis.retrieval.AggregatorUtil;
|
|||
import software.amazon.kinesis.retrieval.RecordsPublisher;
|
||||
import software.amazon.kinesis.retrieval.RetrievalConfig;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.ThreadLocalRandom;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
|
|
@ -89,6 +107,11 @@ import software.amazon.kinesis.retrieval.RetrievalConfig;
|
|||
@Slf4j
|
||||
public class Scheduler implements Runnable {
|
||||
|
||||
private static final int PERIODIC_SHARD_SYNC_MAX_WORKERS_DEFAULT = 1;
|
||||
private static final long LEASE_TABLE_CHECK_FREQUENCY_MILLIS = 3 * 1000L;
|
||||
private static final long MIN_WAIT_TIME_FOR_LEASE_TABLE_CHECK_MILLIS = 5 * 1000L;
|
||||
private static final long MAX_WAIT_TIME_FOR_LEASE_TABLE_CHECK_MILLIS = 30 * 1000L;
|
||||
private static final long HASH_RANGE_COVERAGE_CHECK_FREQUENCY_MILLIS = 5000L;
|
||||
private SchedulerLog slog = new SchedulerLog();
|
||||
|
||||
private final CheckpointConfig checkpointConfig;
|
||||
|
|
@ -113,11 +136,14 @@ public class Scheduler implements Runnable {
|
|||
private final LeaseCoordinator leaseCoordinator;
|
||||
private final Function<StreamIdentifier, ShardSyncTaskManager> shardSyncTaskManagerProvider;
|
||||
private final Map<StreamIdentifier, ShardSyncTaskManager> streamToShardSyncTaskManagerMap = new HashMap<>();
|
||||
private final ShardSyncTaskManager shardSyncTaskManager;
|
||||
private final PeriodicShardSyncManager leaderElectedPeriodicShardSyncManager;
|
||||
private final ShardPrioritization shardPrioritization;
|
||||
private final boolean cleanupLeasesUponShardCompletion;
|
||||
private final boolean skipShardSyncAtWorkerInitializationIfLeasesExist;
|
||||
private final GracefulShutdownCoordinator gracefulShutdownCoordinator;
|
||||
private final WorkerStateChangeListener workerStateChangeListener;
|
||||
private final InitialPositionInStreamExtended initialPosition;
|
||||
private final MetricsFactory metricsFactory;
|
||||
private final long failoverTimeMillis;
|
||||
private final long taskBackoffTimeMillis;
|
||||
|
|
@ -131,6 +157,7 @@ public class Scheduler implements Runnable {
|
|||
private final AggregatorUtil aggregatorUtil;
|
||||
private final HierarchicalShardSyncer hierarchicalShardSyncer;
|
||||
private final long schedulerInitializationBackoffTimeMillis;
|
||||
private final LeaderDecider leaderDecider;
|
||||
|
||||
// Holds consumers for shards the worker is currently tracking. Key is shard
|
||||
// info, value is ShardConsumer.
|
||||
|
|
@ -236,6 +263,9 @@ public class Scheduler implements Runnable {
|
|||
this.workerStateChangeListener = this.coordinatorConfig.coordinatorFactory()
|
||||
.createWorkerStateChangeListener();
|
||||
}
|
||||
this.leaderDecider = new DeterministicShuffleShardSyncLeaderDecider(leaseRefresher,
|
||||
Executors.newSingleThreadScheduledExecutor(), PERIODIC_SHARD_SYNC_MAX_WORKERS_DEFAULT);
|
||||
this.initialPosition = retrievalConfig.initialPositionInStreamExtended();
|
||||
this.failoverTimeMillis = this.leaseManagementConfig.failoverTimeMillis();
|
||||
this.taskBackoffTimeMillis = this.lifecycleConfig.taskBackoffTimeMillis();
|
||||
// this.retryGetRecordsInSeconds = this.retrievalConfig.retryGetRecordsInSeconds();
|
||||
|
|
@ -248,6 +278,7 @@ public class Scheduler implements Runnable {
|
|||
// TODO : Halo : Check if this needs to be per stream.
|
||||
this.hierarchicalShardSyncer = leaseManagementConfig.hierarchicalShardSyncer(isMultiStreamMode);
|
||||
this.schedulerInitializationBackoffTimeMillis = this.coordinatorConfig.schedulerInitializationBackoffTimeMillis();
|
||||
this.leaderElectedPeriodicShardSyncManager = buildPeriodicShardSyncManager();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -267,7 +298,6 @@ public class Scheduler implements Runnable {
|
|||
workerStateChangeListener.onAllInitializationAttemptsFailed(e);
|
||||
shutdown();
|
||||
}
|
||||
|
||||
while (!shouldShutdown()) {
|
||||
runProcessLoop();
|
||||
}
|
||||
|
|
@ -294,6 +324,7 @@ public class Scheduler implements Runnable {
|
|||
if (!skipShardSyncAtWorkerInitializationIfLeasesExist || leaseRefresher.isLeaseTableEmpty()) {
|
||||
// TODO: Resume the shard sync from failed stream in the next attempt, to avoid syncing
|
||||
// TODO: for already synced streams
|
||||
waitUntilLeaseTableIsReady();
|
||||
for(Map.Entry<StreamIdentifier, StreamConfig> streamConfigEntry : currentStreamConfigMap.entrySet()) {
|
||||
final StreamIdentifier streamIdentifier = streamConfigEntry.getKey();
|
||||
log.info("Syncing Kinesis shard info for " + streamIdentifier);
|
||||
|
|
@ -321,6 +352,10 @@ public class Scheduler implements Runnable {
|
|||
} else {
|
||||
log.info("LeaseCoordinator is already running. No need to start it.");
|
||||
}
|
||||
log.info("Scheduling periodicShardSync)");
|
||||
// leaderElectedPeriodicShardSyncManager.start();
|
||||
// TODO: enable periodicShardSync after https://github.com/jushkem/amazon-kinesis-client/pull/2 is merged
|
||||
waitUntilHashRangeCovered();
|
||||
isDone = true;
|
||||
} catch (LeasingException e) {
|
||||
log.error("Caught exception when initializing LeaseCoordinator", e);
|
||||
|
|
@ -332,6 +367,7 @@ public class Scheduler implements Runnable {
|
|||
if (!isDone) {
|
||||
try {
|
||||
Thread.sleep(schedulerInitializationBackoffTimeMillis);
|
||||
leaderElectedPeriodicShardSyncManager.stop();
|
||||
} catch (InterruptedException e) {
|
||||
log.debug("Sleep interrupted while initializing worker.");
|
||||
}
|
||||
|
|
@ -345,6 +381,29 @@ public class Scheduler implements Runnable {
|
|||
}
|
||||
}
|
||||
|
||||
@VisibleForTesting
|
||||
void waitUntilLeaseTableIsReady() throws InterruptedException,
|
||||
DependencyException, ProvisionedThroughputException, InvalidStateException {
|
||||
long waitTime = ThreadLocalRandom.current().nextLong(MIN_WAIT_TIME_FOR_LEASE_TABLE_CHECK_MILLIS, MAX_WAIT_TIME_FOR_LEASE_TABLE_CHECK_MILLIS);
|
||||
long waitUntil = System.currentTimeMillis() + waitTime;
|
||||
|
||||
while (System.currentTimeMillis() < waitUntil && leaseRefresher.isLeaseTableEmpty()) {
|
||||
// check every 3 seconds if lease table is still empty,
|
||||
// to minimize contention between all workers bootstrapping at the same time
|
||||
log.info("Lease table is still empty. Checking again in {} ms", LEASE_TABLE_CHECK_FREQUENCY_MILLIS);
|
||||
Thread.sleep(LEASE_TABLE_CHECK_FREQUENCY_MILLIS);
|
||||
}
|
||||
}
|
||||
|
||||
private void waitUntilHashRangeCovered() throws InterruptedException {
|
||||
|
||||
while (!leaderElectedPeriodicShardSyncManager.hashRangeCovered()) {
|
||||
// wait until entire hash range is covered
|
||||
log.info("Hash range is not covered yet. Checking again in {} ms", HASH_RANGE_COVERAGE_CHECK_FREQUENCY_MILLIS);
|
||||
Thread.sleep(HASH_RANGE_COVERAGE_CHECK_FREQUENCY_MILLIS);
|
||||
}
|
||||
}
|
||||
|
||||
@VisibleForTesting
|
||||
void runProcessLoop() {
|
||||
try {
|
||||
|
|
@ -555,6 +614,7 @@ public class Scheduler implements Runnable {
|
|||
// Lost leases will force Worker to begin shutdown process for all shard consumers in
|
||||
// Worker.run().
|
||||
leaseCoordinator.stop();
|
||||
leaderElectedPeriodicShardSyncManager.stop();
|
||||
workerStateChangeListener.onWorkerStateChange(WorkerStateChangeListener.WorkerState.SHUT_DOWN);
|
||||
}
|
||||
}
|
||||
|
|
@ -662,12 +722,20 @@ public class Scheduler implements Runnable {
|
|||
hierarchicalShardSyncer,
|
||||
metricsFactory);
|
||||
return new ShardConsumer(cache, executorService, shardInfo, lifecycleConfig.logWarningForTaskAfterMillis(),
|
||||
argument, lifecycleConfig.taskExecutionListener(),lifecycleConfig.readTimeoutsToIgnoreBeforeWarning());
|
||||
argument, lifecycleConfig.taskExecutionListener(), lifecycleConfig.readTimeoutsToIgnoreBeforeWarning());
|
||||
}
|
||||
|
||||
private PeriodicShardSyncManager buildPeriodicShardSyncManager() {
|
||||
final ShardSyncTask shardSyncTask = new ShardSyncTask(shardDetectorProvider.apply(getStreamIdentifier), leaseRefresher, initialPosition,
|
||||
cleanupLeasesUponShardCompletion, ignoreUnexpetedChildShards, 0L, hierarchicalShardSyncer,
|
||||
metricsFactory);
|
||||
return new PeriodicShardSyncManager(leaseManagementConfig.workerIdentifier(),
|
||||
leaderDecider, shardSyncTask, metricsFactory);
|
||||
}
|
||||
|
||||
/**
|
||||
* NOTE: This method is internal/private to the Worker class. It has package access solely for testing.
|
||||
*
|
||||
* <p>
|
||||
* This method relies on ShardInfo.equals() method returning true for ShardInfo objects which may have been
|
||||
* instantiated with parentShardIds in a different order (and rest of the fields being the equal). For example
|
||||
* shardInfo1.equals(shardInfo2) should return true with shardInfo1 and shardInfo2 defined as follows. ShardInfo
|
||||
|
|
|
|||
|
|
@ -14,14 +14,6 @@
|
|||
*/
|
||||
package software.amazon.kinesis.leases;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import com.google.common.collect.Collections2;
|
||||
import lombok.EqualsAndHashCode;
|
||||
import lombok.Getter;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
|
@ -30,6 +22,12 @@ import lombok.ToString;
|
|||
import lombok.experimental.Accessors;
|
||||
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* This class contains data pertaining to a Lease. Distributed systems may use leases to partition work across a
|
||||
* fleet of workers. Each unit of work (identified by a leaseKey) has a corresponding Lease. Every worker will contend
|
||||
|
|
|
|||
|
|
@ -0,0 +1,125 @@
|
|||
/*
|
||||
* Copyright 2020 Amazon.com, Inc. or its affiliates.
|
||||
* 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 software.amazon.kinesis.coordinator;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.runners.MockitoJUnitRunner;
|
||||
import software.amazon.kinesis.leases.Lease;
|
||||
import software.amazon.kinesis.leases.LeaseRefresher;
|
||||
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Random;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.mockito.Mockito.when;
|
||||
import static software.amazon.kinesis.coordinator.DeterministicShuffleShardSyncLeaderDecider.DETERMINISTIC_SHUFFLE_SEED;
|
||||
|
||||
@RunWith(MockitoJUnitRunner.class)
|
||||
public class DeterministicShuffleShardSyncLeaderDeciderTest {
|
||||
private static final String LEASE_KEY = "lease_key";
|
||||
private static final String LEASE_OWNER = "lease_owner";
|
||||
private static final String WORKER_ID = "worker-id";
|
||||
|
||||
private DeterministicShuffleShardSyncLeaderDecider leaderDecider;
|
||||
|
||||
@Mock
|
||||
private LeaseRefresher leaseRefresher;
|
||||
|
||||
@Mock
|
||||
private ScheduledExecutorService scheduledExecutorService;
|
||||
|
||||
private int numShardSyncWorkers;
|
||||
|
||||
@Before
|
||||
public void setup() {
|
||||
numShardSyncWorkers = 1;
|
||||
leaderDecider = new DeterministicShuffleShardSyncLeaderDecider(leaseRefresher, scheduledExecutorService, numShardSyncWorkers);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testLeaderElectionWithNullLeases() {
|
||||
boolean isLeader = leaderDecider.isLeader(WORKER_ID);
|
||||
assertTrue("IsLeader should return true if leaders is null", isLeader);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testLeaderElectionWithEmptyLeases() throws Exception {
|
||||
when(leaseRefresher.listLeases()).thenReturn(new ArrayList<>());
|
||||
boolean isLeader = leaderDecider.isLeader(WORKER_ID);
|
||||
assertTrue("IsLeader should return true if no leases are returned", isLeader);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testElectedLeadersAsPerExpectedShufflingOrder()
|
||||
throws Exception {
|
||||
List<Lease> leases = getLeases(5, false /* duplicateLeaseOwner */, true /* activeLeases */);
|
||||
when(leaseRefresher.listLeases()).thenReturn(leases);
|
||||
Set<String> expectedLeaders = getExpectedLeaders(leases);
|
||||
for (String leader : expectedLeaders) {
|
||||
assertTrue(leaderDecider.isLeader(leader));
|
||||
}
|
||||
for (Lease lease : leases) {
|
||||
if (!expectedLeaders.contains(lease.leaseOwner())) {
|
||||
assertFalse(leaderDecider.isLeader(lease.leaseOwner()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testElectedLeadersAsPerExpectedShufflingOrderWhenUniqueWorkersLessThanMaxLeaders() {
|
||||
this.numShardSyncWorkers = 5; // More than number of unique lease owners
|
||||
leaderDecider = new DeterministicShuffleShardSyncLeaderDecider(leaseRefresher, scheduledExecutorService, numShardSyncWorkers);
|
||||
List<Lease> leases = getLeases(3, false /* duplicateLeaseOwner */, true /* activeLeases */);
|
||||
Set<String> expectedLeaders = getExpectedLeaders(leases);
|
||||
// All lease owners should be present in expected leaders set, and they should all be leaders.
|
||||
for (Lease lease : leases) {
|
||||
assertTrue(leaderDecider.isLeader(lease.leaseOwner()));
|
||||
assertTrue(expectedLeaders.contains(lease.leaseOwner()));
|
||||
}
|
||||
}
|
||||
|
||||
private List<Lease> getLeases(int count, boolean duplicateLeaseOwner, boolean activeLeases) {
|
||||
List<Lease> leases = new ArrayList<>();
|
||||
for (int i = 0; i < count; i++) {
|
||||
Lease lease = new Lease();
|
||||
lease.leaseKey(LEASE_KEY + i);
|
||||
lease.checkpoint(activeLeases ? ExtendedSequenceNumber.LATEST : ExtendedSequenceNumber.SHARD_END);
|
||||
lease.leaseCounter(new Random().nextLong());
|
||||
lease.leaseOwner(LEASE_OWNER + (duplicateLeaseOwner ? "" : i));
|
||||
leases.add(lease);
|
||||
}
|
||||
return leases;
|
||||
}
|
||||
|
||||
private Set<String> getExpectedLeaders(List<Lease> leases) {
|
||||
List<String> uniqueHosts = leases.stream().filter(lease -> lease.leaseOwner() != null)
|
||||
.map(Lease::leaseOwner).distinct().sorted().collect(Collectors.toList());
|
||||
|
||||
Collections.shuffle(uniqueHosts, new Random(DETERMINISTIC_SHUFFLE_SEED));
|
||||
int numWorkers = Math.min(uniqueHosts.size(), this.numShardSyncWorkers);
|
||||
return new HashSet<>(uniqueHosts.subList(0, numWorkers));
|
||||
}
|
||||
}
|
||||
|
|
@ -105,6 +105,9 @@ public class SchedulerTest {
|
|||
private final String applicationName = "applicationName";
|
||||
private final String streamName = "streamName";
|
||||
private final String namespace = "testNamespace";
|
||||
private static final long MIN_WAIT_TIME_FOR_LEASE_TABLE_CHECK_MILLIS = 5 * 1000L;
|
||||
private static final long MAX_WAIT_TIME_FOR_LEASE_TABLE_CHECK_MILLIS = 30 * 1000L;
|
||||
private static final long LEASE_TABLE_CHECK_FREQUENCY_MILLIS = 3 * 1000L;
|
||||
|
||||
private Scheduler scheduler;
|
||||
private ShardRecordProcessorFactory shardRecordProcessorFactory;
|
||||
|
|
@ -373,6 +376,43 @@ public class SchedulerTest {
|
|||
|
||||
}
|
||||
|
||||
@Test
|
||||
public final void testInitializationWaitsWhenLeaseTableIsEmpty() throws Exception {
|
||||
final int maxInitializationAttempts = 1;
|
||||
coordinatorConfig.maxInitializationAttempts(maxInitializationAttempts);
|
||||
coordinatorConfig.skipShardSyncAtWorkerInitializationIfLeasesExist(false);
|
||||
scheduler = new Scheduler(checkpointConfig, coordinatorConfig, leaseManagementConfig, lifecycleConfig,
|
||||
metricsConfig, processorConfig, retrievalConfig);
|
||||
|
||||
doNothing().when(leaseCoordinator).initialize();
|
||||
when(dynamoDBLeaseRefresher.isLeaseTableEmpty()).thenReturn(true);
|
||||
|
||||
long startTime = System.currentTimeMillis();
|
||||
scheduler.waitUntilLeaseTableIsReady();
|
||||
long endTime = System.currentTimeMillis();
|
||||
|
||||
assertTrue(endTime - startTime > MIN_WAIT_TIME_FOR_LEASE_TABLE_CHECK_MILLIS);
|
||||
assertTrue(endTime - startTime < (MAX_WAIT_TIME_FOR_LEASE_TABLE_CHECK_MILLIS + LEASE_TABLE_CHECK_FREQUENCY_MILLIS));
|
||||
}
|
||||
|
||||
@Test
|
||||
public final void testInitializationDoesntWaitWhenLeaseTableIsNotEmpty() throws Exception {
|
||||
final int maxInitializationAttempts = 1;
|
||||
coordinatorConfig.maxInitializationAttempts(maxInitializationAttempts);
|
||||
coordinatorConfig.skipShardSyncAtWorkerInitializationIfLeasesExist(false);
|
||||
scheduler = new Scheduler(checkpointConfig, coordinatorConfig, leaseManagementConfig, lifecycleConfig,
|
||||
metricsConfig, processorConfig, retrievalConfig);
|
||||
|
||||
doNothing().when(leaseCoordinator).initialize();
|
||||
when(dynamoDBLeaseRefresher.isLeaseTableEmpty()).thenReturn(false);
|
||||
|
||||
long startTime = System.currentTimeMillis();
|
||||
scheduler.waitUntilLeaseTableIsReady();
|
||||
long endTime = System.currentTimeMillis();
|
||||
|
||||
assertTrue(endTime - startTime < MIN_WAIT_TIME_FOR_LEASE_TABLE_CHECK_MILLIS);
|
||||
}
|
||||
|
||||
@Test
|
||||
public final void testSchedulerShutdown() {
|
||||
scheduler.shutdown();
|
||||
|
|
|
|||
Loading…
Reference in a new issue