diff --git a/spring-integration-kafka/src/checkstyle/checkstyle.xml b/spring-integration-kafka/src/checkstyle/checkstyle.xml index 63332c1d04..14b024b0bb 100644 --- a/spring-integration-kafka/src/checkstyle/checkstyle.xml +++ b/spring-integration-kafka/src/checkstyle/checkstyle.xml @@ -84,7 +84,7 @@ org.hamcrest.Matchers.*, org.mockito.Mockito.*, org.mockito.BDDMockito.*, - org.mockito.Matchers.*, + org.mockito.ArgumentMatchers.*, org.springframework.kafka.test.hamcrest.KafkaMatchers.*, org.springframework.kafka.test.assertj.KafkaConditions.*" /> 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 4e54befc21..fc7a17fb9a 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 @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-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. @@ -21,6 +21,7 @@ import java.util.regex.Pattern; import org.apache.kafka.common.TopicPartition; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; +import org.springframework.integration.kafka.inbound.KafkaMessageSource.KafkaAckCallbackFactory; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; @@ -47,7 +48,8 @@ public final class Kafka { * @return the KafkaProducerMessageHandlerSpec. */ public static > KafkaProducerMessageHandlerSpec - outboundChannelAdapter(KafkaTemplate kafkaTemplate) { + outboundChannelAdapter(KafkaTemplate kafkaTemplate) { + return new KafkaProducerMessageHandlerSpec<>(kafkaTemplate); } @@ -60,10 +62,45 @@ public final class Kafka { * @see Kafka Producer Configs */ public static KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec - outboundChannelAdapter(ProducerFactory producerFactory) { + outboundChannelAdapter(ProducerFactory producerFactory) { + return new KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec<>(producerFactory); } + /** + * Create an initial {@link KafkaInboundChannelAdapterSpec} with the consumer factory and + * topics. + * @param consumerFactory the consumer factory. + * @param topics the topic(s). + * @param the Kafka message key type. + * @param the Kafka message value type. + * @return the spec. + * @since 3.0.1 + */ + public static KafkaInboundChannelAdapterSpec + inboundChannelAdapter(ConsumerFactory consumerFactory, String... topics) { + + return new KafkaInboundChannelAdapterSpec<>(consumerFactory, topics); + } + + /** + * Create an initial {@link KafkaInboundChannelAdapterSpec} with the consumer factory and + * topics with a custom ack callback factory. + * @param consumerFactory the consumer factory. + * @param ackCallbackFactory the callback factory. + * @param topics the topic(s). + * @param the Kafka message key type. + * @param the Kafka message value type. + * @return the spec. + * @since 3.0.1 + */ + public static KafkaInboundChannelAdapterSpec + inboundChannelAdapter(ConsumerFactory consumerFactory, + KafkaAckCallbackFactory ackCallbackFactory, String... topics) { + + return new KafkaInboundChannelAdapterSpec<>(consumerFactory, ackCallbackFactory, topics); + } + /** * Create an initial {@link KafkaMessageDrivenChannelAdapterSpec}. * @param listenerContainer the {@link AbstractMessageListenerContainer}. @@ -73,8 +110,9 @@ public final class Kafka { * @return the KafkaMessageDrivenChannelAdapterSpec. */ public static > - KafkaMessageDrivenChannelAdapterSpec messageDrivenChannelAdapter( + KafkaMessageDrivenChannelAdapterSpec messageDrivenChannelAdapter( AbstractMessageListenerContainer listenerContainer) { + return messageDrivenChannelAdapter(listenerContainer, KafkaMessageDrivenChannelAdapter.ListenerMode.record); } @@ -88,9 +126,10 @@ public final class Kafka { * @return the KafkaMessageDrivenChannelAdapterSpec. */ public static > - KafkaMessageDrivenChannelAdapterSpec messageDrivenChannelAdapter( + KafkaMessageDrivenChannelAdapterSpec messageDrivenChannelAdapter( AbstractMessageListenerContainer listenerContainer, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) { + return new KafkaMessageDrivenChannelAdapterSpec<>(listenerContainer, listenerMode); } @@ -105,7 +144,8 @@ public final class Kafka { */ public static KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, ContainerProperties containerProperties) { + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, ContainerProperties containerProperties) { + return messageDrivenChannelAdapter(consumerFactory, containerProperties, KafkaMessageDrivenChannelAdapter.ListenerMode.record); } @@ -122,8 +162,9 @@ public final class Kafka { */ public static KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, ContainerProperties containerProperties, + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, ContainerProperties containerProperties, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) { + return messageDrivenChannelAdapter( new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory, containerProperties), listenerMode); @@ -140,8 +181,9 @@ public final class Kafka { */ public static KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, TopicPartitionInitialOffset... topicPartitions) { + return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record, topicPartitions); } @@ -158,9 +200,10 @@ public final class Kafka { */ public static KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, TopicPartitionInitialOffset... topicPartitions) { + return messageDrivenChannelAdapter( new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory, topicPartitions), listenerMode); @@ -177,7 +220,8 @@ public final class Kafka { */ public static KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, String... topics) { + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, String... topics) { + return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record, topics); } @@ -194,8 +238,9 @@ public final class Kafka { */ public static KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, String... topics) { + return messageDrivenChannelAdapter( new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory, topics), listenerMode); @@ -212,7 +257,8 @@ public final class Kafka { */ public static KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, Pattern topicPattern) { + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, Pattern topicPattern) { + return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record, topicPattern); } @@ -229,8 +275,9 @@ public final class Kafka { */ public static KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(ConsumerFactory consumerFactory, + messageDrivenChannelAdapter(ConsumerFactory consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, Pattern topicPattern) { + return messageDrivenChannelAdapter( new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory, topicPattern), @@ -239,8 +286,9 @@ public final class Kafka { private static KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec - messageDrivenChannelAdapter(KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec spec, + messageDrivenChannelAdapter(KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec spec, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) { + return new KafkaMessageDrivenChannelAdapterSpec .KafkaMessageDrivenChannelAdapterListenerContainerSpec<>(spec, listenerMode); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java new file mode 100644 index 0000000000..0c075d05cd --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java @@ -0,0 +1,76 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.dsl; + +import java.lang.reflect.Type; + +import org.springframework.integration.dsl.MessageSourceSpec; +import org.springframework.integration.kafka.inbound.KafkaMessageSource; +import org.springframework.integration.kafka.inbound.KafkaMessageSource.KafkaAckCallbackFactory; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.support.converter.RecordMessageConverter; + +/** + * Spec for a polled Kafka inbound channel adapter. + * + * @param the key type. + * @param the value type. + * + * @author Gary Russell + * + * @since 3.0.1 + * + */ +public class KafkaInboundChannelAdapterSpec + extends MessageSourceSpec, KafkaMessageSource> { + + KafkaInboundChannelAdapterSpec(ConsumerFactory consumerFactory, String... topics) { + this.target = new KafkaMessageSource<>(consumerFactory, topics); + } + + KafkaInboundChannelAdapterSpec(ConsumerFactory consumerFactory, + KafkaAckCallbackFactory ackCallbackFactory, String... topics) { + + this.target = new KafkaMessageSource<>(consumerFactory, ackCallbackFactory, topics); + } + + public KafkaInboundChannelAdapterSpec groupId(String groupId) { + this.target.setGroupId(groupId); + return this; + } + + public KafkaInboundChannelAdapterSpec clientId(String clientId) { + this.target.setClientId(clientId); + return this; + } + + public KafkaInboundChannelAdapterSpec pollTimeout(long pollTimeout) { + this.target.setPollTimeout(pollTimeout); + return this; + } + + public KafkaInboundChannelAdapterSpec messageConverter(RecordMessageConverter messageConverter) { + this.target.setMessageConverter(messageConverter); + return this; + } + + public KafkaInboundChannelAdapterSpec payloadType(Type type) { + this.target.setPayloadType(type); + return this; + } + +} 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 new file mode 100644 index 0000000000..963c18b088 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java @@ -0,0 +1,536 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.inbound; + +import java.lang.reflect.Type; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +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.common.TopicPartition; +import org.apache.kafka.common.errors.WakeupException; + +import org.springframework.beans.factory.DisposableBean; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.endpoint.AbstractMessageSource; +import org.springframework.integration.support.AcknowledgmentCallback; +import org.springframework.integration.support.AcknowledgmentCallbackFactory; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.kafka.support.converter.KafkaMessageHeaders; +import org.springframework.kafka.support.converter.MessagingMessageConverter; +import org.springframework.kafka.support.converter.RecordMessageConverter; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; + +/** + * Polled message source for kafka. Only one thread can poll for data (or + * acknowledge a message) at a time. + *

