@Test public void testProcessTaskWithShardEndReached() { processTask = makeProcessTask(processRecordsInput); when(processRecordsInput.isAtShardEnd()).thenReturn(true); TaskResult result = processTask.call(); assertThat(result, shardEndTaskResult(true)); }
@Override public TaskResult call() { final MetricsScope scope = MetricsUtil.createMetricsWithOperation(metricsFactory, PROCESS_TASK_OPERATION); shardInfo.streamIdentifierSerOpt() .ifPresent(streamId -> MetricsUtil.addStreamId(scope, StreamIdentifier.multiStreamInstance(streamId))); MetricsUtil.addShardId(scope, shardInfo.shardId()); long startTimeMillis = System.currentTimeMillis(); boolean success = false; try { 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.millisBehindLatest() != null) { scope.addData(MILLIS_BEHIND_LATEST_METRIC, processRecordsInput.millisBehindLatest(), StandardUnit.MILLISECONDS, MetricsLevel.SUMMARY); } if (processRecordsInput.isAtShardEnd() && processRecordsInput.records().isEmpty()) { log.info("Reached end of shard {} and have no records to process", shardInfoId); return new TaskResult(null, true); } throttlingReporter.success(); List<KinesisClientRecord> records = deaggregateAnyKplRecords(processRecordsInput.records()); if (!records.isEmpty()) { scope.addData(RECORDS_PROCESSED_METRIC, records.size(), StandardUnit.COUNT, MetricsLevel.SUMMARY); } recordProcessorCheckpointer.largestPermittedCheckpointValue(filterAndGetMaxExtendedSequenceNumber( scope, records, recordProcessorCheckpointer.lastCheckpointValue(), recordProcessorCheckpointer.largestPermittedCheckpointValue())); if (shouldCallProcessRecords(records)) { callProcessRecords(processRecordsInput, records); } success = true; } catch (RuntimeException e) { log.error("ShardId {}: Caught exception: ", shardInfoId, e); exception = e; backoff(); } if (processRecordsInput.isAtShardEnd()) { log.info("Reached end of shard {}, and processed {} records", shardInfoId, processRecordsInput.records().size()); return new TaskResult(null, true); } return new TaskResult(exception); } finally { MetricsUtil.addSuccessAndLatency(scope, success, startTimeMillis, MetricsLevel.SUMMARY); MetricsUtil.endScope(scope); } }
ProcessTask implements ConsumerTask { @Override public TaskResult call() { final MetricsScope scope = MetricsUtil.createMetricsWithOperation(metricsFactory, PROCESS_TASK_OPERATION); shardInfo.streamIdentifierSerOpt() .ifPresent(streamId -> MetricsUtil.addStreamId(scope, StreamIdentifier.multiStreamInstance(streamId))); MetricsUtil.addShardId(scope, shardInfo.shardId()); long startTimeMillis = System.currentTimeMillis(); boolean success = false; try { 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.millisBehindLatest() != null) { scope.addData(MILLIS_BEHIND_LATEST_METRIC, processRecordsInput.millisBehindLatest(), StandardUnit.MILLISECONDS, MetricsLevel.SUMMARY); } if (processRecordsInput.isAtShardEnd() && processRecordsInput.records().isEmpty()) { log.info("Reached end of shard {} and have no records to process", shardInfoId); return new TaskResult(null, true); } throttlingReporter.success(); List<KinesisClientRecord> records = deaggregateAnyKplRecords(processRecordsInput.records()); if (!records.isEmpty()) { scope.addData(RECORDS_PROCESSED_METRIC, records.size(), StandardUnit.COUNT, MetricsLevel.SUMMARY); } recordProcessorCheckpointer.largestPermittedCheckpointValue(filterAndGetMaxExtendedSequenceNumber( scope, records, recordProcessorCheckpointer.lastCheckpointValue(), recordProcessorCheckpointer.largestPermittedCheckpointValue())); if (shouldCallProcessRecords(records)) { callProcessRecords(processRecordsInput, records); } success = true; } catch (RuntimeException e) { log.error("ShardId {}: Caught exception: ", shardInfoId, e); exception = e; backoff(); } if (processRecordsInput.isAtShardEnd()) { log.info("Reached end of shard {}, and processed {} records", shardInfoId, processRecordsInput.records().size()); return new TaskResult(null, true); } return new TaskResult(exception); } finally { MetricsUtil.addSuccessAndLatency(scope, success, startTimeMillis, MetricsLevel.SUMMARY); MetricsUtil.endScope(scope); } } }
ProcessTask implements ConsumerTask { @Override public TaskResult call() { final MetricsScope scope = MetricsUtil.createMetricsWithOperation(metricsFactory, PROCESS_TASK_OPERATION); shardInfo.streamIdentifierSerOpt() .ifPresent(streamId -> MetricsUtil.addStreamId(scope, StreamIdentifier.multiStreamInstance(streamId))); MetricsUtil.addShardId(scope, shardInfo.shardId()); long startTimeMillis = System.currentTimeMillis(); boolean success = false; try { 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.millisBehindLatest() != null) { scope.addData(MILLIS_BEHIND_LATEST_METRIC, processRecordsInput.millisBehindLatest(), StandardUnit.MILLISECONDS, MetricsLevel.SUMMARY); } if (processRecordsInput.isAtShardEnd() && processRecordsInput.records().isEmpty()) { log.info("Reached end of shard {} and have no records to process", shardInfoId); return new TaskResult(null, true); } throttlingReporter.success(); List<KinesisClientRecord> records = deaggregateAnyKplRecords(processRecordsInput.records()); if (!records.isEmpty()) { scope.addData(RECORDS_PROCESSED_METRIC, records.size(), StandardUnit.COUNT, MetricsLevel.SUMMARY); } recordProcessorCheckpointer.largestPermittedCheckpointValue(filterAndGetMaxExtendedSequenceNumber( scope, records, recordProcessorCheckpointer.lastCheckpointValue(), recordProcessorCheckpointer.largestPermittedCheckpointValue())); if (shouldCallProcessRecords(records)) { callProcessRecords(processRecordsInput, records); } success = true; } catch (RuntimeException e) { log.error("ShardId {}: Caught exception: ", shardInfoId, e); exception = e; backoff(); } if (processRecordsInput.isAtShardEnd()) { log.info("Reached end of shard {}, and processed {} records", shardInfoId, processRecordsInput.records().size()); return new TaskResult(null, true); } return new TaskResult(exception); } finally { MetricsUtil.addSuccessAndLatency(scope, success, startTimeMillis, MetricsLevel.SUMMARY); MetricsUtil.endScope(scope); } } ProcessTask(@NonNull ShardInfo shardInfo,
                       @NonNull ShardRecordProcessor shardRecordProcessor,
                       @NonNull ShardRecordProcessorCheckpointer recordProcessorCheckpointer,
                       long backoffTimeMillis,
                       boolean skipShardSyncAtWorkerInitializationIfLeasesExist,
                       ShardDetector shardDetector,
                       @NonNull ThrottlingReporter throttlingReporter,
                       ProcessRecordsInput processRecordsInput,
                       boolean shouldCallProcessRecordsEvenForEmptyRecordList,
                       long idleTimeInMilliseconds,
                       @NonNull AggregatorUtil aggregatorUtil,
                       @NonNull MetricsFactory metricsFactory); }
