Increase fetch.min.bytes to 64kb for kafka buffer consumer

Signed-off-by: Taylor Gray <[email protected]>
opensearch-project · Jan 31, 2025 · bf39614 · bf39614
1 parent 414a7c9
commit bf39614
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 3 deletions.
diff --git a/...ntegrationTest/java/org/opensearch/dataprepper/plugins/kafka/buffer/ReadBufferHelper.java b/...ntegrationTest/java/org/opensearch/dataprepper/plugins/kafka/buffer/ReadBufferHelper.java
@@ -19,9 +19,9 @@ class ReadBufferHelper {
     static Map.Entry<Collection<Record<Event>>, CheckpointState> awaitRead(final KafkaBuffer objectUnderTest) {
         final Map.Entry<Collection<Record<Event>>, CheckpointState>[] lastReadResult = new Map.Entry[1];
         await()
-                .atMost(Duration.ofSeconds(30))
+                .atMost(Duration.ofSeconds(45))
                 .until(() -> {
-                    lastReadResult[0] = objectUnderTest.read(500);
+                    lastReadResult[0] = objectUnderTest.read(2000);
                     return lastReadResult[0] != null && lastReadResult[0].getKey() != null && lastReadResult[0].getKey().size() >= 1;
                 });
         return lastReadResult[0];

diff --git a/...gins/src/main/java/org/opensearch/dataprepper/plugins/kafka/buffer/BufferTopicConfig.java b/...gins/src/main/java/org/opensearch/dataprepper/plugins/kafka/buffer/BufferTopicConfig.java
@@ -27,7 +27,7 @@ class BufferTopicConfig extends CommonTopicConfig implements TopicProducerConfig
     static final boolean DEFAULT_AUTO_COMMIT = false;
     static final ByteCount DEFAULT_FETCH_MAX_BYTES = ByteCount.parse("50mb");
     static final Duration DEFAULT_FETCH_MAX_WAIT = Duration.ofMillis(1000);
-    static final ByteCount DEFAULT_FETCH_MIN_BYTES = ByteCount.parse("2kb");
+    static final ByteCount DEFAULT_FETCH_MIN_BYTES = ByteCount.parse("64kb");
     static final ByteCount DEFAULT_MAX_PARTITION_FETCH_BYTES = ByteCount.parse("1mb");
     static final Duration DEFAULT_SESSION_TIMEOUT = Duration.ofSeconds(45);
     static final String DEFAULT_AUTO_OFFSET_RESET = "earliest";