GH-184: Add Polled MessageSource
Resolves https://github.com/spring-projects/spring-integration-kafka/issues/184 Polishing - changes in core Pause/Resume; Add integration tests Synchronize all consumer operations on the consumer, to support noAutoAck() Polishing - PR Comments More polishing Major rework - see commit comment - only fetch one record at a time (warn log and seek if max.poll.records incorrect) - when an application has multiple outstanding records, - track records; apply commits in the right order, deferring if necessary - when requeued, mark later offsets as rolled back to prevent commits - Support transactions - start/end transactions - provide access to the producer Fix transaction synchronization - if there is an existing Kafka transaction, participate in it - if not, bind a new transactional resource to the thread - allows a KafkaTemplate to participate by utilizing the producer factory message header - if there is an existing non-kafka transaction, sync the kafka transaction with it Check for KafkaMessageHeaders when building the message. Polishing - PR Comments Simplify AckInfo Remove internal transaction support - the user can start a transacion before calling receive and send offsets to the transaction using a KafkaTemplate. Polishing * Simple polishing according IDEA warnings * Upgrade to Gradle 4.4.1 * Some upgrades and dependencies polishing
This commit is contained in:
Gary Russell
committed by
Artem Bilan
Artem Bilan
parent
29973e525d
commit
b44dab40fa
@@ -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.*" />
|
||||
</module>
|
||||
|
||||
@@ -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 <K, V, S extends KafkaProducerMessageHandlerSpec<K, V, S>> KafkaProducerMessageHandlerSpec<K, V, S>
|
||||
outboundChannelAdapter(KafkaTemplate<K, V> kafkaTemplate) {
|
||||
outboundChannelAdapter(KafkaTemplate<K, V> kafkaTemplate) {
|
||||
|
||||
return new KafkaProducerMessageHandlerSpec<>(kafkaTemplate);
|
||||
}
|
||||
|
||||
@@ -60,10 +62,45 @@ public final class Kafka {
|
||||
* @see <a href="https://kafka.apache.org/documentation.html#producerconfigs">Kafka Producer Configs</a>
|
||||
*/
|
||||
public static <K, V> KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec<K, V>
|
||||
outboundChannelAdapter(ProducerFactory<K, V> producerFactory) {
|
||||
outboundChannelAdapter(ProducerFactory<K, V> 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 <K> the Kafka message key type.
|
||||
* @param <V> the Kafka message value type.
|
||||
* @return the spec.
|
||||
* @since 3.0.1
|
||||
*/
|
||||
public static <K, V> KafkaInboundChannelAdapterSpec<K, V>
|
||||
inboundChannelAdapter(ConsumerFactory<K, V> 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 <K> the Kafka message key type.
|
||||
* @param <V> the Kafka message value type.
|
||||
* @return the spec.
|
||||
* @since 3.0.1
|
||||
*/
|
||||
public static <K, V> KafkaInboundChannelAdapterSpec<K, V>
|
||||
inboundChannelAdapter(ConsumerFactory<K, V> consumerFactory,
|
||||
KafkaAckCallbackFactory<K, V> 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 <K, V, S extends KafkaMessageDrivenChannelAdapterSpec<K, V, S>>
|
||||
KafkaMessageDrivenChannelAdapterSpec<K, V, S> messageDrivenChannelAdapter(
|
||||
KafkaMessageDrivenChannelAdapterSpec<K, V, S> messageDrivenChannelAdapter(
|
||||
AbstractMessageListenerContainer<K, V> listenerContainer) {
|
||||
|
||||
return messageDrivenChannelAdapter(listenerContainer, KafkaMessageDrivenChannelAdapter.ListenerMode.record);
|
||||
}
|
||||
|
||||
@@ -88,9 +126,10 @@ public final class Kafka {
|
||||
* @return the KafkaMessageDrivenChannelAdapterSpec.
|
||||
*/
|
||||
public static <K, V, A extends KafkaMessageDrivenChannelAdapterSpec<K, V, A>>
|
||||
KafkaMessageDrivenChannelAdapterSpec<K, V, A> messageDrivenChannelAdapter(
|
||||
KafkaMessageDrivenChannelAdapterSpec<K, V, A> messageDrivenChannelAdapter(
|
||||
AbstractMessageListenerContainer<K, V> listenerContainer,
|
||||
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
|
||||
|
||||
return new KafkaMessageDrivenChannelAdapterSpec<>(listenerContainer, listenerMode);
|
||||
}
|
||||
|
||||
@@ -105,7 +144,8 @@ public final class Kafka {
|
||||
*/
|
||||
public static <K, V>
|
||||
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, ContainerProperties containerProperties) {
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, ContainerProperties containerProperties) {
|
||||
|
||||
return messageDrivenChannelAdapter(consumerFactory, containerProperties,
|
||||
KafkaMessageDrivenChannelAdapter.ListenerMode.record);
|
||||
}
|
||||
@@ -122,8 +162,9 @@ public final class Kafka {
|
||||
*/
|
||||
public static <K, V>
|
||||
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, ContainerProperties containerProperties,
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, ContainerProperties containerProperties,
|
||||
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
|
||||
|
||||
return messageDrivenChannelAdapter(
|
||||
new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory,
|
||||
containerProperties), listenerMode);
|
||||
@@ -140,8 +181,9 @@ public final class Kafka {
|
||||
*/
|
||||
public static <K, V>
|
||||
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
|
||||
TopicPartitionInitialOffset... topicPartitions) {
|
||||
|
||||
return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record,
|
||||
topicPartitions);
|
||||
}
|
||||
@@ -158,9 +200,10 @@ public final class Kafka {
|
||||
*/
|
||||
public static <K, V>
|
||||
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
|
||||
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode,
|
||||
TopicPartitionInitialOffset... topicPartitions) {
|
||||
|
||||
return messageDrivenChannelAdapter(
|
||||
new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory,
|
||||
topicPartitions), listenerMode);
|
||||
@@ -177,7 +220,8 @@ public final class Kafka {
|
||||
*/
|
||||
public static <K, V>
|
||||
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, String... topics) {
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, String... topics) {
|
||||
|
||||
return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record,
|
||||
topics);
|
||||
}
|
||||
@@ -194,8 +238,9 @@ public final class Kafka {
|
||||
*/
|
||||
public static <K, V>
|
||||
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
|
||||
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, String... topics) {
|
||||
|
||||
return messageDrivenChannelAdapter(
|
||||
new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory,
|
||||
topics), listenerMode);
|
||||
@@ -212,7 +257,8 @@ public final class Kafka {
|
||||
*/
|
||||
public static <K, V>
|
||||
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, Pattern topicPattern) {
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, Pattern topicPattern) {
|
||||
|
||||
return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record,
|
||||
topicPattern);
|
||||
}
|
||||
@@ -229,8 +275,9 @@ public final class Kafka {
|
||||
*/
|
||||
public static <K, V>
|
||||
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
|
||||
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
|
||||
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, Pattern topicPattern) {
|
||||
|
||||
return messageDrivenChannelAdapter(
|
||||
new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory,
|
||||
topicPattern),
|
||||
@@ -239,8 +286,9 @@ public final class Kafka {
|
||||
|
||||
private static <K, V>
|
||||
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
|
||||
messageDrivenChannelAdapter(KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K, V> spec,
|
||||
messageDrivenChannelAdapter(KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K, V> spec,
|
||||
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
|
||||
|
||||
return new KafkaMessageDrivenChannelAdapterSpec
|
||||
.KafkaMessageDrivenChannelAdapterListenerContainerSpec<>(spec, listenerMode);
|
||||
}
|
||||
|
||||
@@ -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 <K> the key type.
|
||||
* @param <V> the value type.
|
||||
*
|
||||
* @author Gary Russell
|
||||
*
|
||||
* @since 3.0.1
|
||||
*
|
||||
*/
|
||||
public class KafkaInboundChannelAdapterSpec<K, V>
|
||||
extends MessageSourceSpec<KafkaInboundChannelAdapterSpec<K, V>, KafkaMessageSource<K, V>> {
|
||||
|
||||
KafkaInboundChannelAdapterSpec(ConsumerFactory<K, V> consumerFactory, String... topics) {
|
||||
this.target = new KafkaMessageSource<>(consumerFactory, topics);
|
||||
}
|
||||
|
||||
KafkaInboundChannelAdapterSpec(ConsumerFactory<K, V> consumerFactory,
|
||||
KafkaAckCallbackFactory<K, V> ackCallbackFactory, String... topics) {
|
||||
|
||||
this.target = new KafkaMessageSource<>(consumerFactory, ackCallbackFactory, topics);
|
||||
}
|
||||
|
||||
public KafkaInboundChannelAdapterSpec<K, V> groupId(String groupId) {
|
||||
this.target.setGroupId(groupId);
|
||||
return this;
|
||||
}
|
||||
|
||||
public KafkaInboundChannelAdapterSpec<K, V> clientId(String clientId) {
|
||||
this.target.setClientId(clientId);
|
||||
return this;
|
||||
}
|
||||
|
||||
public KafkaInboundChannelAdapterSpec<K, V> pollTimeout(long pollTimeout) {
|
||||
this.target.setPollTimeout(pollTimeout);
|
||||
return this;
|
||||
}
|
||||
|
||||
public KafkaInboundChannelAdapterSpec<K, V> messageConverter(RecordMessageConverter messageConverter) {
|
||||
this.target.setMessageConverter(messageConverter);
|
||||
return this;
|
||||
}
|
||||
|
||||
public KafkaInboundChannelAdapterSpec<K, V> payloadType(Type type) {
|
||||
this.target.setPayloadType(type);
|
||||
return this;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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.
|
||||
* <p>
|
||||
* 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 <K> the key type.
|
||||
* @param <V> the value type.
|
||||
*
|
||||
* @author Gary Russell
|
||||
* @since 3.0.1
|
||||
*
|
||||
*/
|
||||
public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
|
||||
implements DisposableBean {
|
||||
|
||||
private static final long DEFAULT_POLL_TIMEOUT = 50L;
|
||||
|
||||
private final ConsumerFactory<K, V> consumerFactory;
|
||||
|
||||
private final KafkaAckCallbackFactory<K, V> ackCallbackFactory;
|
||||
|
||||
private final String[] topics;
|
||||
|
||||
private final Object consumerMonitor = new Object();
|
||||
|
||||
private final Map<TopicPartition, Set<KafkaAckInfo<K, V>>> 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<K, V> consumer;
|
||||
|
||||
private volatile Collection<TopicPartition> partitions;
|
||||
|
||||
public KafkaMessageSource(ConsumerFactory<K, V> consumerFactory, String... topics) {
|
||||
this(consumerFactory, new KafkaAckCallbackFactory<>(), topics);
|
||||
}
|
||||
|
||||
public KafkaMessageSource(ConsumerFactory<K, V> consumerFactory,
|
||||
KafkaAckCallbackFactory<K, V> 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<K, V> record;
|
||||
TopicPartition topicPartition;
|
||||
synchronized (this.consumerMonitor) {
|
||||
Set<TopicPartition> paused = this.consumer.paused();
|
||||
if (paused.size() > 0) {
|
||||
this.consumer.resume(paused);
|
||||
}
|
||||
ConsumerRecords<K, V> 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<K, V> 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<String, Object> 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<TopicPartition> partitions) {
|
||||
KafkaMessageSource.this.partitions = Collections.emptyList();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
|
||||
KafkaMessageSource.this.partitions = new ArrayList<>(partitions);
|
||||
}
|
||||
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void destroy() {
|
||||
if (this.consumer != null) {
|
||||
Consumer<K, V> consumer2 = this.consumer;
|
||||
this.consumer = null;
|
||||
synchronized (this.consumerMonitor) {
|
||||
consumer2.close(30, TimeUnit.SECONDS);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* AcknowledgmentCallbackFactory for KafkaAckInfo.
|
||||
*
|
||||
* @param <K> the key type.
|
||||
* @param <V> the value type.
|
||||
*
|
||||
*/
|
||||
public static class KafkaAckCallbackFactory<K, V> implements AcknowledgmentCallbackFactory<KafkaAckInfo<K, V>> {
|
||||
|
||||
@Override
|
||||
public AcknowledgmentCallback createCallback(KafkaAckInfo<K, V> info) {
|
||||
return new KafkaAckCallback<>(info);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* AcknowledgmentCallback for Kafka.
|
||||
*
|
||||
* @param <K> the key type.
|
||||
* @param <V> the value type.
|
||||
*
|
||||
*/
|
||||
public static class KafkaAckCallback<K, V> implements AcknowledgmentCallback, Acknowledgment {
|
||||
|
||||
private final Log logger = LogFactory.getLog(getClass());
|
||||
|
||||
private final KafkaAckInfo<K, V> ackInfo;
|
||||
|
||||
private volatile boolean acknowledged;
|
||||
|
||||
private boolean autoAckEnabled = true;
|
||||
|
||||
public KafkaAckCallback(KafkaAckInfo<K, V> 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<K, V> 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<K, V> record) {
|
||||
this.ackInfo.getConsumer().seek(this.ackInfo.getTopicPartition(), record.offset());
|
||||
Set<KafkaAckInfo<K, V>> inflight = this.ackInfo.getOffsets().get(this.ackInfo.getTopicPartition());
|
||||
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());
|
||||
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<K, V> 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<KafkaAckInfo<K, V>> candidates = this.ackInfo.getOffsets().get(this.ackInfo.getTopicPartition());
|
||||
KafkaAckInfo<K, V> ackInfo = null;
|
||||
if (candidates.iterator().next().equals(this.ackInfo)) {
|
||||
// see if there are any pending acks for higher offsets
|
||||
List<KafkaAckInfo<K, V>> toCommit = new ArrayList<>();
|
||||
for (KafkaAckInfo<K, V> 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<K, V> {
|
||||
|
||||
private final ConsumerRecord<K, V> record;
|
||||
|
||||
private final TopicPartition topicPartition;
|
||||
|
||||
private volatile boolean rolledBack;
|
||||
|
||||
private volatile boolean ackDeferred;
|
||||
|
||||
KafkaAckInfoImpl(ConsumerRecord<K, V> 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<K, V> getConsumer() {
|
||||
return KafkaMessageSource.this.consumer;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ConsumerRecord<K, V> getRecord() {
|
||||
return this.record;
|
||||
}
|
||||
|
||||
@Override
|
||||
public TopicPartition getTopicPartition() {
|
||||
return this.topicPartition;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<TopicPartition, Set<KafkaAckInfo<K, V>>> 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<K, V> 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 <K> the key type.
|
||||
* @param <V> the value type.
|
||||
*
|
||||
*/
|
||||
public interface KafkaAckInfo<K, V> extends Comparable<KafkaAckInfo<K, V>> {
|
||||
|
||||
Object getConsumerMonitor();
|
||||
|
||||
String getGroupId();
|
||||
|
||||
Consumer<K, V> getConsumer();
|
||||
|
||||
ConsumerRecord<K, V> getRecord();
|
||||
|
||||
TopicPartition getTopicPartition();
|
||||
|
||||
Map<TopicPartition, Set<KafkaAckInfo<K, V>>> getOffsets();
|
||||
|
||||
boolean isRolledBack();
|
||||
|
||||
void setRolledBack(boolean rolledBack);
|
||||
|
||||
boolean isAckDeferred();
|
||||
|
||||
void setAckDeferred(boolean ackDeferred);
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<Object, Object> 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<Integer, String> 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();
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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<String, Object> consumerProps = KafkaTestUtils.consumerProps("foo", "false", embeddedKafka);
|
||||
consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2);
|
||||
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
|
||||
DefaultKafkaConsumerFactory<Integer, String> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps);
|
||||
KafkaMessageSource<Integer, String> source = new KafkaMessageSource<>(consumerFactory, TOPIC1);
|
||||
|
||||
Map<String, Object> producerProps = KafkaTestUtils.producerProps(embeddedKafka);
|
||||
DefaultKafkaProducerFactory<Object, Object> producerFactory = new DefaultKafkaProducerFactory<>(producerProps);
|
||||
KafkaTemplate<Object, Object> template = new KafkaTemplate<>(producerFactory);
|
||||
template.send(TOPIC1, "foo");
|
||||
template.send(TOPIC1, "bar");
|
||||
template.send(TOPIC1, "baz");
|
||||
template.send(TOPIC1, "qux");
|
||||
Message<Object> 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();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<Set<TopicPartition>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<Set<TopicPartition>> 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<TopicPartition, List<ConsumerRecord>> records1 = new LinkedHashMap<>();
|
||||
records1.put(topicPartition, Collections.singletonList(
|
||||
new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo")));
|
||||
Map<TopicPartition, List<ConsumerRecord>> records2 = new LinkedHashMap<>();
|
||||
records2.put(topicPartition, Collections.singletonList(
|
||||
new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar")));
|
||||
Map<TopicPartition, List<ConsumerRecord>> records3 = new LinkedHashMap<>();
|
||||
records3.put(topicPartition, Collections.singletonList(
|
||||
new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz")));
|
||||
Map<TopicPartition, List<ConsumerRecord>> records4 = new LinkedHashMap<>();
|
||||
records4.put(topicPartition, Collections.singletonList(
|
||||
new ConsumerRecord("foo", 0, 3L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux")));
|
||||
Map<TopicPartition, List<ConsumerRecord>> records5 = new LinkedHashMap<>();
|
||||
records5.put(topicPartition, Collections.singletonList(
|
||||
new ConsumerRecord("foo", 0, 4L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "fiz")));
|
||||
Map<TopicPartition, List<ConsumerRecord>> 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<Set<TopicPartition>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<Set<TopicPartition>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<String> 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();
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user