Removed unused configs

Signed-off-by: Krishna Kondaka <[email protected]>
kkondaka · Aug 16, 2023 · 314874b · 314874b
1 parent eb2af77
commit 314874b
Show file tree

Hide file tree

Showing 5 changed files with 15 additions and 83 deletions.
diff --git a/...gins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/AuthConfig.java b/...gins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/AuthConfig.java
@@ -54,25 +54,29 @@ public boolean hasOnlyOneConfig() {
     }
 
 
-    public static class SslAuthConfig {
-        // TODO Add Support for SSL authentication types like
-        // one-way or two-way authentication
+    /*
+     * TODO 
+        public static class SslAuthConfig {
+            // TODO Add Support for SSL authentication types like
+            // one-way or two-way authentication
 
-        public SslAuthConfig() {
+            public SslAuthConfig() {
+            }
+        }
+
+        @JsonProperty("ssl")
+        private SslAuthConfig sslAuthConfig;
+
+        public SslAuthConfig getSslAuthConfig() {
+            return sslAuthConfig;
         }
-    }
 
-    @JsonProperty("ssl")
-    private SslAuthConfig sslAuthConfig;
+    */
 
     @Valid
     @JsonProperty("sasl")
     private SaslAuthConfig saslAuthConfig;
 
-    public SslAuthConfig getSslAuthConfig() {
-        return sslAuthConfig;
-    }
-
     public SaslAuthConfig getSaslAuthConfig() {
         return saslAuthConfig;
     }

diff --git a/...ins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/TopicConfig.java b/...ins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/TopicConfig.java
@@ -21,12 +21,9 @@ public class TopicConfig {
     static final boolean DEFAULT_AUTO_COMMIT = false;
     static final Duration DEFAULT_COMMIT_INTERVAL = Duration.ofSeconds(5);
     static final Duration DEFAULT_SESSION_TIMEOUT = Duration.ofSeconds(45);
-    static final int DEFAULT_MAX_RETRY_ATTEMPT = Integer.MAX_VALUE;
     static final String DEFAULT_AUTO_OFFSET_RESET = "earliest";
     static final Duration DEFAULT_THREAD_WAITING_TIME = Duration.ofSeconds(5);
     static final Duration DEFAULT_MAX_RECORD_FETCH_TIME = Duration.ofSeconds(4);
-    static final Duration DEFAULT_BUFFER_TIMEOUT = Duration.ofSeconds(5);
-    static final Duration DEFAULT_MAX_RETRY_DELAY = Duration.ofSeconds(1);
     static final Integer DEFAULT_FETCH_MAX_BYTES = 52428800;
     static final Integer DEFAULT_FETCH_MAX_WAIT = 500;
     static final Integer DEFAULT_FETCH_MIN_BYTES = 1;
@@ -53,16 +50,6 @@ public class TopicConfig {
     @Size(min = 1, max = 200, message = "Number of worker threads should lies between 1 and 200")
     private Integer workers = DEFAULT_NUM_OF_WORKERS;
 
-    @JsonProperty("max_retry_attempts")
-    @Valid
-    @Size(min = 1, max = Integer.MAX_VALUE, message = " Max retry attempts should lies between 1 and Integer.MAX_VALUE")
-    private Integer maxRetryAttempts = DEFAULT_MAX_RETRY_ATTEMPT;
-
-    @JsonProperty("max_retry_delay")
-    @Valid
-    @Size(min = 1)
-    private Duration maxRetryDelay = DEFAULT_MAX_RETRY_DELAY;
-
     @JsonProperty("serde_format")
     private MessageFormat serdeFormat= MessageFormat.PLAINTEXT;
 
@@ -82,25 +69,12 @@ public class TopicConfig {
     @JsonProperty("auto_offset_reset")
     private String autoOffsetReset = DEFAULT_AUTO_OFFSET_RESET;
 
-    @JsonProperty("group_name")
-    @Valid
-    @Size(min = 1, max = 255, message = "size of group name should be between 1 and 255")
-    private String groupName;
-
     @JsonProperty("thread_waiting_time")
     private Duration threadWaitingTime = DEFAULT_THREAD_WAITING_TIME;
 
-    @JsonProperty("max_record_fetch_time")
-    private Duration maxRecordFetchTime = DEFAULT_MAX_RECORD_FETCH_TIME;
-
     @JsonProperty("max_partition_fetch_bytes")
     private Integer maxPartitionFetchBytes = DEFAULT_MAX_PARTITION_FETCH_BYTES;
 
-    @JsonProperty("buffer_default_timeout")
-    @Valid
-    @Size(min = 1)
-    private Duration bufferDefaultTimeout = DEFAULT_BUFFER_TIMEOUT;
-
     @JsonProperty("fetch_max_bytes")
     @Valid
     @Size(min = 1, max = 52428800)
@@ -144,10 +118,6 @@ public void setGroupId(String groupId) {
         this.groupId = groupId;
     }
 
-    public void setMaxRetryAttempts(Integer maxRetryAttempts) {
-        this.maxRetryAttempts = maxRetryAttempts;
-    }
-
     public MessageFormat getSerdeFormat() {
         return serdeFormat;
     }
@@ -176,14 +146,6 @@ public void setAutoOffsetReset(String autoOffsetReset) {
         this.autoOffsetReset = autoOffsetReset;
     }
 
-    public String getGroupName() {
-        return groupName;
-    }
-
-    public void setGroupName(String groupName) {
-        this.groupName = groupName;
-    }
-
     public Duration getThreadWaitingTime() {
         return threadWaitingTime;
     }
@@ -192,26 +154,10 @@ public void setThreadWaitingTime(Duration threadWaitingTime) {
         this.threadWaitingTime = threadWaitingTime;
     }
 
-    public Duration getMaxRecordFetchTime() {
-        return maxRecordFetchTime;
-    }
-
     public Integer getMaxPartitionFetchBytes() {
         return maxPartitionFetchBytes;
     }
 
-    public void setMaxRecordFetchTime(Duration maxRecordFetchTime) {
-        this.maxRecordFetchTime = maxRecordFetchTime;
-    }
-
-    public Duration getBufferDefaultTimeout() {
-        return bufferDefaultTimeout;
-    }
-
-    public void setBufferDefaultTimeout(Duration bufferDefaultTimeout) {
-        this.bufferDefaultTimeout = bufferDefaultTimeout;
-    }
-
     public Integer getFetchMaxBytes() {
         return fetchMaxBytes;
     }
@@ -264,14 +210,6 @@ public void setWorkers(Integer workers) {
         this.workers = workers;
     }
 
-    public Duration getMaxRetryDelay() {
-        return maxRetryDelay;
-    }
-
-    public void setMaxRetryDelay(Duration maxRetryDelay) {
-        this.maxRetryDelay = maxRetryDelay;
-    }
-
     public Duration getHeartBeatInterval() {
         return heartBeatInterval;
     }

diff --git a/...src/test/java/org/opensearch/dataprepper/plugins/kafka/configuration/TopicConfigTest.java b/...src/test/java/org/opensearch/dataprepper/plugins/kafka/configuration/TopicConfigTest.java
@@ -73,8 +73,6 @@ void testConfigValues_default() {
         assertEquals(TopicConfig.DEFAULT_SESSION_TIMEOUT, topicConfig.getSessionTimeOut());
         assertEquals(TopicConfig.DEFAULT_AUTO_OFFSET_RESET, topicConfig.getAutoOffsetReset());
         assertEquals(TopicConfig.DEFAULT_THREAD_WAITING_TIME, topicConfig.getThreadWaitingTime());
-        assertEquals(TopicConfig.DEFAULT_MAX_RECORD_FETCH_TIME, topicConfig.getMaxRecordFetchTime());
-        assertEquals(TopicConfig.DEFAULT_BUFFER_TIMEOUT, topicConfig.getBufferDefaultTimeout());
         assertEquals(TopicConfig.DEFAULT_FETCH_MAX_BYTES, topicConfig.getFetchMaxBytes());
         assertEquals(TopicConfig.DEFAULT_FETCH_MAX_WAIT, topicConfig.getFetchMaxWait());
         assertEquals(TopicConfig.DEFAULT_FETCH_MIN_BYTES, topicConfig.getFetchMinBytes());
@@ -96,8 +94,6 @@ void testConfigValues_from_yaml() {
         assertEquals(45000, topicConfig.getSessionTimeOut().toMillis());
         assertEquals("earliest", topicConfig.getAutoOffsetReset());
         assertEquals(Duration.ofSeconds(1), topicConfig.getThreadWaitingTime());
-        assertEquals(Duration.ofSeconds(4), topicConfig.getMaxRecordFetchTime());
-        assertEquals(Duration.ofSeconds(5), topicConfig.getBufferDefaultTimeout());
         assertEquals(52428800, topicConfig.getFetchMaxBytes().longValue());
         assertEquals(500L, topicConfig.getFetchMaxWait().longValue());
         assertEquals(1L, topicConfig.getFetchMinBytes().longValue());
@@ -118,8 +114,6 @@ void testConfigValues_from_yaml_not_null() {
         assertNotNull(topicConfig.getSessionTimeOut());
         assertNotNull(topicConfig.getAutoOffsetReset());
         assertNotNull(topicConfig.getThreadWaitingTime());
-        assertNotNull(topicConfig.getMaxRecordFetchTime());
-        assertNotNull(topicConfig.getBufferDefaultTimeout());
         assertNotNull(topicConfig.getFetchMaxBytes());
         assertNotNull(topicConfig.getFetchMaxWait());
         assertNotNull(topicConfig.getFetchMinBytes());

diff --git a/data-prepper-plugins/kafka-plugins/src/test/resources/sample-pipelines-1.yaml b/data-prepper-plugins/kafka-plugins/src/test/resources/sample-pipelines-1.yaml
@@ -5,7 +5,6 @@ log-pipeline:
         - 127.0.0.1:9093
       topics:
         - name: my-topic-2
-          group_name: kafka-consumer-group-2
           group_id: my-test-group
         - name: my-topic-1
           group_id: my-test-group

diff --git a/data-prepper-plugins/kafka-plugins/src/test/resources/sample-pipelines.yaml b/data-prepper-plugins/kafka-plugins/src/test/resources/sample-pipelines.yaml
@@ -12,12 +12,9 @@ log-pipeline:
           auto_commit: false
           commit_interval: PT5S
           session_timeout: PT45S
-          max_retry_attempts: 1000
           auto_offset_reset: earliest
           thread_waiting_time: PT1S
-          max_record_fetch_time: PT4S
           heart_beat_interval: PT3S
-          buffer_default_timeout: PT5S
           fetch_max_bytes: 52428800
           fetch_max_wait: 500
           fetch_min_bytes: 1