Copes with transaction markers. Cloes #64

obsidiandynamics · Mar 9, 2020 · 28335f3 · 28335f3
1 parent fd737b2
commit 28335f3
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/src/main/java/kafdrop/service/KafkaHighLevelConsumer.java b/src/main/java/kafdrop/service/KafkaHighLevelConsumer.java
@@ -42,6 +42,7 @@ private void initializeClient() {
       properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
       properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
       properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "kafdrop-consumer");
+      properties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
       kafkaConfiguration.applyCommon(properties);
 
       kafkaConsumer = new KafkaConsumer<>(properties);
@@ -92,6 +93,7 @@ synchronized List<ConsumerRecord<String, String>> getLatestRecords(TopicPartitio
                                                                      MessageDeserializer deserializer) {
     initializeClient();
     final var partitions = Collections.singletonList(partition);
+    final var maxEmptyPolls = 3; // caps the number of empty polls
     kafkaConsumer.assign(partitions);
     kafkaConsumer.seek(partition, offset);
 
@@ -102,12 +104,16 @@ synchronized List<ConsumerRecord<String, String>> getLatestRecords(TopicPartitio
     var currentOffset = offset - 1;
 
     // stop if got to count or get to the latest offset
+    var emptyPolls = 0;
     while (rawRecords.size() < count && currentOffset < latestOffset) {
       final var polled = kafkaConsumer.poll(Duration.ofMillis(POLL_TIMEOUT_MS)).records(partition);
 
-      if (!polled.isEmpty()) {
+      if (! polled.isEmpty()) {
         rawRecords.addAll(polled);
         currentOffset = polled.get(polled.size() - 1).offset();
+        emptyPolls = 0;
+      } else if (++emptyPolls == maxEmptyPolls) {
+        break;
       }
     }