Don't commit on assignment with existing commit

Unnecessary to perform the redundant commit on assignment if there
is already a committed offset for a partition.

It is only necessary when using `latest` auto offset reset and when
there is no current committed offset.

Author: garyrussell

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

@@ -23,6 +23,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.LinkedHashSet;
 import java.util.LinkedList;
@@ -2353,25 +2354,7 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 				}
 				ListenerConsumer.this.assignedPartitions.addAll(partitions);
 				if (ListenerConsumer.this.commitCurrentOnAssignment) {
-					// Commit initial positions - this is generally redundant but
-					// it protects us from the case when another consumer starts
-					// and rebalance would cause it to reset at the end
-					// see https://github.com/spring-projects/spring-kafka/issues/110
-					Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
-					for (TopicPartition partition : partitions) {
-						try {
-							offsetsToCommit.put(partition,
-									new OffsetAndMetadata(ListenerConsumer.this.consumer.position(partition)));
-						}
-						catch (NoOffsetForPartitionException e) {
-							ListenerConsumer.this.fatalError = true;
-							ListenerConsumer.this.logger.error(e, "No offset and no reset policy");
-							return;
-						}
-					}
-					if (offsetsToCommit.size() > 0) {
-						commitCurrentOffsets(offsetsToCommit);
-					}
+					collectAndCommitIfNecessary(partitions);
 				}
 				if (ListenerConsumer.this.genericListener instanceof ConsumerSeekAware) {
 					seekPartitions(partitions, false);
@@ -2384,6 +2367,32 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 				}
 			}
 
+			private void collectAndCommitIfNecessary(Collection<TopicPartition> partitions) {
+				// Commit initial positions - this is generally redundant but
+				// it protects us from the case when another consumer starts
+				// and rebalance would cause it to reset at the end
+				// see https://github.com/spring-projects/spring-kafka/issues/110
+				Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
+				Map<TopicPartition, OffsetAndMetadata> committed =
+						ListenerConsumer.this.consumer.committed(new HashSet<>(partitions));
+				for (TopicPartition partition : partitions) {
+					try {
+						if (committed.get(partition) == null) { // no existing commit for this group
+							offsetsToCommit.put(partition,
+									new OffsetAndMetadata(ListenerConsumer.this.consumer.position(partition)));
+						}
+					}
+					catch (NoOffsetForPartitionException e) {
+						ListenerConsumer.this.fatalError = true;
+						ListenerConsumer.this.logger.error(e, "No offset and no reset policy");
+						return;
+					}
+				}
+				if (offsetsToCommit.size() > 0) {
+					commitCurrentOffsets(offsetsToCommit);
+				}
+			}
+
 			private void commitCurrentOffsets(Map<TopicPartition, OffsetAndMetadata> offsetsToCommit) {
 				ListenerConsumer.this.commitLogger.log(() -> "Committing on assignment: " + offsetsToCommit);
 				if (ListenerConsumer.this.transactionTemplate != null

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

@@ -47,6 +47,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.common.TopicPartition;
@@ -527,6 +528,62 @@ void testNoCommitOnAssignmentWithEarliest() throws InterruptedException {
 		verify(consumer, never()).commitSync(any(), any());
 	}
 
+	@Test
+	void testNoInitialCommitIfAlreadyCommitted() throws InterruptedException {
+		testInitialCommitIBasedOnCommitted(true);
+	}
+
+	@Test
+	void testInitialCommitIfNotAlreadyCommitted() throws InterruptedException {
+		testInitialCommitIBasedOnCommitted(false);
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	private void testInitialCommitIBasedOnCommitted(boolean committed) throws InterruptedException {
+		Consumer consumer = mock(Consumer.class);
+		ConsumerRecords records = new ConsumerRecords<>(Collections.emptyMap());
+		CountDownLatch latch = new CountDownLatch(1);
+		willAnswer(inv -> {
+			latch.countDown();
+			Thread.sleep(50);
+			return records;
+		}).given(consumer).poll(any());
+		TopicPartition tp0 = new TopicPartition("foo", 0);
+		List<TopicPartition> assignments = Arrays.asList(tp0);
+		willAnswer(invocation -> {
+			((ConsumerRebalanceListener) invocation.getArgument(1))
+				.onPartitionsAssigned(assignments);
+			return null;
+		}).given(consumer).subscribe(any(Collection.class), any());
+		if (committed) {
+			given(consumer.committed(Collections.singleton(tp0)))
+					.willReturn(Collections.singletonMap(tp0, new OffsetAndMetadata(0L)));
+		}
+		else {
+			given(consumer.committed(Collections.singleton(tp0)))
+					.willReturn(Collections.singletonMap(tp0, null));
+		}
+		ConsumerFactory cf = mock(ConsumerFactory.class);
+		given(cf.createConsumer(any(), any(), any(), any())).willReturn(consumer);
+		given(cf.getConfigurationProperties())
+				.willReturn(Collections.singletonMap(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"));
+		ContainerProperties containerProperties = new ContainerProperties("foo");
+		containerProperties.setGroupId("grp");
+		containerProperties.setMessageListener((MessageListener) rec -> { });
+		containerProperties.setMissingTopicsFatal(false);
+		containerProperties.setAssignmentCommitOption(AssignmentCommitOption.LATEST_ONLY);
+		ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer(cf,
+				containerProperties);
+		container.start();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		if (committed) {
+			verify(consumer, never()).commitSync(any(), any());
+		}
+		else {
+			verify(consumer).commitSync(any(), any());
+		}
+	}
+
 	public static class TestMessageListener1 implements MessageListener<String, String>, ConsumerSeekAware {
 
 		private static ThreadLocal<ConsumerSeekCallback> callbacks = new ThreadLocal<>();