Changes for compatibility with Spring Kafka 2.3

Also remove test for bad partition in DSL tests - the send now gives an error
about the topic not being present in metadata instead of "not in range".

Use sync for publishing in DSL tests, to fail fast.

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -42,7 +42,6 @@ import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
 
-import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.acks.AcknowledgmentCallback;
@@ -56,6 +55,7 @@ import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.converter.KafkaMessageHeaders;
 import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
@@ -80,8 +80,7 @@ import org.springframework.util.Assert;
  * @since 3.0.1
  *
  */
-public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
-		implements DisposableBean, Lifecycle {
+public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> implements Lifecycle {
 
 	private static final long DEFAULT_POLL_TIMEOUT = 50L;
 
@@ -116,6 +115,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
 
 	private boolean rawMessageHeader;
 
+	private Duration commitTimeout;
+
 	private volatile Consumer<K, V> consumer;
 
 	private boolean running;
@@ -232,6 +233,20 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
 		this.rawMessageHeader = rawMessageHeader;
 	}
 
+	protected Duration getCommitTimeout() {
+		return this.commitTimeout;
+	}
+
+	/**
+	 * Set the timeout for commits.
+	 * @param commitTimeout the timeout.
+	 * @since 3.2
+	 */
+	public void setCommitTimeout(Duration commitTimeout) {
+		this.commitTimeout = commitTimeout;
+		this.ackCallbackFactory.setCommitTimeout(commitTimeout);
+	}
+
 	private ConsumerFactory<K, V> fixOrRejectConsumerFactory(ConsumerFactory<K, V> suppliedConsumerFactory) {
 		Object maxPoll = suppliedConsumerFactory.getConfigurationProperties()
 				.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG);
@@ -372,9 +387,15 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
 	 */
 	public static class KafkaAckCallbackFactory<K, V> implements AcknowledgmentCallbackFactory<KafkaAckInfo<K, V>> {
 
+		private Duration commitTimeout;
+
+		public void setCommitTimeout(Duration commitTimeout) {
+			this.commitTimeout = commitTimeout;
+		}
+
 		@Override
 		public AcknowledgmentCallback createCallback(KafkaAckInfo<K, V> info) {
-			return new KafkaAckCallback<>(info);
+			return new KafkaAckCallback<>(info, this.commitTimeout);
 		}
 
 	}
@@ -391,13 +412,20 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
 
 		private final KafkaAckInfo<K, V> ackInfo;
 
+		private final Duration commitTimeout;
+
 		private volatile boolean acknowledged;
 
 		private boolean autoAckEnabled = true;
 
 		public KafkaAckCallback(KafkaAckInfo<K, V> ackInfo) {
+			this(ackInfo, null);
+		}
+
+		public KafkaAckCallback(KafkaAckInfo<K, V> ackInfo, @Nullable Duration commitTimeout) {
 			Assert.notNull(ackInfo, "'ackInfo' cannot be null");
 			this.ackInfo = ackInfo;
+			this.commitTimeout = commitTimeout;
 		}
 
 		@Override
@@ -494,8 +522,15 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
 						this.ackInfo.setAckDeferred(true);
 					}
 					if (ackInfo != null) {
-						ackInfo.getConsumer().commitSync(Collections.singletonMap(ackInfo.getTopicPartition(),
-								new OffsetAndMetadata(ackInfo.getRecord().offset() + 1)));
+						Map<TopicPartition, OffsetAndMetadata> offset =
+								Collections.singletonMap(ackInfo.getTopicPartition(),
+										new OffsetAndMetadata(ackInfo.getRecord().offset() + 1));
+						if (this.commitTimeout == null) {
+							ackInfo.getConsumer().commitSync(offset);
+						}
+						else {
+							ackInfo.getConsumer().commitSync(offset, this.commitTimeout);
+						}
 					}
 					else {
 						if (this.logger.isDebugEnabled()) {

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2018 the original author or authors.
+ * Copyright 2013-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -460,6 +460,7 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 	public void processSendResult(final Message<?> message, final ProducerRecord<K, V> producerRecord,
 			ListenableFuture<SendResult<K, V>> future, MessageChannel metadataChannel)
 			throws InterruptedException, ExecutionException {
+
 		if (getSendFailureChannel() != null || metadataChannel != null) {
 			future.addCallback(new ListenableFutureCallback<SendResult<K, V>>() {
 

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2018 the original author or authors.
+ * Copyright 2015-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,7 +17,6 @@
 package org.springframework.integration.kafka.dsl;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -28,6 +27,7 @@ import java.util.stream.Stream;
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.TopicPartition;
 import org.junit.ClassRule;
 import org.junit.Test;
@@ -44,7 +44,6 @@ import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.Pollers;
-import org.springframework.integration.expression.ValueExpression;
 import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
 import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
 import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
@@ -150,11 +149,6 @@ public class KafkaDslTests {
 	@Test
 	public void testKafkaAdapters() throws Exception {
 
-		assertThatThrownBy(() -> this.sendToKafkaFlowInput.send(new GenericMessage<>("foo")))
-				.hasMessageContaining("10 is not in the range");
-
-		this.kafkaProducer1.setPartitionIdExpression(new ValueExpression<>(0));
-		this.kafkaProducer2.setPartitionIdExpression(new ValueExpression<>(0));
 		this.sendToKafkaFlowInput.send(new GenericMessage<>("foo", Collections.singletonMap("foo", "bar")));
 
 		assertThat(TestUtils.getPropertyValue(this.kafkaProducer1, "headerMapper")).isSameAs(this.mapper);
@@ -298,7 +292,9 @@ public class KafkaDslTests {
 
 		@Bean
 		public ProducerFactory<Integer, String> producerFactory() {
-			return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka()));
+			Map<String, Object> props = KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka());
+			props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);
+			return new DefaultKafkaProducerFactory<>(props);
 		}
 
 		@Bean
@@ -324,13 +320,15 @@ public class KafkaDslTests {
 
 		private KafkaProducerMessageHandlerSpec<Integer, String, ?> kafkaMessageHandler(
 				ProducerFactory<Integer, String> producerFactory, String topic) {
+
 			return Kafka
 					.outboundChannelAdapter(producerFactory)
+					.sync(true)
 					.messageKey(m -> m
 							.getHeaders()
 							.get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER))
 					.headerMapper(mapper())
-					.partitionId(m -> 10)
+					.partitionId(m -> 0)
 					.topicExpression("headers[kafka_topic] ?: '" + topic + "'")
 					.configureKafkaTemplate(t -> t.id("kafkaTemplate:" + topic));
 		}
@@ -353,6 +351,7 @@ public class KafkaDslTests {
 		public IntegrationFlow outboundGateFlow() {
 			return IntegrationFlows.from(Gate.class)
 					.handle(Kafka.outboundGateway(producerFactory(), replyContainer())
+							.sync(true)
 							.configureKafkaTemplate(t -> t.replyTimeout(30_000)))
 					.get();
 		}

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java

@@ -18,6 +18,7 @@ package org.springframework.integration.kafka.inbound;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyMap;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.BDDMockito.given;
@@ -461,7 +462,7 @@ public class MessageDrivenAdapterTests {
 		willAnswer(i -> {
 			commitLatch.countDown();
 			return null;
-		}).given(consumer).commitSync(any(Map.class));
+		}).given(consumer).commitSync(anyMap(), any());
 		given(consumer.assignment()).willReturn(records.keySet());
 		final CountDownLatch pauseLatch = new CountDownLatch(1);
 		willAnswer(i -> {
@@ -488,7 +489,7 @@ public class MessageDrivenAdapterTests {
 		adapter.afterPropertiesSet();
 		adapter.start();
 		assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
-		verify(consumer, times(2)).commitSync(any(Map.class));
+		verify(consumer, times(2)).commitSync(anyMap(), any());
 		assertThat(outputChannel.getQueueSize()).isEqualTo(2);
 		adapter.pause();
 		assertThat(pauseLatch.await(10, TimeUnit.SECONDS)).isTrue();

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -268,6 +268,7 @@ public class MessageSourceTests {
 				.getConfigurationProperties();
 		given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer);
 		KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo");
+		source.setCommitTimeout(Duration.ofSeconds(30));
 
 		Message<?> received = source.receive();
 		assertThat(received.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(0L);
@@ -292,11 +293,13 @@ public class MessageSourceTests {
 		inOrder.verify(consumer).poll(any(Duration.class));
 		inOrder.verify(consumer).seek(topicPartition, 0L); // rollback
 		inOrder.verify(consumer).poll(any(Duration.class));
-		inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(1L)));
+		inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(1L)),
+				Duration.ofSeconds(30));
 		inOrder.verify(consumer).poll(any(Duration.class));
 		inOrder.verify(consumer).seek(topicPartition, 1L); // rollback
 		inOrder.verify(consumer).poll(any(Duration.class));
-		inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(2L)));
+		inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(2L)),
+				Duration.ofSeconds(30));
 		inOrder.verify(consumer).poll(any(Duration.class));
 		inOrder.verify(consumer).close();
 		inOrder.verifyNoMoreInteractions();

spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt

@@ -25,8 +25,10 @@ import assertk.assertions.isNull
 import assertk.assertions.isSameAs
 import assertk.assertions.isTrue
 import assertk.catch
+import kafka.tools.ConsoleProducer
 import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener
+import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.common.TopicPartition
 import org.junit.ClassRule
 import org.junit.Test
@@ -80,6 +82,7 @@ import java.util.stream.Stream
 
 /**
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 3.0.3
  */
@@ -152,11 +155,6 @@ class KafkaDslKotlinTests {
     @Test
     fun testKafkaAdapters() {
 
-        val exception = catch { this.sendToKafkaFlowInput.send(GenericMessage("foo")) }
-        assertThat(exception!!.message).isNotNull().contains("10 is not in the range")
-
-        this.kafkaProducer1.setPartitionIdExpression(ValueExpression(0))
-        this.kafkaProducer2.setPartitionIdExpression(ValueExpression(0))
         this.sendToKafkaFlowInput.send(GenericMessage("foo", hashMapOf<String, Any>("foo" to "bar")))
 
         assertThat(TestUtils.getPropertyValue(this.kafkaProducer1, "headerMapper")).isSameAs(this.mapper)
@@ -280,7 +278,11 @@ class KafkaDslKotlinTests {
                         .get()
 
         @Bean
-        fun producerFactory() = DefaultKafkaProducerFactory<Int, String>(KafkaTestUtils.producerProps(embeddedKafka.embeddedKafka))
+        fun producerFactory(): DefaultKafkaProducerFactory<Int, String> {
+            val props = KafkaTestUtils.producerProps(embeddedKafka.embeddedKafka)
+            props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000")
+            return DefaultKafkaProducerFactory(props)
+        }
 
         @Bean
         fun sendToKafkaFlow() =
@@ -309,7 +311,8 @@ class KafkaDslKotlinTests {
                 Kafka.outboundChannelAdapter(producerFactory)
                         .messageKey<Any> { m -> m.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] }
                         .headerMapper(mapper())
-                        .partitionId<Any> { _ -> 10 }
+                        .sync(true)
+                        .partitionId<Any> { _ -> 0 }
                         .topicExpression("headers[kafka_topic] ?: '$topic'")
                         .configureKafkaTemplate { t -> t.id("kafkaTemplate:$topic") }
 
@@ -334,7 +337,8 @@ class KafkaDslKotlinTests {
         @Bean
         fun outboundGateFlow() =
                 IntegrationFlows.from(Gate::class.java)
-                        .handle(Kafka.outboundGateway(replyingKafkaTemplate()))
+                        .handle(Kafka.outboundGateway(replyingKafkaTemplate())
+                                .sync(true))
                         .get()
 
         private fun replyContainer(): GenericMessageListenerContainer<Int, String> {