diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java index f7bbff9..f138a5f 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java @@ -52,6 +52,7 @@ import software.amazon.kinesis.lifecycle.events.ShutdownRequestedInput; import software.amazon.kinesis.metrics.MetricsConfig; import software.amazon.kinesis.metrics.MetricsLevel; +import software.amazon.kinesis.metrics.NullMetricsFactory; import software.amazon.kinesis.processor.FormerStreamsLeasesDeletionStrategy; import software.amazon.kinesis.processor.MultiStreamTracker; import software.amazon.kinesis.processor.RecordProcessorCheckpointer; @@ -338,6 +339,9 @@ protected void doStart() { MetricsConfig metricsConfig = this.config.metricsConfig(); metricsConfig.metricsLevel(this.metricsLevel); + if (MetricsLevel.NONE.equals(this.metricsLevel)) { + metricsConfig.metricsFactory(new NullMetricsFactory()); + } this.scheduler = new Scheduler( diff --git a/src/test/java/org/springframework/integration/aws/kinesis/KclMessageDrivenChannelAdapterTests.java b/src/test/java/org/springframework/integration/aws/kinesis/KclMessageDrivenChannelAdapterTests.java index 6f13c0a..db88b5d 100644 --- a/src/test/java/org/springframework/integration/aws/kinesis/KclMessageDrivenChannelAdapterTests.java +++ b/src/test/java/org/springframework/integration/aws/kinesis/KclMessageDrivenChannelAdapterTests.java @@ -28,7 +28,9 @@ import software.amazon.awssdk.services.kinesis.model.Consumer; import software.amazon.kinesis.common.InitialPositionInStream; import software.amazon.kinesis.common.InitialPositionInStreamExtended; +import software.amazon.kinesis.metrics.MetricsFactory; import software.amazon.kinesis.metrics.MetricsLevel; +import software.amazon.kinesis.metrics.NullMetricsFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; @@ -123,15 +125,25 @@ void kclChannelAdapterReceivesRecords() { } @Test - public void metricsLevelOfMetricsFactoryShouldBeSetToMetricsLevelOfAdapter() { + public void metricsLevelOfMetricsConfigShouldBeSetToMetricsLevelOfAdapter() { MetricsLevel metricsLevel = TestUtils.getPropertyValue( this.kclMessageDrivenChannelAdapter, - "scheduler.metricsFactory.metricsLevel", + "scheduler.metricsConfig.metricsLevel", MetricsLevel.class ); assertThat(metricsLevel).isEqualTo(MetricsLevel.NONE); } + @Test + public void metricsFactoryOfSchedulerShouldBeSetNullMetricsFactoryIfMetricsLevelIsNone() { + MetricsFactory metricsFactory = TestUtils.getPropertyValue( + this.kclMessageDrivenChannelAdapter, + "scheduler.metricsFactory", + MetricsFactory.class + ); + assertThat(metricsFactory).isInstanceOf(NullMetricsFactory.class); + } + @Configuration @EnableIntegration public static class TestConfiguration {