diff --git a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/RecordProcessorCheckpointer.java b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/RecordProcessorCheckpointer.java index 72e18d73..8e3dfd73 100644 --- a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/RecordProcessorCheckpointer.java +++ b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/RecordProcessorCheckpointer.java @@ -14,6 +14,9 @@ */ package com.amazonaws.services.kinesis.clientlibrary.lib.worker; +import com.amazonaws.services.kinesis.metrics.impl.MetricsHelper; +import com.amazonaws.services.kinesis.metrics.impl.ThreadSafeMetricsDelegatingScope; +import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsFactory; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -50,6 +53,8 @@ class RecordProcessorCheckpointer implements IRecordProcessorCheckpointer { private SequenceNumberValidator sequenceNumberValidator; private ExtendedSequenceNumber sequenceNumberAtShardEnd; + + private IMetricsFactory metricsFactory; /** * Only has package level access, since only the Amazon Kinesis Client Library should be creating these. @@ -59,10 +64,12 @@ class RecordProcessorCheckpointer implements IRecordProcessorCheckpointer { */ RecordProcessorCheckpointer(ShardInfo shardInfo, ICheckpoint checkpoint, - SequenceNumberValidator validator) { + SequenceNumberValidator validator, + IMetricsFactory metricsFactory) { this.shardInfo = shardInfo; this.checkpoint = checkpoint; this.sequenceNumberValidator = validator; + this.metricsFactory = metricsFactory; } /** @@ -283,21 +290,33 @@ class RecordProcessorCheckpointer implements IRecordProcessorCheckpointer { // just checkpoint at SHARD_END checkpointToRecord = ExtendedSequenceNumber.SHARD_END; } + + boolean unsetMetrics = false; // Don't checkpoint a value we already successfully checkpointed - if (extendedSequenceNumber != null && !extendedSequenceNumber.equals(lastCheckpointValue)) { - try { - if (LOG.isDebugEnabled()) { - LOG.debug("Setting " + shardInfo.getShardId() + ", token " + shardInfo.getConcurrencyToken() - + " checkpoint to " + checkpointToRecord); + try { + if (!MetricsHelper.isMetricsScopePresent()) { + MetricsHelper.setMetricsScope(new ThreadSafeMetricsDelegatingScope(metricsFactory.createMetrics())); + unsetMetrics = true; + } + if (extendedSequenceNumber != null && !extendedSequenceNumber.equals(lastCheckpointValue)) { + try { + if (LOG.isDebugEnabled()) { + LOG.debug("Setting " + shardInfo.getShardId() + ", token " + shardInfo.getConcurrencyToken() + + " checkpoint to " + checkpointToRecord); + } + checkpoint.setCheckpoint(shardInfo.getShardId(), checkpointToRecord, shardInfo.getConcurrencyToken()); + lastCheckpointValue = checkpointToRecord; + } catch (ThrottlingException | ShutdownException | InvalidStateException + | KinesisClientLibDependencyException e) { + throw e; + } catch (KinesisClientLibException e) { + LOG.warn("Caught exception setting checkpoint.", e); + throw new KinesisClientLibDependencyException("Caught exception while checkpointing", e); } - checkpoint.setCheckpoint(shardInfo.getShardId(), checkpointToRecord, shardInfo.getConcurrencyToken()); - lastCheckpointValue = checkpointToRecord; - } catch (ThrottlingException | ShutdownException | InvalidStateException - | KinesisClientLibDependencyException e) { - throw e; - } catch (KinesisClientLibException e) { - LOG.warn("Caught exception setting checkpoint.", e); - throw new KinesisClientLibDependencyException("Caught exception while checkpointing", e); + } + } finally { + if (unsetMetrics) { + MetricsHelper.unsetMetricsScope(); } } } diff --git a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumer.java b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumer.java index 95cc663e..4a001b9b 100644 --- a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumer.java +++ b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumer.java @@ -170,7 +170,8 @@ class ShardConsumer { new SequenceNumberValidator( streamConfig.getStreamProxy(), shardInfo.getShardId(), - streamConfig.shouldValidateSequenceNumberBeforeCheckpointing())), + streamConfig.shouldValidateSequenceNumberBeforeCheckpointing()), + metricsFactory), leaseManager, parentShardPollIntervalMillis, cleanupLeasesOfCompletedShards, diff --git a/src/main/java/com/amazonaws/services/kinesis/metrics/impl/MetricsHelper.java b/src/main/java/com/amazonaws/services/kinesis/metrics/impl/MetricsHelper.java index 4599fbaa..bf104cff 100644 --- a/src/main/java/com/amazonaws/services/kinesis/metrics/impl/MetricsHelper.java +++ b/src/main/java/com/amazonaws/services/kinesis/metrics/impl/MetricsHelper.java @@ -72,13 +72,22 @@ public class MetricsHelper { * @param scope */ public static void setMetricsScope(IMetricsScope scope) { - if (currentScope.get() != null) { + if (isMetricsScopePresent()) { throw new RuntimeException(String.format( "Metrics scope is already set for the current thread %s", Thread.currentThread().getName())); } currentScope.set(scope); } + /** + * Checks if current metricsscope is present or not. + * + * @return true if metrics scope is present, else returns false + */ + public static boolean isMetricsScopePresent() { + return currentScope.get() != null; + } + /** * Unsets the metrics scope for the current thread. */ diff --git a/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/RecordProcessorCheckpointerTest.java b/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/RecordProcessorCheckpointerTest.java index 31a1e184..4fa6292f 100644 --- a/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/RecordProcessorCheckpointerTest.java +++ b/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/RecordProcessorCheckpointerTest.java @@ -19,10 +19,13 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map.Entry; +import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsFactory; import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; import org.mockito.Mockito; import com.amazonaws.services.kinesis.clientlibrary.interfaces.ICheckpoint; @@ -32,6 +35,7 @@ import com.amazonaws.services.kinesis.clientlibrary.lib.checkpoint.SentinelCheck import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber; import com.amazonaws.services.kinesis.clientlibrary.types.UserRecord; import com.amazonaws.services.kinesis.model.Record; +import org.mockito.runners.MockitoJUnitRunner; import static org.junit.Assert.fail; import static org.mockito.Mockito.mock; @@ -40,6 +44,7 @@ import static org.mockito.Matchers.anyString; /** * */ +@RunWith(MockitoJUnitRunner.class) public class RecordProcessorCheckpointerTest { private String startingSequenceNumber = "13"; private ExtendedSequenceNumber startingExtendedSequenceNumber = new ExtendedSequenceNumber(startingSequenceNumber); @@ -48,6 +53,9 @@ public class RecordProcessorCheckpointerTest { private ShardInfo shardInfo; private SequenceNumberValidator sequenceNumberValidator; private String shardId = "shardId-123"; + + @Mock + IMetricsFactory metricsFactory; /** * @throws java.lang.Exception @@ -78,7 +86,7 @@ public class RecordProcessorCheckpointerTest { public final void testCheckpoint() throws Exception { // First call to checkpoint RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, null); + new RecordProcessorCheckpointer(shardInfo, checkpoint, null, metricsFactory); processingCheckpointer.setLargestPermittedCheckpointValue(startingExtendedSequenceNumber); processingCheckpointer.checkpoint(); Assert.assertEquals(startingExtendedSequenceNumber, checkpoint.getCheckpoint(shardId)); @@ -98,7 +106,7 @@ public class RecordProcessorCheckpointerTest { @Test public final void testCheckpointRecord() throws Exception { RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator, metricsFactory); processingCheckpointer.setInitialCheckpointValue(startingExtendedSequenceNumber); ExtendedSequenceNumber extendedSequenceNumber = new ExtendedSequenceNumber("5025"); Record record = new Record().withSequenceNumber("5025"); @@ -114,7 +122,7 @@ public class RecordProcessorCheckpointerTest { @Test public final void testCheckpointSubRecord() throws Exception { RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator, metricsFactory); processingCheckpointer.setInitialCheckpointValue(startingExtendedSequenceNumber); ExtendedSequenceNumber extendedSequenceNumber = new ExtendedSequenceNumber("5030"); Record record = new Record().withSequenceNumber("5030"); @@ -131,7 +139,7 @@ public class RecordProcessorCheckpointerTest { @Test public final void testCheckpointSequenceNumber() throws Exception { RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator, metricsFactory); processingCheckpointer.setInitialCheckpointValue(startingExtendedSequenceNumber); ExtendedSequenceNumber extendedSequenceNumber = new ExtendedSequenceNumber("5035"); processingCheckpointer.setLargestPermittedCheckpointValue(extendedSequenceNumber); @@ -146,7 +154,7 @@ public class RecordProcessorCheckpointerTest { @Test public final void testCheckpointExtendedSequenceNumber() throws Exception { RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator, metricsFactory); processingCheckpointer.setInitialCheckpointValue(startingExtendedSequenceNumber); ExtendedSequenceNumber extendedSequenceNumber = new ExtendedSequenceNumber("5040"); processingCheckpointer.setLargestPermittedCheckpointValue(extendedSequenceNumber); @@ -162,7 +170,7 @@ public class RecordProcessorCheckpointerTest { public final void testPrepareCheckpoint() throws Exception { // First call to checkpoint RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator, metricsFactory); processingCheckpointer.setInitialCheckpointValue(startingExtendedSequenceNumber); ExtendedSequenceNumber sequenceNumber1 = new ExtendedSequenceNumber("5001"); @@ -193,7 +201,7 @@ public class RecordProcessorCheckpointerTest { @Test public final void testPrepareCheckpointRecord() throws Exception { RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator, metricsFactory); processingCheckpointer.setInitialCheckpointValue(startingExtendedSequenceNumber); ExtendedSequenceNumber extendedSequenceNumber = new ExtendedSequenceNumber("5025"); Record record = new Record().withSequenceNumber("5025"); @@ -218,7 +226,7 @@ public class RecordProcessorCheckpointerTest { @Test public final void testPrepareCheckpointSubRecord() throws Exception { RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator, metricsFactory); processingCheckpointer.setInitialCheckpointValue(startingExtendedSequenceNumber); ExtendedSequenceNumber extendedSequenceNumber = new ExtendedSequenceNumber("5030"); Record record = new Record().withSequenceNumber("5030"); @@ -244,7 +252,7 @@ public class RecordProcessorCheckpointerTest { @Test public final void testPrepareCheckpointSequenceNumber() throws Exception { RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator, metricsFactory); processingCheckpointer.setInitialCheckpointValue(startingExtendedSequenceNumber); ExtendedSequenceNumber extendedSequenceNumber = new ExtendedSequenceNumber("5035"); processingCheckpointer.setLargestPermittedCheckpointValue(extendedSequenceNumber); @@ -268,7 +276,7 @@ public class RecordProcessorCheckpointerTest { @Test public final void testPrepareCheckpointExtendedSequenceNumber() throws Exception { RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator, metricsFactory); processingCheckpointer.setInitialCheckpointValue(startingExtendedSequenceNumber); ExtendedSequenceNumber extendedSequenceNumber = new ExtendedSequenceNumber("5040"); processingCheckpointer.setLargestPermittedCheckpointValue(extendedSequenceNumber); @@ -291,7 +299,7 @@ public class RecordProcessorCheckpointerTest { @Test public final void testMultipleOutstandingCheckpointersHappyCase() throws Exception { RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator, metricsFactory); processingCheckpointer.setInitialCheckpointValue(startingExtendedSequenceNumber); processingCheckpointer.setLargestPermittedCheckpointValue(new ExtendedSequenceNumber("6040")); @@ -323,7 +331,7 @@ public class RecordProcessorCheckpointerTest { @Test public final void testMultipleOutstandingCheckpointersOutOfOrder() throws Exception { RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, sequenceNumberValidator, metricsFactory); processingCheckpointer.setInitialCheckpointValue(startingExtendedSequenceNumber); processingCheckpointer.setLargestPermittedCheckpointValue(new ExtendedSequenceNumber("7040")); @@ -358,7 +366,7 @@ public class RecordProcessorCheckpointerTest { */ @Test public final void testUpdate() throws Exception { - RecordProcessorCheckpointer checkpointer = new RecordProcessorCheckpointer(shardInfo, checkpoint, null); + RecordProcessorCheckpointer checkpointer = new RecordProcessorCheckpointer(shardInfo, checkpoint, null, metricsFactory); ExtendedSequenceNumber sequenceNumber = new ExtendedSequenceNumber("10"); checkpointer.setLargestPermittedCheckpointValue(sequenceNumber); @@ -379,7 +387,7 @@ public class RecordProcessorCheckpointerTest { SequenceNumberValidator validator = mock(SequenceNumberValidator.class); Mockito.doNothing().when(validator).validateSequenceNumber(anyString()); RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, validator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, validator, metricsFactory); // Several checkpoints we're gonna hit ExtendedSequenceNumber tooSmall = new ExtendedSequenceNumber("2"); @@ -467,7 +475,7 @@ public class RecordProcessorCheckpointerTest { SequenceNumberValidator validator = mock(SequenceNumberValidator.class); Mockito.doNothing().when(validator).validateSequenceNumber(anyString()); RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, validator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, validator, metricsFactory); // Several checkpoints we're gonna hit ExtendedSequenceNumber tooSmall = new ExtendedSequenceNumber("2"); @@ -595,7 +603,7 @@ public class RecordProcessorCheckpointerTest { for (LinkedHashMap testPlan : getMixedCallsTestPlan()) { RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, validator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, validator, metricsFactory); testMixedCheckpointCalls(processingCheckpointer, testPlan, CheckpointerType.CHECKPOINTER); } } @@ -615,7 +623,7 @@ public class RecordProcessorCheckpointerTest { for (LinkedHashMap testPlan : getMixedCallsTestPlan()) { RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, validator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, validator, metricsFactory); testMixedCheckpointCalls(processingCheckpointer, testPlan, CheckpointerType.PREPARED_CHECKPOINTER); } } @@ -636,7 +644,7 @@ public class RecordProcessorCheckpointerTest { for (LinkedHashMap testPlan : getMixedCallsTestPlan()) { RecordProcessorCheckpointer processingCheckpointer = - new RecordProcessorCheckpointer(shardInfo, checkpoint, validator); + new RecordProcessorCheckpointer(shardInfo, checkpoint, validator, metricsFactory); testMixedCheckpointCalls(processingCheckpointer, testPlan, CheckpointerType.PREPARE_THEN_CHECKPOINTER); } } diff --git a/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumerTest.java b/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumerTest.java index 8a91c6e6..216d59cd 100644 --- a/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumerTest.java +++ b/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumerTest.java @@ -342,7 +342,8 @@ public class ShardConsumerTest { streamConfig.getStreamProxy(), shardInfo.getShardId(), streamConfig.shouldValidateSequenceNumberBeforeCheckpointing() - ) + ), + metricsFactory ); dataFetcher = new KinesisDataFetcher(streamConfig.getStreamProxy(), shardInfo); @@ -493,7 +494,8 @@ public class ShardConsumerTest { streamConfig.getStreamProxy(), shardInfo.getShardId(), streamConfig.shouldValidateSequenceNumberBeforeCheckpointing() - ) + ), + metricsFactory ); ShardConsumer consumer = @@ -621,7 +623,8 @@ public class ShardConsumerTest { streamConfig.getStreamProxy(), shardInfo.getShardId(), streamConfig.shouldValidateSequenceNumberBeforeCheckpointing() - ) + ), + metricsFactory ); dataFetcher = new KinesisDataFetcher(streamConfig.getStreamProxy(), shardInfo);