ProcessTask implements ConsumerTask { @Override public TaskResult call() { final MetricsScope scope = MetricsUtil.createMetricsWithOperation(metricsFactory, PROCESS_TASK_OPERATION); shardInfo.streamIdentifierSerOpt() .ifPresent(streamId -> MetricsUtil.addStreamId(scope, StreamIdentifier.multiStreamInstance(streamId))); MetricsUtil.addShardId(scope, shardInfo.shardId()); long startTimeMillis = System.currentTimeMillis(); boolean success = false; try { 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.millisBehindLatest() != null) { scope.addData(MILLIS_BEHIND_LATEST_METRIC, processRecordsInput.millisBehindLatest(), StandardUnit.MILLISECONDS, MetricsLevel.SUMMARY); } if (processRecordsInput.isAtShardEnd() && processRecordsInput.records().isEmpty()) { log.info("Reached end of shard {} and have no records to process", shardInfoId); return new TaskResult(null, true); } throttlingReporter.success(); List<KinesisClientRecord> records = deaggregateAnyKplRecords(processRecordsInput.records()); if (!records.isEmpty()) { scope.addData(RECORDS_PROCESSED_METRIC, records.size(), StandardUnit.COUNT, MetricsLevel.SUMMARY); } recordProcessorCheckpointer.largestPermittedCheckpointValue(filterAndGetMaxExtendedSequenceNumber( scope, records, recordProcessorCheckpointer.lastCheckpointValue(), recordProcessorCheckpointer.largestPermittedCheckpointValue())); if (shouldCallProcessRecords(records)) { callProcessRecords(processRecordsInput, records); } success = true; } catch (RuntimeException e) { log.error("ShardId {}: Caught exception: ", shardInfoId, e); exception = e; backoff(); } if (processRecordsInput.isAtShardEnd()) { log.info("Reached end of shard {}, and processed {} records", shardInfoId, processRecordsInput.records().size()); return new TaskResult(null, true); } return new TaskResult(exception); } finally { MetricsUtil.addSuccessAndLatency(scope, success, startTimeMillis, MetricsLevel.SUMMARY); MetricsUtil.endScope(scope); } } ProcessTask(@NonNull ShardInfo shardInfo,
                       @NonNull ShardRecordProcessor shardRecordProcessor,
                       @NonNull ShardRecordProcessorCheckpointer recordProcessorCheckpointer,
                       long backoffTimeMillis,
                       boolean skipShardSyncAtWorkerInitializationIfLeasesExist,
                       ShardDetector shardDetector,
                       @NonNull ThrottlingReporter throttlingReporter,
                       ProcessRecordsInput processRecordsInput,
                       boolean shouldCallProcessRecordsEvenForEmptyRecordList,
                       long idleTimeInMilliseconds,
                       @NonNull AggregatorUtil aggregatorUtil,
                       @NonNull MetricsFactory metricsFactory); @Override TaskResult call(); @Override TaskType taskType(); }
