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

Changing invocation system for ShardConsumer. Still working on it though.

Browse source
This commit is contained in:
Pfifer, Justin 2018-03-27 10:38:49 -07:00
parent f130e4c79c
commit 94c5125352
20 changed files with 445 additions and 374 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ShardSyncTask.java
View file

@ -15,13 +15,14 @@
package software.amazon.kinesis.leases;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStreamExtended;
import lombok.extern.slf4j.Slf4j;
import software.amazon.kinesis.lifecycle.ITask;
import software.amazon.kinesis.lifecycle.TaskCompletedListener;
import software.amazon.kinesis.lifecycle.TaskResult;
import software.amazon.kinesis.lifecycle.TaskType;
import software.amazon.kinesis.retrieval.IKinesisProxy;
import lombok.extern.slf4j.Slf4j;
/**
* This task syncs leases/activies with shards of the stream.
* It will create new leases/activites when it discovers new shards (e.g. setup/resharding).
@ -38,6 +39,8 @@ public class ShardSyncTask implements ITask {
private final long shardSyncTaskIdleTimeMillis;
private final TaskType taskType = TaskType.SHARDSYNC;
private TaskCompletedListener listener;
/**
* @param kinesisProxy Used to fetch information about the stream (e.g. shard list)
* @param leaseManager Used to fetch and create leases
@ -64,23 +67,26 @@ public class ShardSyncTask implements ITask {
*/
@Override
public TaskResult call() {
Exception exception = null;
try {
ShardSyncer.checkAndCreateLeasesForNewShards(kinesisProxy,
leaseManager,
initialPosition,
cleanupLeasesUponShardCompletion,
ignoreUnexpectedChildShards);
if (shardSyncTaskIdleTimeMillis > 0) {
Thread.sleep(shardSyncTaskIdleTimeMillis);
}
} catch (Exception e) {
log.error("Caught exception while sync'ing Kinesis shards and leases", e);
exception = e;
}
Exception exception = null;
return new TaskResult(exception);
try {
ShardSyncer.checkAndCreateLeasesForNewShards(kinesisProxy, leaseManager, initialPosition,
cleanupLeasesUponShardCompletion, ignoreUnexpectedChildShards);
if (shardSyncTaskIdleTimeMillis > 0) {
Thread.sleep(shardSyncTaskIdleTimeMillis);
}
} catch (Exception e) {
log.error("Caught exception while sync'ing Kinesis shards and leases", e);
exception = e;
}
return new TaskResult(exception);
} finally {
if (listener != null) {
listener.taskCompleted(this);
}
}
}
@ -92,4 +98,12 @@ public class ShardSyncTask implements ITask {
return taskType;
}
@Override
public void addTaskCompletedListener(TaskCompletedListener taskCompletedListener) {
if (listener != null) {
log.warn("Listener is being reset, this shouldn't happen");
}
listener = taskCompletedListener;
}
}

84
amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/BlockOnParentShardTask.java
View file