+ * NOTE: If the application acknowledges messages out of order, the acks + * will be deferred until all messages prior to the offset are ack'd. + * If multiple records are retrieved and an earlier offset is requeued, records + * from the subsequent offsets will be redelivered - even if they were + * processed successfully. Applications should therefore implement + * idempotency. + * + * @param the key type. + * @param the value type. + * + * @author Gary Russell + * @since 3.0.1 + * + */ +public class KafkaMessageSource extends AbstractMessageSource + implements DisposableBean { + + private static final long DEFAULT_POLL_TIMEOUT = 50L; + + private final ConsumerFactory consumerFactory; + + private final KafkaAckCallbackFactory ackCallbackFactory; + + private final String[] topics; + + private final Object consumerMonitor = new Object(); + + private final Map>> inflightRecords = new HashMap<>(); + + private String groupId; + + private String clientId = "message.source"; + + private long pollTimeout = DEFAULT_POLL_TIMEOUT; + + private RecordMessageConverter messageConverter = new MessagingMessageConverter(); + + private Type payloadType; + + private volatile Consumer consumer; + + private volatile Collection partitions; + + public KafkaMessageSource(ConsumerFactory consumerFactory, String... topics) { + this(consumerFactory, new KafkaAckCallbackFactory<>(), topics); + } + + public KafkaMessageSource(ConsumerFactory consumerFactory, + KafkaAckCallbackFactory ackCallbackFactory, String... topics) { + + Assert.notNull(consumerFactory, "'consumerFactory' must not be null"); + Assert.notNull(ackCallbackFactory, "'ackCallbackFactory' must not be null"); + this.consumerFactory = consumerFactory; + this.ackCallbackFactory = ackCallbackFactory; + this.topics = topics; + Object maxPoll = consumerFactory.getConfigurationProperties().get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG); + if (maxPoll == null || (maxPoll instanceof Number && ((Number) maxPoll).intValue() != 1) + || (maxPoll instanceof String && Integer.parseInt((String) maxPoll) != 1)) { + if (this.logger.isWarnEnabled()) { + this.logger.warn("It is advisable to set " + ConsumerConfig.MAX_POLL_RECORDS_CONFIG + + " to 1 to avoid having to seek after each record"); + } + } + } + + protected String getGroupId() { + return this.groupId; + } + + /** + * Set the group.id property for the consumer. + * @param groupId the group id. + */ + public void setGroupId(String groupId) { + this.groupId = groupId; + } + + protected String getClientId() { + return this.clientId; + } + + /** + * Set the client.id property for the consumer. + * @param clientId the client id. + */ + public void setClientId(String clientId) { + this.clientId = clientId; + } + + protected long getPollTimeout() { + return this.pollTimeout; + } + + /** + * Set the pollTimeout for the poll() operations; default 50ms. + * @param pollTimeout the poll timeout. + */ + public void setPollTimeout(long pollTimeout) { + this.pollTimeout = pollTimeout; + } + + protected RecordMessageConverter getMessageConverter() { + return this.messageConverter; + } + + /** + * Set the message converter to replace the default + * {@link MessagingMessageConverter}. + * @param messageConverter the converter. + */ + public void setMessageConverter(RecordMessageConverter messageConverter) { + this.messageConverter = messageConverter; + } + + protected Type getPayloadType() { + return this.payloadType; + } + + /** + * Set the payload type. + * Only applies if a type-aware message converter is provided. + * @param payloadType the type to convert to. + */ + public void setPayloadType(Type payloadType) { + this.payloadType = payloadType; + } + + @Override + public String getComponentType() { + return "kafka:message-source"; + } + + @Override + protected synchronized Object doReceive() { + if (this.consumer == null) { + createConsumer(); + } + ConsumerRecord record; + TopicPartition topicPartition; + synchronized (this.consumerMonitor) { + Set paused = this.consumer.paused(); + if (paused.size() > 0) { + this.consumer.resume(paused); + } + ConsumerRecords records = this.consumer.poll(this.pollTimeout); + this.consumer.pause(this.partitions); + if (records == null || records.count() == 0) { + return null; + } + record = records.iterator().next(); + topicPartition = new TopicPartition(record.topic(), record.partition()); + if (records.count() > 1) { + this.consumer.seek(topicPartition, record.offset() + 1); + } + } + KafkaAckInfo ackInfo = new KafkaAckInfoImpl(record, topicPartition); + AcknowledgmentCallback ackCallback = this.ackCallbackFactory.createCallback(ackInfo); + this.inflightRecords.computeIfAbsent(topicPartition, tp -> new TreeSet<>()).add(ackInfo); + Message message = this.messageConverter.toMessage(record, + ackCallback instanceof Acknowledgment ? (Acknowledgment) ackCallback : null, this.consumer, + this.payloadType); + if (message.getHeaders() instanceof KafkaMessageHeaders) { + Map rawHeaders = ((KafkaMessageHeaders) message.getHeaders()).getRawHeaders(); + rawHeaders.put(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, ackCallback); + return message; + } + else { + return getMessageBuilderFactory().fromMessage(message) + .setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, ackCallback); + } + } + + protected void createConsumer() { + this.consumer = this.consumerFactory.createConsumer(this.groupId, this.clientId, null); + synchronized (this.consumerMonitor) { + this.consumer.subscribe(Arrays.asList(this.topics), new ConsumerRebalanceListener() { + + @Override + public void onPartitionsRevoked(Collection partitions) { + KafkaMessageSource.this.partitions = Collections.emptyList(); + } + + @Override + public void onPartitionsAssigned(Collection partitions) { + KafkaMessageSource.this.partitions = new ArrayList<>(partitions); + } + + }); + } + } + + @Override + public synchronized void destroy() { + if (this.consumer != null) { + Consumer consumer2 = this.consumer; + this.consumer = null; + synchronized (this.consumerMonitor) { + consumer2.close(30, TimeUnit.SECONDS); + } + } + } + + /** + * AcknowledgmentCallbackFactory for KafkaAckInfo. + * + * @param the key type. + * @param the value type. + * + */ + public static class KafkaAckCallbackFactory implements AcknowledgmentCallbackFactory> { + + @Override + public AcknowledgmentCallback createCallback(KafkaAckInfo info) { + return new KafkaAckCallback<>(info); + } + + } + + /** + * AcknowledgmentCallback for Kafka. + * + * @param the key type. + * @param the value type. + * + */ + public static class KafkaAckCallback implements AcknowledgmentCallback, Acknowledgment { + + private final Log logger = LogFactory.getLog(getClass()); + + private final KafkaAckInfo ackInfo; + + private volatile boolean acknowledged; + + private boolean autoAckEnabled = true; + + public KafkaAckCallback(KafkaAckInfo ackInfo) { + Assert.notNull(ackInfo, "'ackInfo' cannot be null"); + this.ackInfo = ackInfo; + } + + @Override + public void acknowledge(Status status) { + Assert.notNull(status, "'status' cannot be null"); + if (this.acknowledged) { + throw new IllegalStateException("Already acknowledged"); + } + synchronized (this.ackInfo.getConsumerMonitor()) { + try { + ConsumerRecord record = this.ackInfo.getRecord(); + switch (status) { + case ACCEPT: + case REJECT: + commitIfPossible(record); + break; + case REQUEUE: + rollback(record); + break; + default: + break; + } + } + catch (WakeupException e) { + throw new IllegalStateException(e); + } + finally { + this.acknowledged = true; + if (!this.ackInfo.isAckDeferred()) { + this.ackInfo.getOffsets().get(this.ackInfo.getTopicPartition()).remove(this.ackInfo); + } + } + } + } + + private void rollback(ConsumerRecord record) { + this.ackInfo.getConsumer().seek(this.ackInfo.getTopicPartition(), record.offset()); + Set> inflight = this.ackInfo.getOffsets().get(this.ackInfo.getTopicPartition()); + 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()); + if (rewound.size() > 0 && this.logger.isWarnEnabled()) { + this.logger.warn("Rolled back " + record + " later in-flight offsets " + + rewound + " will also be re-fetched"); + } + } + } + + private void commitIfPossible(ConsumerRecord record) { + if (this.ackInfo.isRolledBack()) { + if (this.logger.isWarnEnabled()) { + this.logger.warn("Cannot commit offset for " + record + + "; an earlier offset was rolled back"); + } + } + else { + Set> candidates = this.ackInfo.getOffsets().get(this.ackInfo.getTopicPartition()); + KafkaAckInfo ackInfo = null; + if (candidates.iterator().next().equals(this.ackInfo)) { + // see if there are any pending acks for higher offsets + List> toCommit = new ArrayList<>(); + for (KafkaAckInfo info : candidates) { + if (info != this.ackInfo) { + if (info.isAckDeferred()) { + toCommit.add(info); + } + else { + break; + } + } + } + if (toCommit.size() > 0) { + ackInfo = toCommit.get(toCommit.size() - 1); + if (this.logger.isDebugEnabled()) { + this.logger.debug("Committing pending offsets for " + record + " and all deferred to " + + ackInfo.getRecord()); + } + candidates.removeAll(toCommit); + } + else { + ackInfo = this.ackInfo; + } + } + else { // earlier offsets present + this.ackInfo.setAckDeferred(true); + } + if (ackInfo != null) { + ackInfo.getConsumer().commitSync(Collections.singletonMap(ackInfo.getTopicPartition(), + new OffsetAndMetadata(ackInfo.getRecord().offset() + 1))); + } + else { + if (this.logger.isDebugEnabled()) { + this.logger.debug("Deferring commit offset; earlier messages are in flight."); + } + } + } + } + + @Override + public boolean isAcknowledged() { + return this.acknowledged; + } + + @Override + public void acknowledge() { + acknowledge(Status.ACCEPT); + } + + @Override + public void noAutoAck() { + this.autoAckEnabled = false; + } + + @Override + public boolean isAutoAck() { + return this.autoAckEnabled; + } + + } + + /** + * Information for building an KafkaAckCallback. + */ + public class KafkaAckInfoImpl implements KafkaAckInfo { + + private final ConsumerRecord record; + + private final TopicPartition topicPartition; + + private volatile boolean rolledBack; + + private volatile boolean ackDeferred; + + KafkaAckInfoImpl(ConsumerRecord record, TopicPartition topicPartition) { + this.record = record; + this.topicPartition = topicPartition; + } + + @Override + public Object getConsumerMonitor() { + return KafkaMessageSource.this.consumerMonitor; + } + + @Override + public String getGroupId() { + return KafkaMessageSource.this.groupId; + } + + @Override + public Consumer getConsumer() { + return KafkaMessageSource.this.consumer; + } + + @Override + public ConsumerRecord getRecord() { + return this.record; + } + + @Override + public TopicPartition getTopicPartition() { + return this.topicPartition; + } + + @Override + public Map>> getOffsets() { + return KafkaMessageSource.this.inflightRecords; + } + + @Override + public boolean isRolledBack() { + return this.rolledBack; + } + + @Override + public void setRolledBack(boolean rolledBack) { + this.rolledBack = rolledBack; + } + + @Override + public boolean isAckDeferred() { + return this.ackDeferred; + } + + @Override + public void setAckDeferred(boolean ackDeferred) { + this.ackDeferred = ackDeferred; + } + + @Override + public int compareTo(KafkaAckInfo other) { + return Long.compare(this.record.offset(), other.getRecord().offset()); + } + + @Override + public String toString() { + return "KafkaAckInfo [record=" + this.record + ", rolledBack=" + this.rolledBack + ", ackDeferred=" + + this.ackDeferred + "]"; + } + + } + + /** + * Information for building an KafkaAckCallback. + * + * @param the key type. + * @param the value type. + * + */ + public interface KafkaAckInfo extends Comparable> { + + Object getConsumerMonitor(); + + String getGroupId(); + + Consumer getConsumer(); + + ConsumerRecord getRecord(); + + TopicPartition getTopicPartition(); + + Map>> getOffsets(); + + boolean isRolledBack(); + + void setRolledBack(boolean rolledBack); + + boolean isAckDeferred(); + + void setAckDeferred(boolean ackDeferred); + + } + +} 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 3ae4a9cc3f..632e3772ae 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 @@ -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. @@ -21,6 +21,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.util.Collections; import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.stream.Stream; import org.apache.kafka.clients.consumer.ConsumerConfig; @@ -38,6 +40,7 @@ import org.springframework.integration.channel.QueueChannel; 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; @@ -71,6 +74,7 @@ import org.springframework.test.context.junit4.SpringRunner; * @author Artem Bilan * @author Nasko Vasilev * @author Biju Kunjummen + * @author Gary Russell * * @since 3.0 */ @@ -82,8 +86,10 @@ public class KafkaDslTests { private static final String TEST_TOPIC2 = "test-topic2"; + private static final String TEST_TOPIC3 = "test-topic3"; + @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, TEST_TOPIC1, TEST_TOPIC2); + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3); @Autowired @Qualifier("sendToKafkaFlow.input") @@ -112,7 +118,7 @@ public class KafkaDslTests { @Autowired(required = false) @Qualifier("kafkaTemplate:" + TEST_TOPIC1) - private KafkaTemplate kafkaTemplateTopic1; + private KafkaTemplate kafkaTemplateTopic1; @Autowired(required = false) @Qualifier("kafkaTemplate:" + TEST_TOPIC2) @@ -121,8 +127,11 @@ public class KafkaDslTests { @Autowired private DefaultKafkaHeaderMapper mapper; + @Autowired + private ContextConfiguration config; + @Test - public void testKafkaAdapters() { + public void testKafkaAdapters() throws Exception { assertThatThrownBy(() -> this.sendToKafkaFlowInput.send(new GenericMessage<>("foo"))) .hasMessageContaining("10 is not in the range"); @@ -180,12 +189,19 @@ public class KafkaDslTests { assertThat(this.messageListenerContainer).isNotNull(); assertThat(this.kafkaTemplateTopic1).isNotNull(); assertThat(this.kafkaTemplateTopic2).isNotNull(); + + this.kafkaTemplateTopic1.send(TEST_TOPIC3, "foo"); + assertThat(this.config.latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.fromSource).isEqualTo("foo"); } @Configuration @EnableIntegration public static class ContextConfiguration { + private final CountDownLatch latch = new CountDownLatch(1); + + private Object fromSource; @Bean public ConsumerFactory consumerFactory() { @@ -277,6 +293,20 @@ public class KafkaDslTests { .configureKafkaTemplate(t -> t.id("kafkaTemplate:" + topic)); } + + @Bean + public IntegrationFlow sourceFlow() { + return IntegrationFlows + .from(Kafka.inboundChannelAdapter(consumerFactory(), TEST_TOPIC3), + e -> e.poller(Pollers.fixedDelay(100))) + .handle(p -> { + this.fromSource = p.getPayload(); + this.latch.countDown(); + }) + .get(); + } + + } } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java new file mode 100644 index 0000000000..1292783150 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java @@ -0,0 +1,83 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.inbound; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.Map; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.Message; + +/** + * @author Gary Russell + * @since 3.0.1 + * + */ +public class MessageSourceIntegrationTests { + + public static final String TOPIC1 = "MessageSourceIntegrationTests1"; + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 1, TOPIC1); + + @Test + public void testSource() throws Exception { + Map consumerProps = KafkaTestUtils.consumerProps("foo", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps); + KafkaMessageSource source = new KafkaMessageSource<>(consumerFactory, TOPIC1); + + Map producerProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>(producerProps); + KafkaTemplate template = new KafkaTemplate<>(producerFactory); + template.send(TOPIC1, "foo"); + template.send(TOPIC1, "bar"); + template.send(TOPIC1, "baz"); + template.send(TOPIC1, "qux"); + Message received = source.receive(); + int n = 0; + while (n++ < 100 && received == null) { + received = source.receive(); + } + assertThat(received).isNotNull(); + assertThat(received.getPayload()).isEqualTo("foo"); + received = source.receive(); + assertThat(received).isNotNull(); + assertThat(received.getPayload()).isEqualTo("bar"); + received = source.receive(); + assertThat(received).isNotNull(); + assertThat(received.getPayload()).isEqualTo("baz"); + received = source.receive(); + assertThat(received).isNotNull(); + assertThat(received.getPayload()).isEqualTo("qux"); + received = source.receive(); + assertThat(received).isNull(); + source.destroy(); + producerFactory.destroy(); + } + +} 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 new file mode 100644 index 0000000000..d0fb191e40 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java @@ -0,0 +1,433 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +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.anyCollection; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.BDDMockito.willDoNothing; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.commons.logging.Log; +import org.apache.kafka.clients.consumer.Consumer; +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.common.TopicPartition; +import org.apache.kafka.common.record.TimestampType; +import org.junit.Test; +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.kafka.core.ConsumerFactory; +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.Message; + + +/** + * @author Gary Russell + * @since 3.0.1 + * + */ +public class MessageSourceTests { + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + public void testAck() { + Consumer consumer = mock(Consumer.class); + TopicPartition topicPartition = new TopicPartition("foo", 0); + willAnswer(i -> { + ((ConsumerRebalanceListener) i.getArgument(1)) + .onPartitionsAssigned(Collections.singletonList(topicPartition)); + return null; + }).given(consumer).subscribe(anyCollection(), any(ConsumerRebalanceListener.class)); + AtomicReference> paused = new AtomicReference<>(new HashSet<>()); + willAnswer(i -> { + paused.set(new HashSet<>(i.getArgument(0))); + return null; + }).given(consumer).pause(anyCollection()); + willAnswer(i -> paused.get()).given(consumer).paused(); + Map> records1 = new LinkedHashMap<>(); + records1.put(topicPartition, Arrays.asList( + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"), + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + Map> records2 = new LinkedHashMap<>(); + records2.put(topicPartition, Arrays.asList( + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"), + new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz"))); + Map> records3 = new LinkedHashMap<>(); + records3.put(topicPartition, Arrays.asList( + new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz"), + new ConsumerRecord("foo", 0, 3L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux"))); + Map> records4 = new LinkedHashMap<>(); + records4.put(topicPartition, Collections.singletonList( + new ConsumerRecord("foo", 0, 3L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux"))); + ConsumerRecords cr1 = new ConsumerRecords(records1); + ConsumerRecords cr2 = new ConsumerRecords(records2); + ConsumerRecords cr3 = new ConsumerRecords(records3); + ConsumerRecords cr4 = new ConsumerRecords(records4); + ConsumerRecords cr5 = new ConsumerRecords(Collections.emptyMap()); + given(consumer.poll(anyLong())).willReturn(cr1, cr2, cr3, cr4, cr5); + ConsumerFactory consumerFactory = mock(ConsumerFactory.class); + given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer); + KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo"); + + Message received = source.receive(); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received) + .acknowledge(Status.ACCEPT); + received = source.receive(); + assertThat(received).isNotNull(); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received) + .acknowledge(Status.ACCEPT); + received = source.receive(); + assertThat(received).isNotNull(); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received) + .acknowledge(Status.ACCEPT); + received = source.receive(); + assertThat(received).isNotNull(); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received) + .acknowledge(Status.ACCEPT); + received = source.receive(); + assertThat(received).isNull(); + source.destroy(); + InOrder inOrder = inOrder(consumer); + inOrder.verify(consumer).subscribe(anyCollection(), any(ConsumerRebalanceListener.class)); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).seek(any(TopicPartition.class), eq(1L)); + inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(1L))); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).seek(any(TopicPartition.class), eq(2L)); + inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(2L))); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).seek(any(TopicPartition.class), eq(3L)); + inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(3L))); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(4L))); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).close(30, TimeUnit.SECONDS); + inOrder.verifyNoMoreInteractions(); + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + public void testAckOutOfOrder() { + Consumer consumer = mock(Consumer.class); + TopicPartition topicPartition = new TopicPartition("foo", 0); + willAnswer(i -> { + ((ConsumerRebalanceListener) i.getArgument(1)) + .onPartitionsAssigned(Collections.singletonList(topicPartition)); + return null; + }).given(consumer).subscribe(anyCollection(), any(ConsumerRebalanceListener.class)); + AtomicReference> paused = new AtomicReference<>(new HashSet<>()); + willAnswer(i -> { + paused.set(new HashSet<>(i.getArgument(0))); + return null; + }).given(consumer).pause(anyCollection()); + willAnswer(i -> paused.get()).given(consumer).paused(); + Map> records1 = new LinkedHashMap<>(); + records1.put(topicPartition, Collections.singletonList( + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"))); + Map> records2 = new LinkedHashMap<>(); + records2.put(topicPartition, Collections.singletonList( + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + Map> records3 = new LinkedHashMap<>(); + records3.put(topicPartition, Collections.singletonList( + new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz"))); + Map> records4 = new LinkedHashMap<>(); + records4.put(topicPartition, Collections.singletonList( + new ConsumerRecord("foo", 0, 3L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux"))); + Map> records5 = new LinkedHashMap<>(); + records5.put(topicPartition, Collections.singletonList( + new ConsumerRecord("foo", 0, 4L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "fiz"))); + Map> records6 = new LinkedHashMap<>(); + records6.put(topicPartition, Collections.singletonList( + new ConsumerRecord("foo", 0, 5L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "buz"))); + ConsumerRecords cr1 = new ConsumerRecords(records1); + ConsumerRecords cr2 = new ConsumerRecords(records2); + ConsumerRecords cr3 = new ConsumerRecords(records3); + ConsumerRecords cr4 = new ConsumerRecords(records4); + ConsumerRecords cr5 = new ConsumerRecords(records5); + ConsumerRecords cr6 = new ConsumerRecords(records6); + ConsumerRecords cr7 = new ConsumerRecords(Collections.emptyMap()); + given(consumer.poll(anyLong())).willReturn(cr1, cr2, cr3, cr4, cr5, cr6, cr7); + ConsumerFactory consumerFactory = mock(ConsumerFactory.class); + given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer); + KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo"); + + Message received1 = source.receive(); + Message received2 = source.receive(); + Message received3 = source.receive(); + Message received4 = source.receive(); + Message received5 = source.receive(); + Message received6 = source.receive(); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received3) + .acknowledge(Status.ACCEPT); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received2) + .acknowledge(Status.ACCEPT); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received5) + .acknowledge(Status.ACCEPT); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received1) + .acknowledge(Status.ACCEPT); // should commit offset 3 (received 3) + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received6) + .acknowledge(Status.ACCEPT); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received4) + .acknowledge(Status.ACCEPT); // should commit offset 6 (received 6). + assertThat(source.receive()).isNull(); + source.destroy(); + InOrder inOrder = inOrder(consumer); + inOrder.verify(consumer).subscribe(anyCollection(), any(ConsumerRebalanceListener.class)); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(3L))); + inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(6L))); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).close(30, TimeUnit.SECONDS); + inOrder.verifyNoMoreInteractions(); + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + public void testNack() { + Consumer consumer = mock(Consumer.class); + TopicPartition topicPartition = new TopicPartition("foo", 0); + willAnswer(i -> { + ((ConsumerRebalanceListener) i.getArgument(1)) + .onPartitionsAssigned(Collections.singletonList(topicPartition)); + return null; + }).given(consumer).subscribe(anyCollection(), any(ConsumerRebalanceListener.class)); + AtomicReference> paused = new AtomicReference<>(new HashSet<>()); + willAnswer(i -> { + paused.set(new HashSet<>(i.getArgument(0))); + return null; + }).given(consumer).pause(anyCollection()); + willAnswer(i -> paused.get()).given(consumer).paused(); + Map> records1 = new LinkedHashMap<>(); + records1.put(topicPartition, Arrays.asList( + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"), + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + ConsumerRecords cr1 = new ConsumerRecords(records1); + Map> records2 = new LinkedHashMap<>(); + records2.put(topicPartition, Collections.singletonList( + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + ConsumerRecords cr2 = new ConsumerRecords(records2); + ConsumerRecords cr3 = new ConsumerRecords(Collections.emptyMap()); + given(consumer.poll(anyLong())).willReturn(cr1, cr1, cr2, cr2, cr3); + ConsumerFactory consumerFactory = mock(ConsumerFactory.class); + given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer); + KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo"); + + Message received = source.receive(); + assertThat(received.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(0L); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received) + .acknowledge(Status.REQUEUE); + received = source.receive(); + assertThat(received.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(0L); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received) + .acknowledge(Status.ACCEPT); + received = source.receive(); + assertThat(received.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(1L); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received) + .acknowledge(Status.REQUEUE); + received = source.receive(); + assertThat(received.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(1L); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received) + .acknowledge(Status.ACCEPT); + received = source.receive(); + source.destroy(); + assertThat(received).isNull(); + InOrder inOrder = inOrder(consumer); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).seek(topicPartition, 1L); // returned 2 - seek after poll + inOrder.verify(consumer).seek(topicPartition, 0L); // rollback + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).seek(topicPartition, 1L); // seek after poll + inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(1L))); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).seek(topicPartition, 1L); // rollback + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(2L))); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).close(30, TimeUnit.SECONDS); + inOrder.verifyNoMoreInteractions(); + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + public void testNackWithLaterInflight() { + Consumer consumer = mock(Consumer.class); + TopicPartition topicPartition = new TopicPartition("foo", 0); + willAnswer(i -> { + ((ConsumerRebalanceListener) i.getArgument(1)) + .onPartitionsAssigned(Collections.singletonList(topicPartition)); + return null; + }).given(consumer).subscribe(anyCollection(), any(ConsumerRebalanceListener.class)); + AtomicReference> paused = new AtomicReference<>(new HashSet<>()); + willAnswer(i -> { + paused.set(new HashSet<>(i.getArgument(0))); + return null; + }).given(consumer).pause(anyCollection()); + willAnswer(i -> paused.get()).given(consumer).paused(); + Map> records1 = new LinkedHashMap<>(); + records1.put(topicPartition, Arrays.asList( + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"), + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + ConsumerRecords cr1 = new ConsumerRecords(records1); + Map> records2 = new LinkedHashMap<>(); + records2.put(topicPartition, Collections.singletonList( + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + ConsumerRecords cr2 = new ConsumerRecords(records2); + ConsumerRecords cr3 = new ConsumerRecords(Collections.emptyMap()); + given(consumer.poll(anyLong())).willReturn(cr1, cr2, cr1, cr2, cr3); + ConsumerFactory consumerFactory = mock(ConsumerFactory.class); + given(consumerFactory.createConsumer(isNull(), anyString(), isNull())).willReturn(consumer); + KafkaMessageSource source = new KafkaMessageSource(consumerFactory, "foo"); + + Message received1 = source.receive(); + Message received2 = source.receive(); // inflight + assertThat(received1.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(0L); + AcknowledgmentCallback ack1 = StaticMessageHeaderAccessor.getAcknowledgmentCallback(received1); + Log log1 = spy(KafkaTestUtils.getPropertyValue(ack1, "logger", Log.class)); + new DirectFieldAccessor(ack1).setPropertyValue("logger", log1); + given(log1.isWarnEnabled()).willReturn(true); + willDoNothing().given(log1).warn(any()); + ack1.acknowledge(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); + received1 = source.receive(); + assertThat(received1.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(0L); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received1) + .acknowledge(Status.ACCEPT); + received2 = source.receive(); + assertThat(received2.getHeaders().get(KafkaHeaders.OFFSET)).isEqualTo(1L); + StaticMessageHeaderAccessor.getAcknowledgmentCallback(received2) + .acknowledge(Status.ACCEPT); + received1 = source.receive(); + source.destroy(); + assertThat(received1).isNull(); + InOrder inOrder = inOrder(consumer, log1, log2); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).seek(topicPartition, 1L); // returned 2 - seek after poll + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); // in flight + inOrder.verify(consumer).seek(topicPartition, 0L); // rollback + inOrder.verify(log1).isWarnEnabled(); + ArgumentCaptor captor = ArgumentCaptor.forClass(String.class); + inOrder.verify(log1).warn(captor.capture()); + assertThat(captor.getValue()) + .contains("Rolled back") + .contains("later in-flight offsets [1] will also be re-fetched"); + inOrder.verify(log2).isWarnEnabled(); + captor = ArgumentCaptor.forClass(String.class); + inOrder.verify(log2).warn(captor.capture()); + assertThat(captor.getValue()) + .contains("Cannot commit offset for ConsumerRecord") + .contains("; an earlier offset was rolled back"); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).seek(topicPartition, 1L); // seek after poll + inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(1L))); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(2L))); + inOrder.verify(consumer).paused(); + inOrder.verify(consumer).resume(anyCollection()); + inOrder.verify(consumer).poll(anyLong()); + inOrder.verify(consumer).pause(anyCollection()); + inOrder.verify(consumer).close(30, TimeUnit.SECONDS); + inOrder.verifyNoMoreInteractions(); + } + +}