spring-projectsGH-2731: fix count and time ack mode

count ack mode commit offset when next poll return no records

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -1424,6 +1424,10 @@ protected void pollAndInvoke() {
 				this.firstPoll = true;
 				this.consumerSeekAwareListener.onFirstPoll();
 			}
+			if (records != null && records.count() == 0 && this.isCountAck && this.count > 0) {
+				commitIfNecessary();
+				this.count = 0;
+			}
 			debugRecords(records);
 
 			invokeIfHaveRecords(records);

spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

@@ -3060,12 +3060,12 @@ public void testAckModeCount() throws Exception {
 		final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records2 = new HashMap<>();
 		records2.put(topicPartition, Arrays.asList(
 				new ConsumerRecord<>("foo", 0, 2L, 1, "baz"),
-				new ConsumerRecord<>("foo", 0, 3L, 1, "qux"))); // commit (4 >= 3)
+				new ConsumerRecord<>("foo", 0, 3L, 1, "qux"))); // commit (3 = 3)
 		final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records3 = new HashMap<>();
 		records3.put(topicPartition, Arrays.asList(
 				new ConsumerRecord<>("foo", 0, 4L, 1, "fiz"),
-				new ConsumerRecord<>("foo", 0, 5L, 1, "buz"),
-				new ConsumerRecord<>("foo", 0, 6L, 1, "bif"))); // commit (3 >= 3)
+				new ConsumerRecord<>("foo", 0, 5L, 1, "buz"), // commit (3 = 3)
+				new ConsumerRecord<>("foo", 0, 6L, 1, "bif"))); // commit (1 when next poll returns no records)
 		ConsumerRecords<Integer, String> consumerRecords1 = new ConsumerRecords<>(records1);
 		ConsumerRecords<Integer, String> consumerRecords2 = new ConsumerRecords<>(records2);
 		ConsumerRecords<Integer, String> consumerRecords3 = new ConsumerRecords<>(records3);
@@ -3085,7 +3085,7 @@ public void testAckModeCount() throws Exception {
 					return emptyRecords;
 			}
 		});
-		final CountDownLatch commitLatch = new CountDownLatch(2);
+		final CountDownLatch commitLatch = new CountDownLatch(3);
 		willAnswer(i -> {
 			commitLatch.countDown();
 			return null;
@@ -3116,6 +3116,8 @@ public void testAckModeCount() throws Exception {
 				Duration.ofSeconds(42));
 		verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(6L)),
 				Duration.ofSeconds(42));
+		verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(7L)),
+				Duration.ofSeconds(42));
 		container.stop();
 	}