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`

Author: Min3953

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(

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 {