RemoteFetcherTierStatemachine logging

aiven · Jan 8, 2024 · 5921104 · 5921104
1 parent 3df4601
commit 5921104
Showing 1 changed file with 8 additions and 4 deletions.
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java b/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java
@@ -164,8 +164,12 @@ private void processConsumerRecord(ConsumerRecord<byte[], byte[]> record) {
             log.debug("The event {} is skipped because it is either already processed or not assigned to this consumer",
                     remoteLogMetadata);
         }
-        log.error("Updating consumed offset: {} for partition {}", record.offset(), record.partition());
-        readOffsetsByMetadataPartition.put(record.partition(), record.offset());
+        if(!readOffsetsByMetadataPartition.containsKey(record.partition()) ||
+                readOffsetsByMetadataPartition.get(record.partition()) == record.offset() -1) {
+            log.error("Updating consumed offset: {} for partition {} previously read offsets {}",
+                    record.offset(), record.partition(), readOffsetsByMetadataPartition.get(record.partition()));
+            readOffsetsByMetadataPartition.put(record.partition(), record.offset());
+        }
     }
 
     private boolean shouldProcess(final RemoteLogMetadata metadata, final long recordOffset) {
@@ -259,8 +263,8 @@ void maybeWaitForPartitionAssignments() throws InterruptedException {
             remoteLogPartitions.stream()
                 .filter(tp -> !seekToBeginOffsetPartitions.contains(tp) &&
                     readOffsetsByMetadataPartition.containsKey(tp.partition()))
-                .peek(tp -> log.error("Reading from the offset where the processing left last time for partition: {}",
-                    tp.partition()))
+                .peek(tp -> log.error("Reading from the offset {} where the processing left last time for partition: {}",
+                    readOffsetsByMetadataPartition.get(tp.partition()), tp.partition()))
                 .forEach(tp -> consumer.seek(tp, readOffsetsByMetadataPartition.get(tp.partition())));
             Set<TopicIdPartition> processedAssignmentPartitions = new HashSet<>();
             // mark all the user-topic-partitions as assigned to the consumer.