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 a1479cd..f7bbff9 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 @@ -1,5 +1,5 @@ /* - * Copyright 2019-2023 the original author or authors. + * Copyright 2019-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -50,6 +50,8 @@ import software.amazon.kinesis.lifecycle.events.LeaseLostInput; import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput; import software.amazon.kinesis.lifecycle.events.ShardEndedInput; import software.amazon.kinesis.lifecycle.events.ShutdownRequestedInput; +import software.amazon.kinesis.metrics.MetricsConfig; +import software.amazon.kinesis.metrics.MetricsLevel; import software.amazon.kinesis.processor.FormerStreamsLeasesDeletionStrategy; import software.amazon.kinesis.processor.MultiStreamTracker; import software.amazon.kinesis.processor.RecordProcessorCheckpointer; @@ -92,6 +94,7 @@ import org.springframework.util.Assert; * @author Artem Bilan * @author Dirk Bonhomme * @author Siddharth Jain + * @author Minkyu Moon * * @since 2.2.0 */ @@ -145,6 +148,8 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport private volatile Scheduler scheduler; + private MetricsLevel metricsLevel = MetricsLevel.DETAILED; + public KclMessageDrivenChannelAdapter(String... streams) { this(KinesisAsyncClient.create(), CloudWatchAsyncClient.create(), DynamoDbAsyncClient.create(), streams); } @@ -267,6 +272,16 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport this.fanOut = fanOut; } + /** + * Specify a metrics level to emit. + * Defaults to {@link MetricsLevel#DETAILED}. + * @param metricsLevel the {@link MetricsLevel} for emitting (or not) metrics into Cloud Watch. + */ + public void setMetricsLevel(MetricsLevel metricsLevel) { + Assert.notNull(metricsLevel, "'metricsLevel' must not be null"); + this.metricsLevel = metricsLevel; + } + @Override protected void onInit() { super.onInit(); @@ -321,13 +336,16 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport .glueSchemaRegistryDeserializer(this.glueSchemaRegistryDeserializer) .retrievalSpecificConfig(retrievalSpecificConfig); + MetricsConfig metricsConfig = this.config.metricsConfig(); + metricsConfig.metricsLevel(this.metricsLevel); + this.scheduler = new Scheduler( this.config.checkpointConfig(), this.config.coordinatorConfig(), this.config.leaseManagementConfig(), lifecycleConfig, - this.config.metricsConfig(), + metricsConfig, this.config.processorConfig(), retrievalConfig); 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 65f498c..6f13c0a 100644 --- a/src/test/java/org/springframework/integration/aws/kinesis/KclMessageDrivenChannelAdapterTests.java +++ b/src/test/java/org/springframework/integration/aws/kinesis/KclMessageDrivenChannelAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2023 the original author or authors. + * Copyright 2023-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -28,6 +28,7 @@ import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; 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.MetricsLevel; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; @@ -38,6 +39,7 @@ import org.springframework.integration.aws.inbound.kinesis.KclMessageDrivenChann import org.springframework.integration.aws.support.AwsHeaders; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; import org.springframework.test.annotation.DirtiesContext; @@ -48,6 +50,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Artem Bilan * @author Siddharth Jain + * @author Minkyu Moon * * @since 3.0 */ @@ -66,6 +69,9 @@ public class KclMessageDrivenChannelAdapterTests implements LocalstackContainerT @Autowired private PollableChannel kinesisReceiveChannel; + @Autowired + private KclMessageDrivenChannelAdapter kclMessageDrivenChannelAdapter; + @BeforeAll static void setup() { AMAZON_KINESIS = LocalstackContainerTest.kinesisClient(); @@ -116,6 +122,16 @@ public class KclMessageDrivenChannelAdapterTests implements LocalstackContainerT assertThat(streamConsumers).hasSize(0); } + @Test + public void metricsLevelOfMetricsFactoryShouldBeSetToMetricsLevelOfAdapter() { + MetricsLevel metricsLevel = TestUtils.getPropertyValue( + this.kclMessageDrivenChannelAdapter, + "scheduler.metricsFactory.metricsLevel", + MetricsLevel.class + ); + assertThat(metricsLevel).isEqualTo(MetricsLevel.NONE); + } + @Configuration @EnableIntegration public static class TestConfiguration { @@ -130,6 +146,7 @@ public class KclMessageDrivenChannelAdapterTests implements LocalstackContainerT adapter.setConverter(String::new); adapter.setConsumerGroup("single_stream_group"); adapter.setFanOut(false); + adapter.setMetricsLevel(MetricsLevel.NONE); adapter.setBindSourceRecord(true); return adapter; }