Add support for max.request.size and batch.size configuration Kafka Event Listener

Author: vmtuan12

docs/src/main/sphinx/admin/event-listeners-kafka.md

@@ -96,6 +96,14 @@ Use the following properties for further configuration.
     distinction in Kafka, if multiple Trino clusters send events to the same
     Kafka system.
   - 
+* - `kafka-event-listener.max-request-size`
+  - [Size value](prop-type-data-size) that specifies the maximum size
+    of a request that the Kafka producer will send. Useful to control the size of log events being sent to Kafka.
+  - `5MB`
+* - `kafka-event-listener.batch-size`
+  - [Size value](prop-type-data-size) that specifies the size to batch before sending records to Kafka. 
+    Helps improve throughput by batching multiple events into fewer requests.
+  - `16384B`
 * - `kafka-event-listener.publish-created-event`
   - [Boolean](prop-type-boolean) switch to control publishing of query creation
     events.

plugin/trino-kafka-event-listener/src/main/java/io/trino/plugin/eventlistener/kafka/KafkaEventListenerConfig.java

@@ -19,6 +19,8 @@
 import io.airlift.configuration.ConfigDescription;
 import io.airlift.configuration.DefunctConfig;
 import io.airlift.configuration.validation.FileExists;
+import io.airlift.units.DataSize;
+import io.airlift.units.DataSize.Unit;
 import io.airlift.units.Duration;
 import io.airlift.units.MinDuration;
 import jakarta.validation.constraints.AssertTrue;
@@ -48,6 +50,8 @@ public class KafkaEventListenerConfig
     private Optional<String> splitCompletedTopicName = Optional.empty();
     private String brokerEndpoints;
     private Optional<String> clientId = Optional.empty();
+    private DataSize maxRequestSize = DataSize.of(5, Unit.MEGABYTE); // Greater than default value because the size of completed events are quite large
+    private DataSize batchSize = DataSize.of(16384, Unit.BYTE); // Default value of batch.size
     private Set<String> excludedFields = Collections.emptySet();
     private Duration requestTimeout = new Duration(10, SECONDS);
     private boolean terminateOnInitializationFailure = true;
@@ -91,6 +95,32 @@ public KafkaEventListenerConfig setClientId(String clientId)
         return this;
     }
 
+    public DataSize getMaxRequestSize()
+    {
+        return maxRequestSize;
+    }
+
+    @ConfigDescription("The maximum size of a request/message in bytes")
+    @Config("kafka-event-listener.max-request-size")
+    public KafkaEventListenerConfig setMaxRequestSize(DataSize maxRequestSize)
+    {
+        this.maxRequestSize = maxRequestSize;
+        return this;
+    }
+
+    public DataSize getBatchSize()
+    {
+        return batchSize;
+    }
+
+    @ConfigDescription("Attempt to batch records/messages together up to batch.size bytes before sending")
+    @Config("kafka-event-listener.batch-size")
+    public KafkaEventListenerConfig setBatchSize(DataSize batchSize)
+    {
+        this.batchSize = batchSize;
+        return this;
+    }
+
     public Optional<String> getCompletedTopicName()
     {
         return completedTopicName;

plugin/trino-kafka-event-listener/src/main/java/io/trino/plugin/eventlistener/kafka/producer/BaseKafkaProducerFactory.java

@@ -39,7 +39,8 @@ protected Map<String, Object> baseConfig(KafkaEventListenerConfig config)
         kafkaClientConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
         kafkaClientConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
         kafkaClientConfig.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "zstd");
-        kafkaClientConfig.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "5242880");
+        kafkaClientConfig.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, Long.toString(config.getMaxRequestSize().toBytes()));
+        kafkaClientConfig.put(ProducerConfig.BATCH_SIZE_CONFIG, Long.toString(config.getBatchSize().toBytes()));
         kafkaClientConfig.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, Long.toString(config.getRequestTimeout().toMillis()));
         return kafkaClientConfig;
     }

plugin/trino-kafka-event-listener/src/test/java/io/trino/plugin/eventlistener/kafka/TestKafkaEventListenerConfig.java

@@ -27,6 +27,8 @@
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
+import io.airlift.units.DataSize;
+import io.airlift.units.DataSize.Unit;
 import static io.airlift.configuration.testing.ConfigAssertions.assertFullMapping;
 import static io.airlift.configuration.testing.ConfigAssertions.assertRecordedDefaults;
 import static io.airlift.configuration.testing.ConfigAssertions.recordDefaults;
@@ -46,6 +48,8 @@ void testDefaults()
                 .setCreatedTopicName(null)
                 .setSplitCompletedTopicName(null)
                 .setBrokerEndpoints(null)
+                .setMaxRequestSize(DataSize.of(5, Unit.MEGABYTE))
+                .setBatchSize(DataSize.of(16384, Unit.BYTE))
                 .setClientId(null)
                 .setExcludedFields(Set.of())
                 .setRequestTimeout(new Duration(10, TimeUnit.SECONDS))
@@ -66,6 +70,8 @@ void testExplicitPropertyMappings()
                 .put("kafka-event-listener.publish-completed-event", "false")
                 .put("kafka-event-listener.publish-split-completed-event", "true")
                 .put("kafka-event-listener.broker-endpoints", "kafka-host-1:9093,kafka-host-2:9093")
+                .put("kafka-event-listener.max-request-size", "1048576")
+                .put("kafka-event-listener.batch-size", "81920")
                 .put("kafka-event-listener.created-event.topic", "query_created")
                 .put("kafka-event-listener.completed-event.topic", "query_completed")
                 .put("kafka-event-listener.split-completed-event.topic", "split_completed")
@@ -84,6 +90,8 @@ void testExplicitPropertyMappings()
                 .setPublishCompletedEvent(false)
                 .setPublishSplitCompletedEvent(true)
                 .setBrokerEndpoints("kafka-host-1:9093,kafka-host-2:9093")
+                .setMaxRequestSize(DataSize.of(1048576, Unit.BYTE))
+                .setBatchSize(DataSize.of(81920, Unit.BYTE))
                 .setCreatedTopicName("query_created")
                 .setCompletedTopicName("query_completed")
                 .setSplitCompletedTopicName("split_completed")