ProcessTask implements ConsumerTask { @Override public TaskResult call() { final MetricsScope scope = MetricsUtil.createMetricsWithOperation(metricsFactory, PROCESS_TASK_OPERATION); shardInfo.streamIdentifierSerOpt() .ifPresent(streamId -> MetricsUtil.addStreamId(scope, StreamIdentifier.multiStreamInstance(streamId))); MetricsUtil.addShardId(scope, shardInfo.shardId()); long startTimeMillis = System.currentTimeMillis(); boolean success = false; try { 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.millisBehindLatest() != null) { scope.addData(MILLIS_BEHIND_LATEST_METRIC, processRecordsInput.millisBehindLatest(), StandardUnit.MILLISECONDS, MetricsLevel.SUMMARY); } if (processRecordsInput.isAtShardEnd() && processRecordsInput.records().isEmpty()) { log.info("Reached end of shard {} and have no records to process", shardInfoId); return new TaskResult(null, true); } throttlingReporter.success(); List<KinesisClientRecord> records = deaggregateAnyKplRecords(processRecordsInput.records()); if (!records.isEmpty()) { scope.addData(RECORDS_PROCESSED_METRIC, records.size(), StandardUnit.COUNT, MetricsLevel.SUMMARY); } recordProcessorCheckpointer.largestPermittedCheckpointValue(filterAndGetMaxExtendedSequenceNumber( scope, records, recordProcessorCheckpointer.lastCheckpointValue(), recordProcessorCheckpointer.largestPermittedCheckpointValue())); if (shouldCallProcessRecords(records)) { callProcessRecords(processRecordsInput, records); } success = true; } catch (RuntimeException e) { log.error("ShardId {}: Caught exception: ", shardInfoId, e); exception = e; backoff(); } if (processRecordsInput.isAtShardEnd()) { log.info("Reached end of shard {}, and processed {} records", shardInfoId, processRecordsInput.records().size()); return new TaskResult(null, true); } return new TaskResult(exception); } finally { MetricsUtil.addSuccessAndLatency(scope, success, startTimeMillis, MetricsLevel.SUMMARY); MetricsUtil.endScope(scope); } } ProcessTask(@NonNull ShardInfo shardInfo,
                       @NonNull ShardRecordProcessor shardRecordProcessor,
                       @NonNull ShardRecordProcessorCheckpointer recordProcessorCheckpointer,
                       long backoffTimeMillis,
                       boolean skipShardSyncAtWorkerInitializationIfLeasesExist,
                       ShardDetector shardDetector,
                       @NonNull ThrottlingReporter throttlingReporter,
                       ProcessRecordsInput processRecordsInput,
                       boolean shouldCallProcessRecordsEvenForEmptyRecordList,
                       long idleTimeInMilliseconds,
                       @NonNull AggregatorUtil aggregatorUtil,
                       @NonNull MetricsFactory metricsFactory); @Override TaskResult call(); @Override TaskType taskType();  }