spring-projectsGH-241: Support changing metrics-level for KclMessageDrivenChannelAdapter

Author: 문민규

src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java

@@ -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;
@@ -145,6 +147,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 +271,15 @@ public void setFanOut(boolean fanOut) {
 		this.fanOut = fanOut;
 	}
 
+    /**
+     * Specify a metrics level to emit (DETAILED; default).
+     * @param metricsLevel {@link MetricsLevel} DETAILED for emitting all metrics.
+	 * @author Minkyu Moon
+     */
+	public void setMetricsLevel(MetricsLevel metricsLevel) {
+		this.metricsLevel = metricsLevel;
+	}
+
 	@Override
 	protected void onInit() {
 		super.onInit();
@@ -321,13 +334,18 @@ protected void doStart() {
 				.glueSchemaRegistryDeserializer(this.glueSchemaRegistryDeserializer)
 				.retrievalSpecificConfig(retrievalSpecificConfig);
 
+		MetricsConfig metricsConfig = this.config.metricsConfig();
+		if(this.metricsLevel != null) {
+			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);