GH-241: Add metrics-level option to KclMessageDrivenChannelAdapter

Fixes: #241

* Add test that `metricsLevel` is set correctly

Author: Min3953

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.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 @@
  * @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 void setFanOut(boolean fanOut) {
 		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 @@ protected void doStart() {
 				.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);
 

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.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.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 @@
 /**
  * @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 @@ void kclChannelAdapterReceivesRecords() {
 		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 KclMessageDrivenChannelAdapter kclMessageDrivenChannelAdapter() {
 			adapter.setConverter(String::new);
 			adapter.setConsumerGroup("single_stream_group");
 			adapter.setFanOut(false);
+			adapter.setMetricsLevel(MetricsLevel.NONE);
 			adapter.setBindSourceRecord(true);
 			return adapter;
 		}