Single instance per topic rather than per worker

Signed-off-by: Chase Engelbrecht <[email protected]>
engechas · Oct 25, 2023 · abb69fa · abb69fa
1 parent 22b33ac
commit abb69fa
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 2 deletions.
diff --git a/...in/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumerFactory.java b/...in/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumerFactory.java
@@ -80,6 +80,7 @@ public List<KafkaCustomConsumer> createConsumersForTopic(final KafkaConsumerConf
 
         try {
             final int numWorkers = topic.getWorkers();
+            final TopicEmptinessMetadata topicEmptinessMetadata = new TopicEmptinessMetadata();
             IntStream.range(0, numWorkers).forEach(index -> {
                 KafkaDataConfig dataConfig = new KafkaDataConfigAdapter(keyFactory, topic);
                 Deserializer<Object> keyDeserializer = (Deserializer<Object>) serializationFactory.getDeserializer(PlaintextKafkaDataConfig.plaintextDataConfig(dataConfig));
@@ -93,7 +94,7 @@ public List<KafkaCustomConsumer> createConsumersForTopic(final KafkaConsumerConf
                 final KafkaConsumer kafkaConsumer = new KafkaConsumer<>(consumerProperties, keyDeserializer, valueDeserializer);
 
                 consumers.add(new KafkaCustomConsumer(kafkaConsumer, shutdownInProgress, buffer, kafkaConsumerConfig, topic,
-                    schemaType, acknowledgementSetManager, byteDecoder, topicMetrics, new TopicEmptinessMetadata()));
+                    schemaType, acknowledgementSetManager, byteDecoder, topicMetrics, topicEmptinessMetadata));
 
             });
         } catch (Exception e) {

diff --git a/...ka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/source/KafkaSource.java b/...ka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/source/KafkaSource.java
@@ -117,6 +117,7 @@ public void start(Buffer<Record<Event>> buffer) {
                 int numWorkers = topic.getWorkers();
                 executorService = Executors.newFixedThreadPool(numWorkers);
                 allTopicExecutorServices.add(executorService);
+                final TopicEmptinessMetadata topicEmptinessMetadata = new TopicEmptinessMetadata();
 
                 IntStream.range(0, numWorkers).forEach(index -> {
                     while (true) {
@@ -140,7 +141,7 @@ public void start(Buffer<Record<Event>> buffer) {
 
                     }
                     consumer = new KafkaCustomConsumer(kafkaConsumer, shutdownInProgress, buffer, sourceConfig, topic, schemaType,
-                            acknowledgementSetManager, null, topicMetrics, new TopicEmptinessMetadata());
+                            acknowledgementSetManager, null, topicMetrics, topicEmptinessMetadata);
                     allTopicConsumers.add(consumer);
 
                     executorService.submit(consumer);