Start version 3.1 for compatibility with SK-2.2

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java

@@ -26,8 +26,8 @@ import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
+import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.GenericMessageListenerContainer;
-import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
 

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java

@@ -27,8 +27,9 @@ import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
 import org.springframework.kafka.listener.AcknowledgingMessageListener;
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
+import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.ErrorHandler;
-import org.springframework.kafka.listener.config.ContainerProperties;
+import org.springframework.kafka.listener.GenericErrorHandler;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
 
 /**
@@ -88,8 +89,8 @@ public class KafkaMessageListenerContainerSpec<K, V>
 	 * @return the spec.
 	 * @see ErrorHandler
 	 */
-	public KafkaMessageListenerContainerSpec<K, V> errorHandler(ErrorHandler errorHandler) {
-		this.target.getContainerProperties().setErrorHandler(errorHandler);
+	public KafkaMessageListenerContainerSpec<K, V> errorHandler(GenericErrorHandler<?> errorHandler) {
+		this.target.setGenericErrorHandler(errorHandler);
 		return this;
 	}
 
@@ -105,11 +106,11 @@ public class KafkaMessageListenerContainerSpec<K, V>
 	 * <li>MANUAL: Listener is responsible for acking - use a
 	 * {@link AcknowledgingMessageListener}.
 	 * </ul>
-	 * @param ackMode the {@link AbstractMessageListenerContainer.AckMode}; default BATCH.
+	 * @param ackMode the {@link ContainerProperties.AckMode}; default BATCH.
 	 * @return the spec.
-	 * @see AbstractMessageListenerContainer.AckMode
+	 * @see ContainerProperties.AckMode
 	 */
-	public KafkaMessageListenerContainerSpec<K, V> ackMode(AbstractMessageListenerContainer.AckMode ackMode) {
+	public KafkaMessageListenerContainerSpec<K, V> ackMode(ContainerProperties.AckMode ackMode) {
 		this.target.getContainerProperties().setAckMode(ackMode);
 		return this;
 	}
@@ -127,8 +128,8 @@ public class KafkaMessageListenerContainerSpec<K, V>
 
 	/**
 	 * Set the number of outstanding record count after which offsets should be
-	 * committed when {@link AbstractMessageListenerContainer.AckMode#COUNT}
-	 * or {@link AbstractMessageListenerContainer.AckMode#COUNT_TIME} is being used.
+	 * committed when {@link ContainerProperties.AckMode#COUNT}
+	 * or {@link ContainerProperties.AckMode#COUNT_TIME} is being used.
 	 * @param count the count
 	 * @return the spec.
 	 * @see ContainerProperties#setAckCount(int)
@@ -140,8 +141,8 @@ public class KafkaMessageListenerContainerSpec<K, V>
 
 	/**
 	 * Set the time (ms) after which outstanding offsets should be committed when
-	 * {@link AbstractMessageListenerContainer.AckMode#TIME} or
-	 * {@link AbstractMessageListenerContainer.AckMode#COUNT_TIME} is being used.
+	 * {@link ContainerProperties.AckMode#TIME} or
+	 * {@link ContainerProperties.AckMode#COUNT_TIME} is being used.
 	 * Should be larger than zero.
 	 * @param millis the time
 	 * @return the spec.

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

@@ -43,10 +43,10 @@ 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;
+import org.springframework.integration.acks.AcknowledgmentCallbackFactory;
 import org.springframework.integration.endpoint.AbstractMessageSource;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
-import org.springframework.integration.support.AcknowledgmentCallback;
-import org.springframework.integration.support.AcknowledgmentCallbackFactory;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.support.Acknowledgment;
@@ -73,6 +73,8 @@ import org.springframework.util.Assert;
  *
  * @author Gary Russell
  * @author Mark Norkin
+ * @author Artem Bilan
+ *
  * @since 3.0.1
  *
  */
@@ -229,8 +231,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
 			}
 			if (this.logger.isWarnEnabled()) {
 				this.logger.warn("'" + ConsumerConfig.MAX_POLL_RECORDS_CONFIG
-					+ "' has been forced to from " + (maxPoll == null ? "unspecified" : maxPoll)
-					+ "to 1, to avoid having to seek after each record");
+						+ "' has been forced to from " + (maxPoll == null ? "unspecified" : maxPoll)
+						+ "to 1, to avoid having to seek after each record");
 			}
 			Map<String, Object> configs = new HashMap<>(suppliedConsumerFactory.getConfigurationProperties());
 			configs.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