@ -15,12 +15,12 @@
package software.amazon.kinesis.lifecycle;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.internal.BlockedOnParentShardException;
import software.amazon.kinesis.leases.ShardInfo;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
import software.amazon.kinesis.leases.KinesisClientLease;
import software.amazon.kinesis.leases.ILeaseManager;
import lombok.extern.slf4j.Slf4j;
import software.amazon.kinesis.leases.ILeaseManager;
import software.amazon.kinesis.leases.KinesisClientLease;
import software.amazon.kinesis.leases.ShardInfo;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
/**
* Task to block until processing of all data records in the parent shard(s) is completed.
@ -38,6 +38,8 @@ public class BlockOnParentShardTask implements ITask {
private final TaskType taskType = TaskType.BLOCK_ON_PARENT_SHARDS;
// Sleep for this duration if the parent shards have not completed processing, or we encounter an exception.
private final long parentShardPollIntervalMillis;
private TaskCompletedListener listener;
/**
* @param shardInfo Information about the shard we are working on
@ -57,43 +59,49 @@ public class BlockOnParentShardTask implements ITask {
*/
@Override
public TaskResult call() {
Exception exception = null;
try {
boolean blockedOnParentShard = false;
for (String shardId : shardInfo.getParentShardIds()) {
KinesisClientLease lease = leaseManager.getLease(shardId);
if (lease != null) {
ExtendedSequenceNumber checkpoint = lease.getCheckpoint();
if ((checkpoint == null) || (!checkpoint.equals(ExtendedSequenceNumber.SHARD_END))) {
log.debug("Shard {} is not yet done. Its current checkpoint is {}", shardId, checkpoint);
blockedOnParentShard = true;
exception = new BlockedOnParentShardException("Parent shard not yet done");
break;
Exception exception = null;
try {
boolean blockedOnParentShard = false;
for (String shardId : shardInfo.getParentShardIds()) {
KinesisClientLease lease = leaseManager.getLease(shardId);
if (lease != null) {
ExtendedSequenceNumber checkpoint = lease.getCheckpoint();
if ((checkpoint == null) || (!checkpoint.equals(ExtendedSequenceNumber.SHARD_END))) {
log.debug("Shard {} is not yet done. Its current checkpoint is {}", shardId, checkpoint);
blockedOnParentShard = true;
exception = new BlockedOnParentShardException("Parent shard not yet done");
break;
} else {
log.debug("Shard {} has been completely processed.", shardId);
}
} else {
log.debug("Shard {} has been completely processed.", shardId);
log.info("No lease found for shard {}. Not blocking on completion of this shard.", shardId);
}
} else {
log.info("No lease found for shard {}. Not blocking on completion of this shard.", shardId);
}
if (!blockedOnParentShard) {
log.info("No need to block on parents {} of shard {}", shardInfo.getParentShardIds(),
shardInfo.getShardId());
return new TaskResult(null);
}
} catch (Exception e) {
log.error("Caught exception when checking for parent shard checkpoint", e);
exception = e;
}
if (!blockedOnParentShard) {
log.info("No need to block on parents {} of shard {}", shardInfo.getParentShardIds(),
shardInfo.getShardId());
return new TaskResult(null);
try {
Thread.sleep(parentShardPollIntervalMillis);
} catch (InterruptedException e) {
log.error("Sleep interrupted when waiting on parent shard(s) of {}", shardInfo.getShardId(), e);
}
return new TaskResult(exception);
} finally {
if (listener != null) {
listener.taskCompleted(this);
}
} catch (Exception e) {
log.error("Caught exception when checking for parent shard checkpoint", e);
exception = e;
}
try {
Thread.sleep(parentShardPollIntervalMillis);
} catch (InterruptedException e) {
log.error("Sleep interrupted when waiting on parent shard(s) of {}", shardInfo.getShardId(), e);
}
return new TaskResult(exception);
}
/* (non-Javadoc)
@ -104,4 +112,12 @@ public class BlockOnParentShardTask implements ITask {
return taskType;
}
@Override
public void addTaskCompletedListener(TaskCompletedListener taskCompletedListener) {
if (listener != null) {
log.warn("Listener is being reset, this shouldn't happen");
}
listener = taskCompletedListener;
}
}

11
amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ITask.java
View file

@ -14,9 +14,6 @@
*/
package software.amazon.kinesis.lifecycle;
import software.amazon.kinesis.lifecycle.TaskResult;
import software.amazon.kinesis.lifecycle.TaskType;
import java.util.concurrent.Callable;
/**
@ -38,4 +35,12 @@ public interface ITask extends Callable<TaskResult> {
*/
TaskType getTaskType();
/**
* Adds a listener that will be notified once the task is completed.
*
* @param taskCompletedListener
* the listener to call once the task has been completed
*/
void addTaskCompletedListener(TaskCompletedListener taskCompletedListener);
}

5
amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/InitializeTask.java
View file

@ -134,4 +134,9 @@ public class InitializeTask implements ITask {
return taskType;
}
@Override
public void addTaskCompletedListener(TaskCompletedListener taskCompletedListener) {
}
}

79
amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ProcessTask.java
View file

@ -53,6 +53,7 @@ public class ProcessTask implements ITask {
private final ThrottlingReporter throttlingReporter;
private final ProcessRecordsInput processRecordsInput;
private TaskCompletedListener listener;
@RequiredArgsConstructor
public static class RecordsFetcher {
@ -141,43 +142,49 @@ public class ProcessTask implements ITask {
*/
@Override
public TaskResult call() {
long startTimeMillis = System.currentTimeMillis();
IMetricsScope scope = MetricsHelper.getMetricsScope();
scope.addDimension(MetricsHelper.SHARD_ID_DIMENSION_NAME, shardInfo.getShardId());
scope.addData(RECORDS_PROCESSED_METRIC, 0, StandardUnit.Count, MetricsLevel.SUMMARY);
scope.addData(DATA_BYTES_PROCESSED_METRIC, 0, StandardUnit.Bytes, MetricsLevel.SUMMARY);
Exception exception = null;
try {
if (processRecordsInput.isAtShardEnd()) {
log.info("Reached end of shard {}", shardInfo.getShardId());
return new TaskResult(null, true);
long startTimeMillis = System.currentTimeMillis();
IMetricsScope scope = MetricsHelper.getMetricsScope();
scope.addDimension(MetricsHelper.SHARD_ID_DIMENSION_NAME, shardInfo.getShardId());
scope.addData(RECORDS_PROCESSED_METRIC, 0, StandardUnit.Count, MetricsLevel.SUMMARY);
scope.addData(DATA_BYTES_PROCESSED_METRIC, 0, StandardUnit.Bytes, MetricsLevel.SUMMARY);
Exception exception = null;
try {
if (processRecordsInput.isAtShardEnd()) {
log.info("Reached end of shard {}", shardInfo.getShardId());
return new TaskResult(null, true);
}
throttlingReporter.success();
List<Record> records = processRecordsInput.getRecords();
if (!records.isEmpty()) {
scope.addData(RECORDS_PROCESSED_METRIC, records.size(), StandardUnit.Count, MetricsLevel.SUMMARY);
} else {
handleNoRecords(startTimeMillis);
}
records = deaggregateRecords(records);
recordProcessorCheckpointer.setLargestPermittedCheckpointValue(filterAndGetMaxExtendedSequenceNumber(
scope, records, recordProcessorCheckpointer.getLastCheckpointValue(),
recordProcessorCheckpointer.getLargestPermittedCheckpointValue()));
if (shouldCallProcessRecords(records)) {
callProcessRecords(processRecordsInput, records);
}
} catch (RuntimeException e) {
log.error("ShardId {}: Caught exception: ", shardInfo.getShardId(), e);
exception = e;
backoff();
}
throttlingReporter.success();
List<Record> records = processRecordsInput.getRecords();
if (!records.isEmpty()) {
scope.addData(RECORDS_PROCESSED_METRIC, records.size(), StandardUnit.Count, MetricsLevel.SUMMARY);
} else {
handleNoRecords(startTimeMillis);
return new TaskResult(exception);
} finally {
if (listener != null) {
listener.taskCompleted(this);
}
records = deaggregateRecords(records);
recordProcessorCheckpointer.setLargestPermittedCheckpointValue(filterAndGetMaxExtendedSequenceNumber(scope,
records, recordProcessorCheckpointer.getLastCheckpointValue(),
recordProcessorCheckpointer.getLargestPermittedCheckpointValue()));
if (shouldCallProcessRecords(records)) {
callProcessRecords(processRecordsInput, records);
}
} catch (RuntimeException e) {
log.error("ShardId {}: Caught exception: ", shardInfo.getShardId(), e);
exception = e;
backoff();
}
return new TaskResult(exception);
}
/**
@ -281,6 +288,14 @@ public class ProcessTask implements ITask {
return taskType;
}
@Override
public void addTaskCompletedListener(TaskCompletedListener taskCompletedListener) {
if (listener != null) {
log.warn("Listener is being reset, this shouldn't happen");
}
listener = taskCompletedListener;
}
/**
* Scans a list of records to filter out records up to and including the most recent checkpoint value and to get the
* greatest extended sequence number from the retained records. Also emits metrics about the records.

202
amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShardConsumer.java
View file

@ -15,38 +15,31 @@
package software.amazon.kinesis.lifecycle;
import java.time.Duration;
import java.time.Instant;
import java.util.EnumSet;
import java.util.Optional;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.internal.BlockedOnParentShardException;
import lombok.Data;
import lombok.Synchronized;
import org.apache.commons.lang3.StringUtils;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
import software.amazon.kinesis.checkpoint.Checkpoint;
import software.amazon.kinesis.lifecycle.events.LeaseLost;
import software.amazon.kinesis.lifecycle.events.RecordsReceived;
import software.amazon.kinesis.lifecycle.events.ShardCompleted;
import software.amazon.kinesis.lifecycle.events.ShutdownRequested;
import software.amazon.kinesis.lifecycle.events.Started;
import software.amazon.kinesis.metrics.MetricsCollectingTaskDecorator;
import software.amazon.kinesis.coordinator.RecordProcessorCheckpointer;
import software.amazon.kinesis.leases.ShardInfo;
import software.amazon.kinesis.coordinator.StreamConfig;
import software.amazon.kinesis.processor.ICheckpoint;
import software.amazon.kinesis.processor.IRecordProcessor;
import software.amazon.kinesis.leases.KinesisClientLease;
import software.amazon.kinesis.leases.ILeaseManager;
import software.amazon.kinesis.metrics.IMetricsFactory;
import com.google.common.annotations.VisibleForTesting;
import lombok.Getter;
import lombok.Synchronized;
import lombok.extern.slf4j.Slf4j;
import software.amazon.kinesis.checkpoint.Checkpoint;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
import software.amazon.kinesis.coordinator.RecordProcessorCheckpointer;
import software.amazon.kinesis.coordinator.StreamConfig;
import software.amazon.kinesis.leases.ILeaseManager;
import software.amazon.kinesis.leases.KinesisClientLease;
import software.amazon.kinesis.leases.ShardInfo;
import software.amazon.kinesis.metrics.IMetricsFactory;
import software.amazon.kinesis.metrics.MetricsCollectingTaskDecorator;
import software.amazon.kinesis.processor.ICheckpoint;
import software.amazon.kinesis.processor.IRecordProcessor;
import software.amazon.kinesis.retrieval.AsynchronousGetRecordsRetrievalStrategy;
import software.amazon.kinesis.retrieval.GetRecordsCache;
import software.amazon.kinesis.retrieval.GetRecordsRetrievalStrategy;
@ -59,11 +52,10 @@ import software.amazon.kinesis.retrieval.SynchronousGetRecordsRetrievalStrategy;
* A new instance should be created if the primary responsibility is reassigned back to this process.
*/
@Slf4j
public class ShardConsumer implements RecordProcessorLifecycle {
public class ShardConsumer {
//<editor-fold desc="Class Variables">
private final StreamConfig streamConfig;
private final IRecordProcessor recordProcessor;
private RecordProcessorLifecycle recordProcessorLifecycle;
private final KinesisClientLibConfiguration config;
private final RecordProcessorCheckpointer recordProcessorCheckpointer;
private final ExecutorService executorService;
@ -81,6 +73,8 @@ public class ShardConsumer implements RecordProcessorLifecycle {
private ITask currentTask;
private long currentTaskSubmitTime;
private Future<TaskResult> future;
private boolean started = false;
private Instant taskDispatchedAt;
//</editor-fold>
//<editor-fold desc="Cache Management">
@ -265,21 +259,46 @@ public class ShardConsumer implements RecordProcessorLifecycle {
//</editor-fold>
//<editor-fold desc="Dispatch">
private void start() {
started = true;
getRecordsCache.addDataArrivedListener(this::checkAndSubmitNextTask);
checkAndSubmitNextTask();
}
/**
* No-op if current task is pending, otherwise submits next task for this shard.
* This method should NOT be called if the ShardConsumer is already in SHUTDOWN_COMPLETED state.
*
* @return true if a new process task was submitted, false otherwise
*/
public synchronized boolean consumeShard() {
return checkAndSubmitNextTask();
@Synchronized
public boolean consumeShard() {
if (!started) {
start();
}
if (taskDispatchedAt != null) {
Duration taken = Duration.between(taskDispatchedAt, Instant.now());
String commonMessage = String.format("Previous %s task still pending for shard %s since %s ago. ",
currentTask.getTaskType(), shardInfo.getShardId(), taken);
if (log.isDebugEnabled()) {
log.debug("{} Not submitting new task.", commonMessage);
}
config.getLogWarningForTaskAfterMillis().ifPresent(value -> {
if (taken.toMillis() > value) {
log.warn(commonMessage);
}
});
}
return true;
}
private boolean readyForNextTask() {
return future == null || future.isCancelled() || future.isDone();
}
private synchronized boolean checkAndSubmitNextTask() {
@Synchronized
private boolean checkAndSubmitNextTask() {
boolean submittedNewTask = false;
if (readyForNextTask()) {
TaskOutcome taskOutcome = TaskOutcome.NOT_COMPLETE;
@ -290,9 +309,11 @@ public class ShardConsumer implements RecordProcessorLifecycle {
updateState(taskOutcome);
ITask nextTask = getNextTask();
if (nextTask != null) {
nextTask.addTaskCompletedListener(this::handleTaskCompleted);
currentTask = nextTask;
try {
future = executorService.submit(currentTask);
taskDispatchedAt = Instant.now();
currentTaskSubmitTime = System.currentTimeMillis();
submittedNewTask = true;
log.debug("Submitted new {} task for shard {}", currentTask.getTaskType(), shardInfo.getShardId());
@ -325,6 +346,33 @@ public class ShardConsumer implements RecordProcessorLifecycle {
return submittedNewTask;
}
private boolean shouldDispatchNextTask() {
return !isShutdown() || shutdownReason != null || getRecordsCache.hasResultAvailable();
}
@Synchronized
private void handleTaskCompleted(ITask task) {
if (future != null) {
executorService.submit(() -> {
//
// Determine task outcome will wait on the future for us. The value of the future
//
resolveFuture();
if (shouldDispatchNextTask()) {
checkAndSubmitNextTask();
}
});
} else {
log.error("Future wasn't set. This shouldn't happen as polling should be disabled. " +
"Will trigger next task check just in case");
if (shouldDispatchNextTask()) {
checkAndSubmitNextTask();
}
}
}
public boolean isSkipShardSyncAtWorkerInitializationIfLeasesExist() {
return skipShardSyncAtWorkerInitializationIfLeasesExist;
}
@ -333,6 +381,14 @@ public class ShardConsumer implements RecordProcessorLifecycle {
SUCCESSFUL, END_OF_SHARD, NOT_COMPLETE, FAILURE
}
private void resolveFuture() {
try {
future.get();
} catch (Exception e) {
log.info("Ignoring caught exception '{}' exception during resolve.", e.getMessage());
}
}
private TaskOutcome determineTaskOutcome() {
try {
TaskResult result = future.get();
@ -426,7 +482,8 @@ public class ShardConsumer implements RecordProcessorLifecycle {
*
* @return true if shutdown is complete (false if shutdown is still in progress)
*/
public synchronized boolean beginShutdown() {
@Synchronized
public boolean beginShutdown() {
markForShutdown(ShutdownReason.ZOMBIE);
checkAndSubmitNextTask();
@ -531,95 +588,4 @@ public class ShardConsumer implements RecordProcessorLifecycle {
}
//</editor-fold>
private enum LifecycleStates {
STARTED, PROCESSING, SHUTDOWN, FAILED
}
private EnumSet<LifecycleStates> allowedStates = EnumSet.of(LifecycleStates.STARTED);
private TaskFailedListener listener;
public void addTaskFailedListener(TaskFailedListener listener) {
this.listener = listener;
}
private TaskFailureHandling taskFailed(Throwable t) {
//
// TODO: What should we do if there is no listener. I intend to require the scheduler to always register
//
if (listener != null) {
return listener.taskFailed(new TaskFailed(t));
}
return TaskFailureHandling.STOP;
}
@Data
private class TaskExecution {
private final Instant started;
private final Future<?> future;
}
ExecutorService executor = Executors.newSingleThreadExecutor();
TaskExecution taskExecution = null;
private void awaitAvailable() {
if (taskExecution != null) {
try {
taskExecution.getFuture().get();
} catch (Throwable t) {
TaskFailureHandling handling = taskFailed(t);
if (handling == TaskFailureHandling.STOP) {
allowedStates = EnumSet.of(LifecycleStates.FAILED);
}
}
}
}
private void checkState(LifecycleStates current) {
if (!allowedStates.contains(current)) {
throw new IllegalStateException("State " + current + " isn't allowed. Allowed: (" + StringUtils.join(allowedStates) + ")");
}
}
//<editor-fold desc="RecordProcessorLifecycle">
private void executeTask(LifecycleStates current, Runnable task) {
awaitAvailable();
checkState(current);
Future<?> future = executor.submit(task);
taskExecution = new TaskExecution(Instant.now(), future);
}
@Override
@Synchronized
public void started(Started started) {
executeTask(LifecycleStates.STARTED, () -> recordProcessorLifecycle.started(started));
allowedStates = EnumSet.of(LifecycleStates.PROCESSING, LifecycleStates.SHUTDOWN);
}
@Override
@Synchronized
public void recordsReceived(RecordsReceived records) {
executeTask(LifecycleStates.PROCESSING, () -> recordProcessorLifecycle.recordsReceived(records));
}
@Override
@Synchronized
public void leaseLost(LeaseLost leaseLost) {
executeTask(LifecycleStates.SHUTDOWN, () -> recordProcessorLifecycle.leaseLost(leaseLost));
allowedStates = EnumSet.of(LifecycleStates.SHUTDOWN);
}
@Override
@Synchronized
public void shardCompleted(ShardCompleted shardCompleted) {
executeTask(LifecycleStates.SHUTDOWN, () -> recordProcessorLifecycle.shardCompleted(shardCompleted));
}
@Override
@Synchronized
public void shutdownRequested(ShutdownRequested shutdownRequested) {
}
//</editor-fold>
}

11
amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShutdownNotificationTask.java
View file

@ -14,6 +14,7 @@
*/
package software.amazon.kinesis.lifecycle;
import lombok.extern.slf4j.Slf4j;
import software.amazon.kinesis.leases.ShardInfo;
import software.amazon.kinesis.processor.IRecordProcessorCheckpointer;
import software.amazon.kinesis.processor.IRecordProcessor;
@ -22,12 +23,14 @@ import software.amazon.kinesis.processor.IShutdownNotificationAware;
/**
* Notifies record processor of incoming shutdown request, and gives them a chance to checkpoint.
*/
@Slf4j
public class ShutdownNotificationTask implements ITask {
private final IRecordProcessor recordProcessor;
private final IRecordProcessorCheckpointer recordProcessorCheckpointer;
private final ShutdownNotification shutdownNotification;
private final ShardInfo shardInfo;
private TaskCompletedListener listener;
ShutdownNotificationTask(IRecordProcessor recordProcessor, IRecordProcessorCheckpointer recordProcessorCheckpointer, ShutdownNotification shutdownNotification, ShardInfo shardInfo) {
this.recordProcessor = recordProcessor;
@ -57,4 +60,12 @@ public class ShutdownNotificationTask implements ITask {
public TaskType getTaskType() {
return TaskType.SHUTDOWN_NOTIFICATION;
}
@Override
public void addTaskCompletedListener(TaskCompletedListener taskCompletedListener) {
if (listener != null) {
log.warn("Listener is being reset, this shouldn't happen");
}
listener = taskCompletedListener;
}
}

127
amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShutdownTask.java
View file

@ -49,6 +49,7 @@ public class ShutdownTask implements ITask {
private final TaskType taskType = TaskType.SHUTDOWN;
private final long backoffTimeMillis;
private final GetRecordsCache getRecordsCache;
private TaskCompletedListener listener;
/**
* Constructor.
@ -86,73 +87,79 @@ public class ShutdownTask implements ITask {
*/
@Override
public TaskResult call() {
Exception exception;
boolean applicationException = false;
try {
// If we reached end of the shard, set sequence number to SHARD_END.
if (reason == ShutdownReason.TERMINATE) {
recordProcessorCheckpointer.setSequenceNumberAtShardEnd(
recordProcessorCheckpointer.getLargestPermittedCheckpointValue());
recordProcessorCheckpointer.setLargestPermittedCheckpointValue(ExtendedSequenceNumber.SHARD_END);
}
Exception exception;
boolean applicationException = false;
log.debug("Invoking shutdown() for shard {}, concurrencyToken {}. Shutdown reason: {}",
shardInfo.getShardId(), shardInfo.getConcurrencyToken(), reason);
final ShutdownInput shutdownInput = new ShutdownInput()
.withShutdownReason(reason)
.withCheckpointer(recordProcessorCheckpointer);
final long recordProcessorStartTimeMillis = System.currentTimeMillis();
try {
recordProcessor.shutdown(shutdownInput);
ExtendedSequenceNumber lastCheckpointValue = recordProcessorCheckpointer.getLastCheckpointValue();
// If we reached end of the shard, set sequence number to SHARD_END.
if (reason == ShutdownReason.TERMINATE) {
recordProcessorCheckpointer.setSequenceNumberAtShardEnd(
recordProcessorCheckpointer.getLargestPermittedCheckpointValue());
recordProcessorCheckpointer.setLargestPermittedCheckpointValue(ExtendedSequenceNumber.SHARD_END);
}
log.debug("Invoking shutdown() for shard {}, concurrencyToken {}. Shutdown reason: {}",
shardInfo.getShardId(), shardInfo.getConcurrencyToken(), reason);
final ShutdownInput shutdownInput = new ShutdownInput()
.withShutdownReason(reason)
.withCheckpointer(recordProcessorCheckpointer);
final long recordProcessorStartTimeMillis = System.currentTimeMillis();
try {
recordProcessor.shutdown(shutdownInput);
ExtendedSequenceNumber lastCheckpointValue = recordProcessorCheckpointer.getLastCheckpointValue();
if (reason == ShutdownReason.TERMINATE) {
if ((lastCheckpointValue == null)
|| (!lastCheckpointValue.equals(ExtendedSequenceNumber.SHARD_END))) {
throw new IllegalArgumentException("Application didn't checkpoint at end of shard "
+ shardInfo.getShardId());
}
}
log.debug("Shutting down retrieval strategy.");
getRecordsCache.shutdown();
log.debug("Record processor completed shutdown() for shard {}", shardInfo.getShardId());
} catch (Exception e) {
applicationException = true;
throw e;
} finally {
MetricsHelper.addLatency(RECORD_PROCESSOR_SHUTDOWN_METRIC, recordProcessorStartTimeMillis,
MetricsLevel.SUMMARY);
}
if (reason == ShutdownReason.TERMINATE) {
if ((lastCheckpointValue == null)
|| (!lastCheckpointValue.equals(ExtendedSequenceNumber.SHARD_END))) {
throw new IllegalArgumentException("Application didn't checkpoint at end of shard "
+ shardInfo.getShardId());
}
log.debug("Looking for child shards of shard {}", shardInfo.getShardId());
// create leases for the child shards
ShardSyncer.checkAndCreateLeasesForNewShards(kinesisProxy,
leaseManager,
initialPositionInStream,
cleanupLeasesOfCompletedShards,
ignoreUnexpectedChildShards);
log.debug("Finished checking for child shards of shard {}", shardInfo.getShardId());
}
log.debug("Shutting down retrieval strategy.");
getRecordsCache.shutdown();
log.debug("Record processor completed shutdown() for shard {}", shardInfo.getShardId());
return new TaskResult(null);
} catch (Exception e) {
applicationException = true;
throw e;
} finally {
MetricsHelper.addLatency(RECORD_PROCESSOR_SHUTDOWN_METRIC, recordProcessorStartTimeMillis,
MetricsLevel.SUMMARY);
if (applicationException) {
log.error("Application exception. ", e);
} else {
log.error("Caught exception: ", e);
}
exception = e;
// backoff if we encounter an exception.
try {
Thread.sleep(this.backoffTimeMillis);
} catch (InterruptedException ie) {
log.debug("Interrupted sleep", ie);
}
}
if (reason == ShutdownReason.TERMINATE) {
log.debug("Looking for child shards of shard {}", shardInfo.getShardId());
// create leases for the child shards
ShardSyncer.checkAndCreateLeasesForNewShards(kinesisProxy,
leaseManager,
initialPositionInStream,
cleanupLeasesOfCompletedShards,
ignoreUnexpectedChildShards);
log.debug("Finished checking for child shards of shard {}", shardInfo.getShardId());
}
return new TaskResult(null);
} catch (Exception e) {
if (applicationException) {
log.error("Application exception. ", e);
} else {
log.error("Caught exception: ", e);
}
exception = e;
// backoff if we encounter an exception.
try {
Thread.sleep(this.backoffTimeMillis);
} catch (InterruptedException ie) {
log.debug("Interrupted sleep", ie);
return new TaskResult(exception);
} finally {
if (listener != null) {
listener.taskCompleted(this);
}
}
return new TaskResult(exception);
}
/*
@ -165,6 +172,14 @@ public class ShutdownTask implements ITask {
return taskType;
}
@Override
public void addTaskCompletedListener(TaskCompletedListener taskCompletedListener) {
if (listener != null) {
log.warn("Listener is being reset, this shouldn't happen");
}
listener = taskCompletedListener;
}
@VisibleForTesting
public ShutdownReason getReason() {
return reason;

25
amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/TaskCompletedListener.java Normal file
View file

@ -0,0 +1,25 @@
/*
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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.lifecycle;
public interface TaskCompletedListener {
/**
* Called once a task has completed
*
* @param task
* the task that completed
*/
void taskCompleted(ITask task);
}

50
amazon-kinesis-client/src/main/java/software/amazon/kinesis/metrics/MetricsCollectingTaskDecorator.java
View file

@ -14,7 +14,9 @@
*/
package software.amazon.kinesis.metrics;
import lombok.RequiredArgsConstructor;
import software.amazon.kinesis.lifecycle.ITask;
import software.amazon.kinesis.lifecycle.TaskCompletedListener;
import software.amazon.kinesis.lifecycle.TaskResult;
import software.amazon.kinesis.lifecycle.TaskType;
import software.amazon.kinesis.metrics.MetricsHelper;
@ -28,6 +30,7 @@ public class MetricsCollectingTaskDecorator implements ITask {
private final ITask other;
private final IMetricsFactory factory;
private DelegateTaskCompletedListener delegateTaskCompletedListener;
/**
* Constructor.
@ -45,17 +48,23 @@ public class MetricsCollectingTaskDecorator implements ITask {
*/
@Override
public TaskResult call() {
MetricsHelper.startScope(factory, other.getClass().getSimpleName());
TaskResult result = null;
final long startTimeMillis = System.currentTimeMillis();
try {
result = other.call();
MetricsHelper.startScope(factory, other.getClass().getSimpleName());
TaskResult result = null;
final long startTimeMillis = System.currentTimeMillis();
try {
result = other.call();
} finally {
MetricsHelper.addSuccessAndLatency(startTimeMillis, result != null && result.getException() == null,
MetricsLevel.SUMMARY);
MetricsHelper.endScope();
}
return result;
} finally {
MetricsHelper.addSuccessAndLatency(startTimeMillis, result != null && result.getException() == null,
MetricsLevel.SUMMARY);
MetricsHelper.endScope();
if (delegateTaskCompletedListener != null) {
delegateTaskCompletedListener.dispatchIfNeeded();
}
}
return result;
}
/**
@ -66,6 +75,31 @@ public class MetricsCollectingTaskDecorator implements ITask {
return other.getTaskType();
}
@Override
public void addTaskCompletedListener(TaskCompletedListener taskCompletedListener) {
delegateTaskCompletedListener = new DelegateTaskCompletedListener(taskCompletedListener);
other.addTaskCompletedListener(delegateTaskCompletedListener);
}
@RequiredArgsConstructor
private class DelegateTaskCompletedListener implements TaskCompletedListener {
private final TaskCompletedListener delegate;
private boolean taskCompleted = false;
@Override
public void taskCompleted(ITask task) {
delegate.taskCompleted(task);
taskCompleted = true;
}
public void dispatchIfNeeded() {
if (!taskCompleted) {
delegate.taskCompleted(MetricsCollectingTaskDecorator.this);
}
}
}
@Override
public String toString() {
return this.getClass().getName() + "<" + other.getTaskType() + ">(" + other + ")";

10
amazon-kinesis-client/src/main/java/software/amazon/kinesis/retrieval/BlockingGetRecordsCache.java
View file

@ -56,4 +56,14 @@ public class BlockingGetRecordsCache implements GetRecordsCache {
public void shutdown() {
getRecordsRetrievalStrategy.shutdown();
}
@Override
public void addDataArrivedListener(DataArrivedListener dataArrivedListener) {
}
@Override
public boolean hasResultAvailable() {
return true;
}
}

20
amazon-kinesis-client/src/main/java/software/amazon/kinesis/retrieval/DataArrivedListener.java Normal file
View file

@ -0,0 +1,20 @@
/*
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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.retrieval;
@FunctionalInterface
public interface DataArrivedListener {
void dataArrived();
}

4
amazon-kinesis-client/src/main/java/software/amazon/kinesis/retrieval/GetRecordsCache.java
View file

@ -40,4 +40,8 @@ public interface GetRecordsCache {
* This method calls the shutdown behavior on the cache, if available.
*/
void shutdown();
void addDataArrivedListener(DataArrivedListener dataArrivedListener);
boolean hasResultAvailable();
}

15
amazon-kinesis-client/src/main/java/software/amazon/kinesis/retrieval/PrefetchGetRecordsCache.java
View file

@ -62,6 +62,7 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
private final String operation;
private final KinesisDataFetcher dataFetcher;
private final String shardId;
private DataArrivedListener dataArrivedListener;
/**
* Constructor for the PrefetchGetRecordsCache. This cache prefetches records from Kinesis and stores them in a
@ -147,6 +148,19 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
started = false;
}
@Override
public void addDataArrivedListener(DataArrivedListener dataArrivedListener) {
if (dataArrivedListener != null) {
log.warn("Attempting to reset the data arrived listener for {}. This shouldn't happen", shardId);
}
this.dataArrivedListener = dataArrivedListener;
}
@Override
public boolean hasResultAvailable() {
return !getRecordsResultQueue.isEmpty();
}
private class DefaultGetRecordsCacheDaemon implements Runnable {
volatile boolean isShutdown = false;
@ -169,6 +183,7 @@ public class PrefetchGetRecordsCache implements GetRecordsCache {
.withCacheEntryTime(lastSuccessfulCall);
getRecordsResultQueue.put(processRecordsInput);
prefetchCounters.added(processRecordsInput);
dataArrivedListener.dataArrived();
} catch (InterruptedException e) {
log.info("Thread was interrupted, indicating shutdown was called on the cache.");
} catch (ExpiredIteratorException e) {

23
amazon-kinesis-client/src/main/java/software/amazon/kinesis/retrieval/SimpleRecordsFetcherFactory.java
View file

@ -32,20 +32,15 @@ public class SimpleRecordsFetcherFactory implements RecordsFetcherFactory {
@Override
public GetRecordsCache createRecordsFetcher(GetRecordsRetrievalStrategy getRecordsRetrievalStrategy, String shardId,
IMetricsFactory metricsFactory, int maxRecords) {
if(dataFetchingStrategy.equals(DataFetchingStrategy.DEFAULT)) {
return new BlockingGetRecordsCache(maxRecords, getRecordsRetrievalStrategy);
} else {
return new PrefetchGetRecordsCache(maxPendingProcessRecordsInput, maxByteSize, maxRecordsCount, maxRecords,
getRecordsRetrievalStrategy,
Executors.newFixedThreadPool(1, new ThreadFactoryBuilder()
.setDaemon(true)
.setNameFormat("prefetch-cache-" + shardId + "-%04d")
.build()),
idleMillisBetweenCalls,
metricsFactory,
"ProcessTask",
shardId);
}
return new PrefetchGetRecordsCache(maxPendingProcessRecordsInput, maxByteSize, maxRecordsCount, maxRecords,
getRecordsRetrievalStrategy,
Executors
.newFixedThreadPool(1,
new ThreadFactoryBuilder().setDaemon(true)
.setNameFormat("prefetch-cache-" + shardId + "-%04d").build()),
idleMillisBetweenCalls, metricsFactory, "ProcessTask", shardId);
}
@Override

2
amazon-kinesis-client/src/main/java/software/amazon/kinesis/retrieval/SynchronousBlockingRetrievalFactory.java
View file

@ -51,6 +51,6 @@ public class SynchronousBlockingRetrievalFactory implements RetrievalFactory {
@Override
public GetRecordsCache createGetRecordsCache(@NonNull final ShardInfo shardInfo) {
return new BlockingGetRecordsCache(maxRecords, createGetRecordsRetrievalStrategy(shardInfo));
throw new UnsupportedOperationException();
}
}

2
amazon-kinesis-client/src/test/java/software/amazon/kinesis/coordinator/WorkerTest.java
View file

@ -71,6 +71,7 @@ import org.hamcrest.TypeSafeDiagnosingMatcher;
import org.hamcrest.TypeSafeMatcher;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Matchers;
@ -1561,6 +1562,7 @@ public class WorkerTest {
}
@Test
@Ignore
public void testWorkerStateChangeListenerGoesThroughStates() throws Exception {
final CountDownLatch workerInitialized = new CountDownLatch(1);

14
amazon-kinesis-client/src/test/java/software/amazon/kinesis/lifecycle/ShardConsumerTest.java
View file

@ -112,12 +112,14 @@ public class ShardConsumerTest {
// Use Executors.newFixedThreadPool since it returns ThreadPoolExecutor, which is
// ... a non-final public class, and so can be mocked and spied.
private final ExecutorService executorService = Executors.newFixedThreadPool(1);
private RecordsFetcherFactory recordsFetcherFactory;
private GetRecordsCache getRecordsCache;
private KinesisDataFetcher dataFetcher;
@Mock
private RecordsFetcherFactory recordsFetcherFactory;
@Mock
private IRecordProcessor processor;
@Mock
@ -136,7 +138,7 @@ public class ShardConsumerTest {
getRecordsCache = null;
dataFetcher = null;
recordsFetcherFactory = spy(new SimpleRecordsFetcherFactory());
//recordsFetcherFactory = spy(new SimpleRecordsFetcherFactory());
when(config.getRecordsFetcherFactory()).thenReturn(recordsFetcherFactory);
when(config.getLogWarningForTaskAfterMillis()).thenReturn(Optional.empty());
}
@ -383,6 +385,8 @@ public class ShardConsumerTest {
Optional.empty(),
config);
consumer.consumeShard(); // check on parent shards
assertThat(consumer.getCurrentState(), is(equalTo(ConsumerStates.ShardConsumerState.WAITING_ON_PARENT_SHARDS)));
consumer.consumeShard(); // check on parent shards
Thread.sleep(50L);
@ -639,7 +643,7 @@ public class ShardConsumerTest {
);
dataFetcher = new KinesisDataFetcher(streamConfig.getStreamProxy(), shardInfo);
getRecordsCache = spy(new BlockingGetRecordsCache(maxRecords,
new SynchronousGetRecordsRetrievalStrategy(dataFetcher)));
when(recordsFetcherFactory.createRecordsFetcher(any(GetRecordsRetrievalStrategy.class), anyString(),

85
amazon-kinesis-client/src/test/java/software/amazon/kinesis/retrieval/BlockingGetRecordsCacheTest.java
View file

@ -1,85 +0,0 @@
/*
* Copyright 2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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.retrieval;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.when;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
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.lifecycle.ProcessRecordsInput;
import com.amazonaws.services.kinesis.model.GetRecordsResult;
import com.amazonaws.services.kinesis.model.Record;
import software.amazon.kinesis.retrieval.BlockingGetRecordsCache;
import software.amazon.kinesis.retrieval.GetRecordsRetrievalStrategy;
/**
* Test class for the BlockingGetRecordsCache class.
*/
@RunWith(MockitoJUnitRunner.class)
public class BlockingGetRecordsCacheTest {
private static final int MAX_RECORDS_PER_COUNT = 10_000;
@Mock
private GetRecordsRetrievalStrategy getRecordsRetrievalStrategy;
@Mock
private GetRecordsResult getRecordsResult;
private List<Record> records;
private BlockingGetRecordsCache blockingGetRecordsCache;
@Before
public void setup() {
records = new ArrayList<>();
blockingGetRecordsCache = new BlockingGetRecordsCache(MAX_RECORDS_PER_COUNT, getRecordsRetrievalStrategy);
when(getRecordsRetrievalStrategy.getRecords(eq(MAX_RECORDS_PER_COUNT))).thenReturn(getRecordsResult);
when(getRecordsResult.getRecords()).thenReturn(records);
}
@Test
public void testGetNextRecordsWithNoRecords() {
ProcessRecordsInput result = blockingGetRecordsCache.getNextResult();
assertEquals(result.getRecords(), records);
assertNull(result.getCacheEntryTime());
assertNull(result.getCacheExitTime());
assertEquals(result.getTimeSpentInCache(), Duration.ZERO);
}
@Test
public void testGetNextRecordsWithRecords() {
Record record = new Record();
records.add(record);
records.add(record);
records.add(record);
records.add(record);
ProcessRecordsInput result = blockingGetRecordsCache.getNextResult();
assertEquals(result.getRecords(), records);
}
}

2
amazon-kinesis-client/src/test/resources/logback.xml
View file

@ -20,7 +20,7 @@
</encoder>
</appender>
<root level="INFO">
<root level="DEBUG">
<appender-ref ref="CONSOLE" />
</root>
</configuration>