Rename kinesis metrics config to kcl_metrics_enabled and default to true

Signed-off-by: RashmiRam <ras.xena@gmail.com>

Author: RashmiRam

data-prepper-plugins/kinesis-source/src/main/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisService.java

@@ -200,7 +200,7 @@ public Scheduler createScheduler(final Buffer<Record<Event>> buffer) {
                             kinesisSourceConfig.getPollingConfig().getIdleTimeBetweenReads().toMillis()));
         }
 
-        MetricsConfig metricsConfig = kinesisSourceConfig.isMetrics()
+        MetricsConfig metricsConfig = kinesisSourceConfig.isKclMetricsEnabled()
                 ? configsBuilder.metricsConfig()
                 : configsBuilder.metricsConfig().metricsFactory(new NullMetricsFactory());
 

data-prepper-plugins/kinesis-source/src/main/java/org/opensearch/dataprepper/plugins/kinesis/source/configuration/KinesisSourceConfig.java

@@ -81,8 +81,8 @@ public Duration getShardAcknowledgmentTimeout() {
     private Duration initializationBackoffTime = DEFAULT_INITIALIZATION_BACKOFF_TIME;
 
     @Getter
-    @JsonProperty("metrics")
-    private boolean metrics = false;
+    @JsonProperty("kcl_metrics_enabled")
+    private boolean kclMetricsEnabled = true;
 }
 
 

data-prepper-plugins/kinesis-source/src/test/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisServiceTest.java

@@ -262,7 +262,7 @@ void testCreateScheduler() {
 
     @Test
     void testCreateSchedulerUsesNullMetricsFactoryWhenMetricsDisabled() {
-        when(kinesisSourceConfig.isMetrics()).thenReturn(false);
+        when(kinesisSourceConfig.isKclMetricsEnabled()).thenReturn(false);
         KinesisService kinesisService = new KinesisService(kinesisSourceConfig, kinesisClientFactory, pluginMetrics, pluginFactory,
                 pipelineDescription, acknowledgementSetManager, kinesisLeaseConfigSupplier, workerIdentifierGenerator);
         Scheduler schedulerObjectUnderTest = kinesisService.createScheduler(buffer);
@@ -274,7 +274,7 @@ void testCreateSchedulerUsesNullMetricsFactoryWhenMetricsDisabled() {
 
     @Test
     void testCreateSchedulerUsesCloudWatchMetricsFactoryWhenMetricsEnabled() {
-        when(kinesisSourceConfig.isMetrics()).thenReturn(true);
+        when(kinesisSourceConfig.isKclMetricsEnabled()).thenReturn(true);
         KinesisService kinesisService = new KinesisService(kinesisSourceConfig, kinesisClientFactory, pluginMetrics, pluginFactory,
                 pipelineDescription, acknowledgementSetManager, kinesisLeaseConfigSupplier, workerIdentifierGenerator);
         Scheduler schedulerObjectUnderTest = kinesisService.createScheduler(buffer);

data-prepper-plugins/kinesis-source/src/test/java/org/opensearch/dataprepper/plugins/kinesis/source/configuration/KinesisSourceConfigTest.java

@@ -82,7 +82,7 @@ void testSourceConfig() {
         assertEquals(KinesisSourceConfig.DEFAULT_MAX_INITIALIZATION_ATTEMPTS, kinesisSourceConfig.getMaxInitializationAttempts());
         assertEquals(KinesisSourceConfig.DEFAULT_INITIALIZATION_BACKOFF_TIME, kinesisSourceConfig.getInitializationBackoffTime());
         assertTrue(kinesisSourceConfig.isAcknowledgments());
-        assertFalse(kinesisSourceConfig.isMetrics());
+        assertTrue(kinesisSourceConfig.isKclMetricsEnabled());
         assertEquals(KinesisSourceConfig.DEFAULT_SHARD_ACKNOWLEDGEMENT_TIMEOUT, kinesisSourceConfig.getShardAcknowledgmentTimeout());
         assertThat(kinesisSourceConfig.getAwsAuthenticationConfig(), notNullValue());
         assertEquals(kinesisSourceConfig.getAwsAuthenticationConfig().getAwsRegion(), Region.US_EAST_1);
@@ -116,7 +116,7 @@ void testSourceConfigWithStreamCodec() {
         assertEquals(KinesisSourceConfig.DEFAULT_MAX_INITIALIZATION_ATTEMPTS, kinesisSourceConfig.getMaxInitializationAttempts());
         assertEquals(KinesisSourceConfig.DEFAULT_INITIALIZATION_BACKOFF_TIME, kinesisSourceConfig.getInitializationBackoffTime());
         assertFalse(kinesisSourceConfig.isAcknowledgments());
-        assertFalse(kinesisSourceConfig.isMetrics());
+        assertTrue(kinesisSourceConfig.isKclMetricsEnabled());
         assertEquals(KinesisSourceConfig.DEFAULT_SHARD_ACKNOWLEDGEMENT_TIMEOUT, kinesisSourceConfig.getShardAcknowledgmentTimeout());
         assertThat(kinesisSourceConfig.getAwsAuthenticationConfig(), notNullValue());
         assertEquals(kinesisSourceConfig.getAwsAuthenticationConfig().getAwsRegion(), Region.US_EAST_1);
@@ -243,7 +243,7 @@ void testSourceConfigWithStreamArnConsumerArn() {
     void testSourceConfigWithMetricsEnabled() {
 
         assertThat(kinesisSourceConfig, notNullValue());
-        assertTrue(kinesisSourceConfig.isMetrics());
+        assertTrue(kinesisSourceConfig.isKclMetricsEnabled());
         assertEquals(KinesisSourceConfig.DEFAULT_NUMBER_OF_RECORDS_TO_ACCUMULATE, kinesisSourceConfig.getNumberOfRecordsToAccumulate());
         assertEquals(KinesisSourceConfig.DEFAULT_TIME_OUT_IN_MILLIS, kinesisSourceConfig.getBufferTimeout());
         assertFalse(kinesisSourceConfig.isAcknowledgments());

data-prepper-plugins/kinesis-source/src/test/resources/pipeline_with_metrics_enabled.yaml

@@ -9,5 +9,5 @@ source:
     aws:
       sts_role_arn: "arn:aws:iam::123456789012:role/OSI-PipelineRole"
       region: "us-east-1"
-    metrics: true
+    kcl_metrics_enabled: true