From 1be6a6a3fa97cd2e399b772610a56b98a2cd0bbb Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 13 Apr 2018 21:01:46 -0400 Subject: [PATCH] Start version 3.1 for compatibility with SK-2.2 --- .../integration/kafka/dsl/Kafka.java | 2 +- .../KafkaMessageListenerContainerSpec.java | 21 +++++----- .../kafka/inbound/KafkaMessageSource.java | 42 +++++++++---------- .../main/resources/META-INF/spring.schemas | 4 +- ...0.xsd => spring-integration-kafka-3.1.xsd} | 0 ...rivenChannelAdapterParserTests-context.xml | 4 +- ...essageDrivenChannelAdapterParserTests.java | 10 ++--- .../integration/kafka/dsl/KafkaDslTests.java | 9 ++-- .../kafka/inbound/InboundGatewayTests.java | 2 +- .../inbound/MessageDrivenAdapterTests.java | 7 ++-- .../kafka/inbound/MessageSourceTests.java | 41 +++++++++--------- .../KafkaProducerMessageHandlerTests.java | 6 ++- 12 files changed, 74 insertions(+), 74 deletions(-) rename spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/{spring-integration-kafka-3.0.xsd => spring-integration-kafka-3.1.xsd} (100%) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java index 47a6ee3bad..562d1970f7 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java +++ b/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; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java index e3929be86a..9d2eb9946d 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java +++ b/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 * @return the spec. * @see ErrorHandler */ - public KafkaMessageListenerContainerSpec errorHandler(ErrorHandler errorHandler) { - this.target.getContainerProperties().setErrorHandler(errorHandler); + public KafkaMessageListenerContainerSpec errorHandler(GenericErrorHandler errorHandler) { + this.target.setGenericErrorHandler(errorHandler); return this; } @@ -105,11 +106,11 @@ public class KafkaMessageListenerContainerSpec *
  • MANUAL: Listener is responsible for acking - use a * {@link AcknowledgingMessageListener}. * - * @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 ackMode(AbstractMessageListenerContainer.AckMode ackMode) { + public KafkaMessageListenerContainerSpec ackMode(ContainerProperties.AckMode ackMode) { this.target.getContainerProperties().setAckMode(ackMode); return this; } @@ -127,8 +128,8 @@ public class KafkaMessageListenerContainerSpec /** * 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 /** * 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. diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java index 5a0bc32601..5ca3a8d003 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java +++ b/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 extends AbstractMessageSource } 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 configs = new HashMap<>(suppliedConsumerFactory.getConfigurationProperties()); configs.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1); @@ -349,7 +351,6 @@ public class KafkaMessageSource extends AbstractMessageSource /** * AcknowledgmentCallbackFactory for KafkaAckInfo. - * * @param the key type. * @param the value type. * @@ -365,7 +366,6 @@ public class KafkaMessageSource extends AbstractMessageSource /** * AcknowledgmentCallback for Kafka. - * * @param the key type. * @param the value type. * @@ -395,15 +395,15 @@ public class KafkaMessageSource extends AbstractMessageSource try { ConsumerRecord 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 extends AbstractMessageSource if (inflight.size() > 1) { List 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"); diff --git a/spring-integration-kafka/src/main/resources/META-INF/spring.schemas b/spring-integration-kafka/src/main/resources/META-INF/spring.schemas index 4f05d6cf4b..6e5474a1f4 100644 --- a/spring-integration-kafka/src/main/resources/META-INF/spring.schemas +++ b/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 diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.0.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.1.xsd similarity index 100% rename from spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.0.xsd rename to spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.1.xsd diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml index 460e80dc53..5b8088848f 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml @@ -50,7 +50,7 @@ - + @@ -67,7 +67,7 @@ - + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java index 020f1cb53b..a409472373 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java +++ b/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 cf = - new DefaultKafkaConsumerFactory<>(Collections.emptyMap()); + new DefaultKafkaConsumerFactory<>(Collections.emptyMap()); ContainerProperties containerProps = new ContainerProperties("foo"); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index 6509849a13..deb65db98c 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java +++ b/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(); } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java index 7b5bb6bba2..e95c5fa419 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java +++ b/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; diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java index 6129d8a744..84bae87ebd 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java +++ b/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 container = diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java index 638ec6e61e..b6b62f3112 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java +++ b/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(); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java index eb63dbcd4b..8612923a6d 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java +++ b/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 container = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties); + KafkaMessageListenerContainer container = + new KafkaMessageListenerContainer<>(consumerFactory, containerProperties); + ProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( KafkaTestUtils.producerProps(embeddedKafka)); ReplyingKafkaTemplate template = new ReplyingKafkaTemplate<>(producerFactory, container);