@@ -349,7 +351,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
 
 	/**
 	 * AcknowledgmentCallbackFactory for KafkaAckInfo.
-	 *
 	 * @param <K> the key type.
 	 * @param <V> the value type.
 	 *
@@ -365,7 +366,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
 
 	/**
 	 * AcknowledgmentCallback for Kafka.
-	 *
 	 * @param <K> the key type.
 	 * @param <V> the value type.
 	 *
@@ -395,15 +395,15 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
 				try {
 					ConsumerRecord<K, V> record = this.ackInfo.getRecord();
 					switch (status) {
-					case ACCEPT:
-					case REJECT:
-						commitIfPossible(record);
-						break;
-					case REQUEUE:
-						rollback(record);
-						break;
-					default:
-						break;
+						case ACCEPT:
+						case REJECT:
+							commitIfPossible(record);
+							break;
+						case REQUEUE:
+							rollback(record);
+							break;
+						default:
+							break;
 					}
 				}
 				catch (WakeupException e) {
@@ -424,12 +424,12 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
 			if (inflight.size() > 1) {
 				List<Long> rewound =
 						inflight.stream()
-							.filter(i -> i.getRecord().offset() > record.offset())
-							.map(i -> {
-								i.setRolledBack(true);
-								return i.getRecord().offset();
-							})
-							.collect(Collectors.toList());
+								.filter(i -> i.getRecord().offset() > record.offset())
+								.map(i -> {
+									i.setRolledBack(true);
+									return i.getRecord().offset();
+								})
+								.collect(Collectors.toList());
 				if (rewound.size() > 0 && this.logger.isWarnEnabled()) {
 					this.logger.warn("Rolled back " + record + " later in-flight offsets "
 							+ rewound + " will also be re-fetched");

spring-integration-kafka/src/main/resources/META-INF/spring.schemas

@@ -1,2 +1,2 @@
-http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka-3.0.xsd=org/springframework/integration/kafka/config/spring-integration-kafka-3.0.xsd
-http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd=org/springframework/integration/kafka/config/spring-integration-kafka-3.0.xsd
+http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka-3.1.xsd=org/springframework/integration/kafka/config/spring-integration-kafka-3.1.xsd
+http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd=org/springframework/integration/kafka/config/spring-integration-kafka-3.1.xsd

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml

@@ -50,7 +50,7 @@
 			</bean>
 		</constructor-arg>
 		<constructor-arg>
-			<bean class="org.springframework.kafka.listener.config.ContainerProperties">
+			<bean class="org.springframework.kafka.listener.ContainerProperties">
 				<constructor-arg name="topics" value="foo" />
 			</bean>
 		</constructor-arg>
@@ -67,7 +67,7 @@
 			</bean>
 		</constructor-arg>
 		<constructor-arg>
-			<bean class="org.springframework.kafka.listener.config.ContainerProperties">
+			<bean class="org.springframework.kafka.listener.ContainerProperties">
 				<constructor-arg name="topics" value="foo" />
 			</bean>
 		</constructor-arg>

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -33,11 +33,11 @@ import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAd
 import org.springframework.integration.support.ErrorMessageStrategy;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.KafkaMessageListenerContainer;
 import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
 import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter;
-import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.retry.RecoveryCallback;
 import org.springframework.retry.support.RetryTemplate;
 import org.springframework.test.context.ContextConfiguration;
@@ -73,7 +73,7 @@ public class KafkaMessageDrivenChannelAdapterParserTests {
 	private RecoveryCallback<?> recoveryCallback;
 
 	@Test
-	public void testKafkaMessageDrivenChannelAdapterParser() throws Exception {
+	public void testKafkaMessageDrivenChannelAdapterParser() {
 		assertThat(this.kafkaListener.isAutoStartup()).isFalse();
 		assertThat(this.kafkaListener.isRunning()).isFalse();
 		assertThat(this.kafkaListener.getPhase()).isEqualTo(100);
@@ -94,7 +94,7 @@ public class KafkaMessageDrivenChannelAdapterParserTests {
 	}
 
 	@Test
-	public void testKafkaBatchMessageDrivenChannelAdapterParser() throws Exception {
+	public void testKafkaBatchMessageDrivenChannelAdapterParser() {
 		assertThat(this.kafkaBatchListener.isAutoStartup()).isFalse();
 		assertThat(this.kafkaBatchListener.isRunning()).isFalse();
 		assertThat(this.kafkaBatchListener.getPhase()).isEqualTo(100);
@@ -112,7 +112,7 @@ public class KafkaMessageDrivenChannelAdapterParserTests {
 	@SuppressWarnings("unchecked")
 	public void testKafkaMessageDrivenChannelAdapterOptions() {
 		DefaultKafkaConsumerFactory<Integer, String> cf =
-				new DefaultKafkaConsumerFactory<>(Collections.<String, Object>emptyMap());
+				new DefaultKafkaConsumerFactory<>(Collections.emptyMap());
 		ContainerProperties containerProps = new ContainerProperties("foo");
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);

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

@@ -57,11 +57,10 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
-import org.springframework.kafka.listener.AbstractMessageListenerContainer;
+import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.GenericMessageListenerContainer;
 import org.springframework.kafka.listener.KafkaMessageListenerContainer;
 import org.springframework.kafka.listener.MessageListenerContainer;
-import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
 import org.springframework.kafka.support.KafkaHeaders;
@@ -246,7 +245,7 @@ public class KafkaDslTests {
 					.from(Kafka.messageDrivenChannelAdapter(consumerFactory(),
 							KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1)
 							.configureListenerContainer(c ->
-									c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL)
+									c.ackMode(ContainerProperties.AckMode.MANUAL)
 											.id("topic1ListenerContainer"))
 							.recoveryCallback(new ErrorMessageSendingRecoverer(errorChannel(),
 									new RawRecordHeaderErrorMessageStrategy()))
@@ -266,7 +265,7 @@ public class KafkaDslTests {
 					.from(Kafka.messageDrivenChannelAdapter(consumerFactory(),
 							KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2)
 							.configureListenerContainer(c ->
-									c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL))
+									c.ackMode(ContainerProperties.AckMode.MANUAL))
 							.recoveryCallback(new ErrorMessageSendingRecoverer(errorChannel(),
 									new RawRecordHeaderErrorMessageStrategy()))
 							.retryTemplate(new RetryTemplate())
@@ -336,7 +335,7 @@ public class KafkaDslTests {
 		public IntegrationFlow outboundGateFlow() {
 			return IntegrationFlows.from(Gate.class)
 					.handle(Kafka.outboundGateway(producerFactory(), replyContainer())
-								.configureKafkaTemplate(t -> t.replyTimeout(30_000)))
+							.configureKafkaTemplate(t -> t.replyTimeout(30_000)))
 					.get();
 		}
 

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

@@ -43,8 +43,8 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.KafkaMessageListenerContainer;
-import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.converter.MessagingMessageConverter;

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

@@ -48,21 +48,20 @@ import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.junit.ClassRule;
 import org.junit.Test;
 
+import org.springframework.integration.StaticMessageHeaderAccessor;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
 import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.ListenerMode;
 import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
 import org.springframework.integration.support.MessageBuilder;
-import org.springframework.integration.support.StaticMessageHeaderAccessor;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
-import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
+import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.KafkaMessageListenerContainer;
-import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
 import org.springframework.kafka.support.KafkaHeaders;
@@ -466,7 +465,7 @@ public class MessageDrivenAdapterTests {
 		TopicPartitionInitialOffset[] topicPartition = new TopicPartitionInitialOffset[] {
 				new TopicPartitionInitialOffset("foo", 0) };
 		ContainerProperties containerProps = new ContainerProperties(topicPartition);
-		containerProps.setAckMode(AckMode.RECORD);
+		containerProps.setAckMode(ContainerProperties.AckMode.RECORD);
 		containerProps.setClientId("clientId");
 		containerProps.setIdleEventInterval(100L);
 		KafkaMessageListenerContainer<Integer, String> container =

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

@@ -55,9 +55,8 @@ import org.mockito.ArgumentCaptor;
 import org.mockito.InOrder;
 
 import org.springframework.beans.DirectFieldAccessor;
-import org.springframework.integration.support.AcknowledgmentCallback;
-import org.springframework.integration.support.AcknowledgmentCallback.Status;
-import org.springframework.integration.support.StaticMessageHeaderAccessor;
+import org.springframework.integration.StaticMessageHeaderAccessor;
+import org.springframework.integration.acks.AcknowledgmentCallback;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
@@ -118,19 +117,19 @@ public class MessageSourceTests {
 		assertThat(received).isNotNull();
 		assertThat(received.getHeaders().get(KafkaHeaders.RAW_DATA)).isInstanceOf(ConsumerRecord.class);
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received)
-				.acknowledge(Status.ACCEPT);
+				.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
 		received = source.receive();
 		assertThat(received).isNotNull();
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received)
-				.acknowledge(Status.ACCEPT);
+				.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
 		received = source.receive();
 		assertThat(received).isNotNull();
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received)
-				.acknowledge(Status.ACCEPT);
+				.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
 		received = source.receive();
 		assertThat(received).isNotNull();
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received)
-				.acknowledge(Status.ACCEPT);
+				.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
 		received = source.receive();
 		assertThat(received).isNull();
 		source.destroy();
@@ -209,17 +208,17 @@ public class MessageSourceTests {
 		consumer.paused(); // need some other interaction with mock between polls for InOrder
 		Message<?> received6 = source.receive();
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received3)
-				.acknowledge(Status.ACCEPT);
+				.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received2)
-				.acknowledge(Status.ACCEPT);
+				.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received5)
-				.acknowledge(Status.ACCEPT);
+				.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received1)
-				.acknowledge(Status.ACCEPT); // should commit offset 3 (received 3)
+				.acknowledge(AcknowledgmentCallback.Status.ACCEPT); // should commit offset 3 (received 3)
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received6)
-				.acknowledge(Status.ACCEPT);
+				.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received4)
-				.acknowledge(Status.ACCEPT); // should commit offset 6 (received 6).
+				.acknowledge(AcknowledgmentCallback.Status.ACCEPT); // should commit offset 6 (received 6).
 		assertThat(source.receive()).isNull();
 		source.destroy();
 		InOrder inOrder = inOrder(consumer);
@@ -274,19 +273,19 @@ public class MessageSourceTests {
 		Message<?> received = source.receive();
 		assertThat(received.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(0L);
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received)
-				.acknowledge(Status.REQUEUE);
+				.acknowledge(AcknowledgmentCallback.Status.REQUEUE);
 		received = source.receive();
 		assertThat(received.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(0L);
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received)
-				.acknowledge(Status.ACCEPT);
+				.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
 		received = source.receive();
 		assertThat(received.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(1L);
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received)
-				.acknowledge(Status.REQUEUE);
+				.acknowledge(AcknowledgmentCallback.Status.REQUEUE);
 		received = source.receive();
 		assertThat(received.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(1L);
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received)
-				.acknowledge(Status.ACCEPT);
+				.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
 		received = source.receive();
 		source.destroy();
 		assertThat(received).isNull();
@@ -345,22 +344,22 @@ public class MessageSourceTests {
 		new DirectFieldAccessor(ack1).setPropertyValue("logger", log1);
 		given(log1.isWarnEnabled()).willReturn(true);
 		willDoNothing().given(log1).warn(any());
-		ack1.acknowledge(Status.REQUEUE);
+		ack1.acknowledge(AcknowledgmentCallback.Status.REQUEUE);
 		assertThat(received2.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(1L);
 		AcknowledgmentCallback ack2 = StaticMessageHeaderAccessor.getAcknowledgmentCallback(received2);
 		Log log2 = spy(KafkaTestUtils.getPropertyValue(ack1, "logger", Log.class));
 		new DirectFieldAccessor(ack2).setPropertyValue("logger", log2);
 		given(log2.isWarnEnabled()).willReturn(true);
 		willDoNothing().given(log2).warn(any());
-		ack2.acknowledge(Status.ACCEPT);
+		ack2.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
 		received1 = source.receive();
 		assertThat(received1.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(0L);
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received1)
-				.acknowledge(Status.ACCEPT);
+				.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
 		received2 = source.receive();
 		assertThat(received2.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(1L);
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received2)
-				.acknowledge(Status.ACCEPT);
+				.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
 		received1 = source.receive();
 		source.destroy();
 		assertThat(received1).isNull();

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java

@@ -61,8 +61,8 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.KafkaMessageListenerContainer;
-import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
 import org.springframework.kafka.support.KafkaHeaders;
@@ -315,7 +315,9 @@ public class KafkaProducerMessageHandlerTests {
 			}
 
 		});
-		KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);
+
 		ProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
 				KafkaTestUtils.producerProps(embeddedKafka));
 		ReplyingKafkaTemplate<Integer, String, String> template = new ReplyingKafkaTemplate<>(producerFactory, container);