fix(server-kafka): ignore commits without record changes

Author: steinerda

sda-commons-server-kafka/src/main/java/org/sdase/commons/server/kafka/consumer/strategies/synccommit/SyncCommitMLS.java

@@ -44,8 +44,11 @@ public SyncCommitMLS(
 
   @Override
   public void processRecords(ConsumerRecords<K, V> records, KafkaConsumer<K, V> consumer) {
-    super.processRecords(records, consumer);
-    commitSync(consumer);
+    // Only process and commit if there are records to process
+    if (!records.isEmpty()) {
+      super.processRecords(records, consumer);
+      this.commitSync(consumer);
+    }
   }
 
   private void commitSync(KafkaConsumer<K, V> consumer) {

sda-commons-server-kafka/src/test/java/org/sdase/commons/server/kafka/consumer/SyncCommitStrategyTest.java

@@ -6,6 +6,8 @@
 import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.when;
 
+import java.util.Collections;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.junit.jupiter.api.BeforeEach;
@@ -83,4 +85,18 @@ void shouldStillThrowOtherExceptionsIfNoSyncCommitErrorHandlerConfigured() {
 
     Mockito.verify(consumer, timeout(100).times(1)).commitSync();
   }
+
+  @Test
+  void shouldNotCommitOnEmptyRecords() {
+    // given
+    SyncCommitMLS<String, String> strategy = new SyncCommitMLS<>(handler, errorHandler);
+    strategy.init(null);
+    ConsumerRecords<String, String> emptyRecords = new ConsumerRecords<>(Collections.emptyMap());
+
+    // when the strategy processes the empty records
+    strategy.processRecords(emptyRecords, consumer);
+
+    // then commitSync() should never be called on the consumer
+    Mockito.verify(consumer, Mockito.never()).commitSync();
+  }
 }