Changed the sequence of commit&reset before consuming againq

Signed-off-by: Krishna Kondaka <[email protected]>
kkondaka · Aug 21, 2023 · 251dfd1 · 251dfd1
1 parent 784a9d5
commit 251dfd1
Showing 1 changed file with 3 additions and 3 deletions.
diff --git a/...ain/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaSourceCustomConsumer.java b/...ain/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaSourceCustomConsumer.java
@@ -287,12 +287,12 @@ public void run() {
                     Thread.sleep(10000);
                 }
                 synchronized(this) {
+                    commitOffsets();
+                    resetOffsets();
                     if (currentEpoch != newEpoch) {
                         partitionCommitTrackerMap.clear();
+                        currentEpoch = newEpoch;
                     }
-                    resetOffsets();
-                    commitOffsets();
-                    currentEpoch = newEpoch;
                 }
                 consumeRecords();
                 topicMetrics.update(consumer);