GH-403 fix ConsumerSeekAware with acknowledgingML

Resolves #403

remove test duplicates and use dedicated topic for testSeekAck

* Remove in the `ListenerConsumer` redundant `kafkaDataListener`
property in favor of `theListener`
* Some simple polishing

Author: tbinias

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

@@ -75,6 +75,7 @@
  * @author Martin Dam
  * @author Artem Bilan
  * @author Loic Talhouarne
+ * @author Thorsten Binias
  */
 public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListenerContainer<K, V> {
 
@@ -258,8 +259,6 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final Map<String, Map<Integer, Long>> offsets = new HashMap<>();
 
-		private final GenericMessageListener<?> genericListener;
-
 		private final MessageListener<K, V> listener;
 
 		private final AcknowledgingMessageListener<K, V> acknowledgingMessageListener;
@@ -319,13 +318,12 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 		ListenerConsumer(GenericMessageListener<?> listener, GenericAcknowledgingMessageListener<?> ackListener) {
 			Assert.state(!this.isAnyManualAck || !this.autoCommit,
 					"Consumer cannot be configured for auto commit for ackMode " + this.containerProperties.getAckMode());
-			@SuppressWarnings("deprecation")
-			final Consumer<K, V> consumer =
+			@SuppressWarnings("deprecation") final Consumer<K, V> consumer =
 					KafkaMessageListenerContainer.this.consumerFactory instanceof
-									org.springframework.kafka.core.ClientIdSuffixAware
+							org.springframework.kafka.core.ClientIdSuffixAware
 							? ((org.springframework.kafka.core.ClientIdSuffixAware<K, V>) KafkaMessageListenerContainer
-									.this.consumerFactory)
-										.createConsumer(KafkaMessageListenerContainer.this.clientIdSuffix)
+							.this.consumerFactory)
+							.createConsumer(KafkaMessageListenerContainer.this.clientIdSuffix)
 							: KafkaMessageListenerContainer.this.consumerFactory.createConsumer();
 
 			this.theListener = listener == null ? ackListener : listener;
@@ -351,7 +349,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 			}
 			this.consumer = consumer;
 			GenericErrorHandler<?> errHandler = this.containerProperties.getGenericErrorHandler();
-			this.genericListener = listener;			if (this.theListener instanceof BatchAcknowledgingMessageListener) {
+			if (this.theListener instanceof BatchAcknowledgingMessageListener) {
 				this.listener = null;
 				this.batchListener = null;
 				this.acknowledgingMessageListener = null;
@@ -457,7 +455,7 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 									KafkaMessageListenerContainer.this.getContainerProperties().getCommitCallback());
 						}
 					}
-					if (ListenerConsumer.this.genericListener instanceof ConsumerSeekAware) {
+					if (ListenerConsumer.this.theListener instanceof ConsumerSeekAware) {
 						seekPartitions(partitions, false);
 					}
 					// We will not start the invoker thread if we are in autocommit mode,
@@ -489,10 +487,10 @@ public void seek(String topic, int partition, long offset) {
 
 			};
 			if (idle) {
-				((ConsumerSeekAware) ListenerConsumer.this.genericListener).onIdleContainer(current, callback);
+				((ConsumerSeekAware) ListenerConsumer.this.theListener).onIdleContainer(current, callback);
 			}
 			else {
-				((ConsumerSeekAware) ListenerConsumer.this.genericListener).onPartitionsAssigned(current, callback);
+				((ConsumerSeekAware) ListenerConsumer.this.theListener).onPartitionsAssigned(current, callback);
 			}
 		}
 
@@ -530,8 +528,8 @@ public boolean isLongLived() {
 
 		@Override
 		public void run() {
-			if (this.genericListener instanceof ConsumerSeekAware) {
-				((ConsumerSeekAware) this.genericListener).registerSeekCallback(this);
+			if (this.theListener instanceof ConsumerSeekAware) {
+				((ConsumerSeekAware) this.theListener).registerSeekCallback(this);
 			}
 			this.count = 0;
 			this.last = System.currentTimeMillis();
@@ -588,7 +586,7 @@ public void run() {
 									&& now > lastAlertAt + this.containerProperties.getIdleEventInterval()) {
 								publishIdleContainerEvent(now - lastReceive);
 								lastAlertAt = now;
-								if (this.genericListener instanceof ConsumerSeekAware) {
+								if (this.theListener instanceof ConsumerSeekAware) {
 									seekPartitions(getAssignedPartitions(), true);
 								}
 							}

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

@@ -82,6 +82,7 @@
  * @author Martin Dam
  * @author Artem Bilan
  * @author Loic Talhouarne
+ * @author Thorsten Binias
  */
 public class KafkaMessageListenerContainerTests {
 
@@ -119,9 +120,11 @@ public class KafkaMessageListenerContainerTests {
 
 	private static String topic16 = "testTopic16";
 
+	private static String topic17 = "testTopic17";
+
 	@ClassRule
 	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5,
-			topic6, topic7, topic8, topic9, topic10, topic11, topic12, topic13, topic14, topic15, topic16);
+			topic6, topic7, topic8, topic9, topic10, topic11, topic12, topic13, topic14, topic15, topic16, topic17);
 
 	@Rule
 	public TestName testName = new TestName();
@@ -849,20 +852,20 @@ public void testBatchListenerErrors() throws Exception {
 	@Test
 	public void testSeek() throws Exception {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("test11", "false", embeddedKafka);
-		testSeekGuts(props, topic11, false);
+		testSeekGuts(props, topic11, false, false);
 	}
 
 	@Test
 	public void testSeekAutoCommit() throws Exception {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("test12", "true", embeddedKafka);
-		testSeekGuts(props, topic12, true);
+		testSeekGuts(props, topic12, true, false);
 	}
 
 	@Test
 	public void testSeekAutoCommitDefault() throws Exception {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("test15", "true", embeddedKafka);
 		props.remove(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG); // test true by default
-		testSeekGuts(props, topic15, true);
+		testSeekGuts(props, topic15, true, false);
 	}
 
 	@Test
@@ -912,7 +915,13 @@ public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekC
 		container.stop();
 	}
 
-	private void testSeekGuts(Map<String, Object> props, String topic, boolean autoCommit) throws Exception {
+	@Test
+	public void testSeekAck() throws Exception {
+		Map<String, Object> props = KafkaTestUtils.consumerProps("test17", "false", embeddedKafka);
+		testSeekGuts(props, topic17, false, true);
+	}
+
+	private void testSeekGuts(Map<String, Object> props, String topic, boolean autoCommit, boolean acknowledging) throws Exception {
 		logger.info("Start seek " + topic);
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
 		ContainerProperties containerProps = new ContainerProperties(topic);
@@ -964,7 +973,17 @@ public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekC
 			}
 
 		}
-		Listener messageListener = new Listener();
+
+		class AcknowledgingListener extends Listener implements AcknowledgingMessageListener<Integer, String> {
+
+			@Override
+			public void onMessage(ConsumerRecord<Integer, String> data, Acknowledgment acknowledgment) {
+				super.onMessage(data);
+			}
+
+		}
+
+		Listener messageListener = acknowledging ? new AcknowledgingListener() : new Listener();
 		containerProps.setMessageListener(messageListener);
 		containerProps.setSyncCommits(true);
 		containerProps.setAckMode(AckMode.RECORD);