Using multiple successive polls to get as many messages as needed

obsidiandynamics · Jun 10, 2019 · 7e0bc62 · 7e0bc62
1 parent 56ddcab
commit 7e0bc62
Show file tree

Hide file tree

Showing 6 changed files with 15 additions and 11 deletions.
diff --git a/pom.xml b/pom.xml
@@ -5,7 +5,7 @@
 
     <groupId>com.obsidiandynamics.kafdrop</groupId>
     <artifactId>kafdrop</artifactId>
-    <version>3.4.0-SNAPSHOT</version>
+    <version>3.3.1</version>
 
     <description>For when you have a Kafka cluster to monitor</description>
 

diff --git a/src/main/java/kafdrop/controller/MessageController.java b/src/main/java/kafdrop/controller/MessageController.java
@@ -93,7 +93,7 @@ public String viewMessageForm(@PathVariable("name") String topicName,
                          messageInspector.getMessages(topicName,
                                                       messageForm.getPartition(),
                                                       messageForm.getOffset(),
-                                                      messageForm.getCount(),
+                                                      messageForm.getCount().intValue(),
                                                       deserializer));
 
     }
@@ -119,7 +119,7 @@ List<Object> getPartitionOrMessages(
       @PathVariable("name") String topicName,
       @RequestParam(name = "partition", required = false) Integer partition,
       @RequestParam(name = "offset", required = false) Long offset,
-      @RequestParam(name = "count", required = false) Long count
+      @RequestParam(name = "count", required = false) Integer count
   ) {
     if (partition == null || offset == null || count == null) {
       final TopicVO topic = kafkaMonitor.getTopic(topicName)

diff --git a/src/main/java/kafdrop/service/CuratorKafkaMonitor.java b/src/main/java/kafdrop/service/CuratorKafkaMonitor.java
@@ -221,7 +221,7 @@ private Map<String, TopicVO> getTopicMetadata(String... topics) {
   }
 
   @Override
-  public List<MessageVO> getMessages(TopicPartition topicPartition, long offset, long count,
+  public List<MessageVO> getMessages(TopicPartition topicPartition, long offset, int count,
                                      MessageDeserializer deserializer) {
     final List<ConsumerRecord<String, String>> records =
         kafkaHighLevelConsumer.getLatestRecords(topicPartition, offset, count, deserializer);

diff --git a/src/main/java/kafdrop/service/KafkaHighLevelConsumer.java b/src/main/java/kafdrop/service/KafkaHighLevelConsumer.java
@@ -86,16 +86,20 @@ synchronized Map<Integer, TopicPartitionVO> getPartitionSize(String topic) {
     return partitionsVo;
   }
 
-  synchronized List<ConsumerRecord<String, String>> getLatestRecords(TopicPartition topicPartition, long offset, Long count,
+  synchronized List<ConsumerRecord<String, String>> getLatestRecords(TopicPartition topicPartition, long offset, int count,
                                                                      MessageDeserializer deserializer) {
     initializeClient();
     kafkaConsumer.assign(Collections.singletonList(topicPartition));
     kafkaConsumer.seek(topicPartition, offset);
 
-    final var rawRecords = kafkaConsumer.poll(Duration.ofMillis(POLL_TIMEOUT_MS));
-    final var numRecords = rawRecords.count();
-    return rawRecords.records(topicPartition)
-        .subList(0, Math.min(count.intValue(), numRecords))
+    final var rawRecords = new ArrayList<ConsumerRecord<String, byte[]>>(count);
+    while (rawRecords.size() <= count) {
+      final var polled = kafkaConsumer.poll(Duration.ofMillis(POLL_TIMEOUT_MS)).records(topicPartition);
+      if (polled.isEmpty()) break;
+      rawRecords.addAll(polled);
+    }
+    return rawRecords
+        .subList(0, Math.min(count, rawRecords.size()))
         .stream()
         .map(rec -> new ConsumerRecord<>(rec.topic(),
                                          rec.partition(),

diff --git a/src/main/java/kafdrop/service/KafkaMonitor.java b/src/main/java/kafdrop/service/KafkaMonitor.java
@@ -31,7 +31,7 @@ public interface KafkaMonitor {
 
   List<TopicVO> getTopics();
 
-  List<MessageVO> getMessages(TopicPartition topicPartition, long offset, long count,
+  List<MessageVO> getMessages(TopicPartition topicPartition, long offset, int count,
                               MessageDeserializer deserializer);
 
   Optional<TopicVO> getTopic(String topic);

diff --git a/src/main/java/kafdrop/service/MessageInspector.java b/src/main/java/kafdrop/service/MessageInspector.java
@@ -33,7 +33,7 @@ public MessageInspector(KafkaMonitor kafkaMonitor) {
     this.kafkaMonitor = kafkaMonitor;
   }
 
-  public List<MessageVO> getMessages(String topicName, int partitionId, long offset, long count,
+  public List<MessageVO> getMessages(String topicName, int partitionId, long offset, int count,
                                      MessageDeserializer deserializer) {
     final var topicPartition = new TopicPartition(topicName, partitionId);
     return kafkaMonitor.getMessages(topicPartition, offset, count, deserializer);