luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add a WorkerState of SHUT_DOWN_STARTED (#457)

Browse source 
Added a new WorkerState that indicates when a shutdown has started
This commit is contained in:
jiaxul 2019-01-24 12:53:39 -08:00 committed by Justin Pfifer
parent 03c15eb275
commit 8d9427e06c
3 changed files with 13 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/Scheduler.java
View file

@ -476,6 +476,7 @@ public class Scheduler implements Runnable {
log.warn("Shutdown requested a second time.");
return;
}
workerStateChangeListener.onWorkerStateChange(WorkerStateChangeListener.WorkerState.SHUT_DOWN_STARTED);
log.info("Worker shutdown requested.");
// Set shutdown flag, so Worker.run can start shutdown process.

1
amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/WorkerStateChangeListener.java
View file

@ -23,6 +23,7 @@ public interface WorkerStateChangeListener {
CREATED,
INITIALIZING,
STARTED,
SHUT_DOWN_STARTED,
SHUT_DOWN
}

12
amazon-kinesis-client/src/test/java/software/amazon/kinesis/coordinator/SchedulerTest.java
View file

@ -116,13 +116,15 @@ public class SchedulerTest {
private ShardDetector shardDetector;
@Mock
private Checkpointer checkpoint;
@Mock
private WorkerStateChangeListener workerStateChangeListener;
@Before
public void setup() {
shardRecordProcessorFactory = new TestShardRecordProcessorFactory();
checkpointConfig = new CheckpointConfig().checkpointFactory(new TestKinesisCheckpointFactory());
coordinatorConfig = new CoordinatorConfig(applicationName).parentShardPollIntervalMillis(100L);
coordinatorConfig = new CoordinatorConfig(applicationName).parentShardPollIntervalMillis(100L).workerStateChangeListener(workerStateChangeListener);
leaseManagementConfig = new LeaseManagementConfig(tableName, dynamoDBClient, kinesisClient, streamName,
workerIdentifier).leaseManagementFactory(new TestKinesisLeaseManagementFactory());
lifecycleConfig = new LifecycleConfig();
@ -259,6 +261,14 @@ public class SchedulerTest {
verify(shardDetector, times(maxInitializationAttempts)).listShards();
}
@Test
public final void testSchedulerShutdown() {
scheduler.shutdown();
verify(workerStateChangeListener, times(1)).onWorkerStateChange(WorkerStateChangeListener.WorkerState.SHUT_DOWN_STARTED);
verify(leaseCoordinator, times(1)).stop();
verify(workerStateChangeListener, times(1)).onWorkerStateChange(WorkerStateChangeListener.WorkerState.SHUT_DOWN);
}
/*private void runAndTestWorker(int numShards, int threadPoolSize) throws Exception {
final int numberOfRecordsPerShard = 10;