From 67038137869f15a5a851213da208b5e607dc98e7 Mon Sep 17 00:00:00 2001
From: "Minkyu Moon (Manggo)" <32037034+Min3953@users.noreply.github.com>
Date: Mon, 8 Apr 2024 22:43:44 +0900
Subject: [PATCH] KCL: metricsFactory to NullMetricsFactory if metrics level is
 NONE

* GH-241: Support changing metrics-level for `KclMessageDrivenChannelAdapter`

* Add test that `metricsLevel` is set correctly

* Modify test body and add author

* Add reference to instance variable

* Set `metricsFactory` to `NullMetricsFactory` if metrics level is `NONE`
---
 .../kinesis/KclMessageDrivenChannelAdapter.java  |  4 ++++
 .../KclMessageDrivenChannelAdapterTests.java     | 16 ++++++++++++++--
 2 files changed, 18 insertions(+), 2 deletions(-)

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.ShardEndedInput;
 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 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport
 
 		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.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.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 @@ public class KclMessageDrivenChannelAdapterTests implements LocalstackContainerT
 	}
 
 	@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 {