From aef99fcd9d1e4a6dd5df83c60747f09516a667e5 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Mon, 2 Mar 2015 17:42:21 -0500 Subject: [PATCH] INTEXT-143 Add Support for Manual Acknowledgment - Introduce an Acknowledgment object that message processors can invoke; - Introduce an AcknowedgingMessageListener variation of the MessageListener that receives an Acknowledgment reference for the processed message; - Add 'autoCommitOffset' settings to the KafkaMessageListenerContainer and KafkaMessageDrivenChannelAdapter, and the ability to inject a MessageListener or an AcknowledgingMessageListener in either (the allowed type depending on the offset management strategy) - Prepopulate a message header for SI messages created by the KafkaMessageDrivenChannelAdapter if autoCommit is enabled or disabled; - OffsetManager only sets values that are higher than the ones already set in a session (barring reset) - this is to prevent asynchrous acks to mistakenly revert checkpoints Addressed PR comments - removed autoOffsetCommit flag for the KafkaMessageListenerContainer, relying only on the messageListener type to detect one versus the other; - added since tags - added copyright @since tags Polishing --- .../KafkaMessageDrivenChannelAdapter.java | 65 ++++++-- ...tDecodingAcknowledgingMessageListener.java | 62 ++++++++ .../kafka/listener/AbstractOffsetManager.java | 15 +- .../AcknowledgingMessageListener.java | 38 +++++ .../kafka/listener/Acknowledgment.java | 37 +++++ .../ConcurrentMessageListenerDispatcher.java | 9 +- .../kafka/listener/DefaultAcknowledgment.java | 52 ++++++ .../KafkaMessageListenerContainer.java | 47 +++--- .../QueueingMessageListenerInvoker.java | 34 +++- .../kafka/support/KafkaHeaders.java | 7 +- ...KafkaMessageDrivenChannelAdapterTests.java | 106 ++++++++++++- .../SingleBrokerWithManualAckTests.java | 149 ++++++++++++++++++ 12 files changed, 566 insertions(+), 55 deletions(-) create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AbstractDecodingAcknowledgingMessageListener.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AcknowledgingMessageListener.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/Acknowledgment.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/DefaultAcknowledgment.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/SingleBrokerWithManualAckTests.java diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java index 6115ae25d0..f2a75ff563 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java @@ -16,30 +16,34 @@ package org.springframework.integration.kafka.inbound; -import kafka.serializer.Decoder; -import kafka.serializer.DefaultDecoder; - import org.springframework.integration.context.OrderlyShutdownCapable; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.kafka.core.KafkaMessageMetadata; +import org.springframework.integration.kafka.listener.AbstractDecodingAcknowledgingMessageListener; import org.springframework.integration.kafka.listener.AbstractDecodingMessageListener; +import org.springframework.integration.kafka.listener.Acknowledgment; import org.springframework.integration.kafka.listener.KafkaMessageListenerContainer; import org.springframework.integration.kafka.support.KafkaHeaders; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.messaging.Message; -import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; +import kafka.serializer.Decoder; +import kafka.serializer.DefaultDecoder; + /** * @author Marius Bogoevici */ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport implements OrderlyShutdownCapable { - private KafkaMessageListenerContainer messageListenerContainer; + private final KafkaMessageListenerContainer messageListenerContainer; private Decoder keyDecoder = new DefaultDecoder(null); private Decoder payloadDecoder = new DefaultDecoder(null); + private boolean autoCommitOffset = true; + public KafkaMessageDrivenChannelAdapter(KafkaMessageListenerContainer messageListenerContainer) { Assert.notNull(messageListenerContainer); Assert.isNull(messageListenerContainer.getMessageListener()); @@ -55,9 +59,15 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport imp this.payloadDecoder = payloadDecoder; } + public void setAutoCommitOffset(boolean autoCommitOffset) { + this.autoCommitOffset = autoCommitOffset; + } + @Override protected void onInit() { - this.messageListenerContainer.setMessageListener(new ChannelForwardingMessageListener()); + this.messageListenerContainer.setMessageListener(autoCommitOffset ? + new AutoAcknowledgingChannelForwardingMessageListener() + : new AcknowledgingChannelForwardingMessageListener()); super.onInit(); } @@ -88,25 +98,48 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport imp } @SuppressWarnings("rawtypes") - private class ChannelForwardingMessageListener extends AbstractDecodingMessageListener { + private class AutoAcknowledgingChannelForwardingMessageListener extends AbstractDecodingMessageListener { @SuppressWarnings("unchecked") - public ChannelForwardingMessageListener() { + public AutoAcknowledgingChannelForwardingMessageListener() { super(keyDecoder, payloadDecoder); } @Override public void doOnMessage(Object key, Object payload, KafkaMessageMetadata metadata) { - Message message = getMessageBuilderFactory() - .withPayload(payload) - .setHeader(KafkaHeaders.MESSAGE_KEY, key) - .setHeader(KafkaHeaders.TOPIC, metadata.getPartition().getTopic()) - .setHeader(KafkaHeaders.PARTITION_ID, metadata.getPartition().getId()) - .setHeader(KafkaHeaders.OFFSET, metadata.getOffset()) - .build(); - KafkaMessageDrivenChannelAdapter.this.sendMessage(message); + KafkaMessageDrivenChannelAdapter.this.sendMessage(toMessage(key, payload, metadata, null)); } } + @SuppressWarnings("rawtypes") + private class AcknowledgingChannelForwardingMessageListener extends AbstractDecodingAcknowledgingMessageListener { + + @SuppressWarnings("unchecked") + public AcknowledgingChannelForwardingMessageListener() { + super(keyDecoder, payloadDecoder); + } + + @Override + public void doOnMessage(Object key, Object payload, KafkaMessageMetadata metadata, + Acknowledgment acknowledgment) { + KafkaMessageDrivenChannelAdapter.this.sendMessage(toMessage(key, payload, metadata, acknowledgment)); + } + + } + + private Message toMessage(Object key, Object payload, KafkaMessageMetadata metadata, + Acknowledgment acknowledgment) { + AbstractIntegrationMessageBuilder messageBuilder = getMessageBuilderFactory().withPayload(payload) + .setHeader(KafkaHeaders.MESSAGE_KEY, key) + .setHeader(KafkaHeaders.TOPIC, metadata.getPartition().getTopic()) + .setHeader(KafkaHeaders.PARTITION_ID, metadata.getPartition().getId()) + .setHeader(KafkaHeaders.OFFSET, metadata.getOffset()) + .setHeader(KafkaHeaders.NEXT_OFFSET, metadata.getNextOffset()); + if (acknowledgment != null) { + messageBuilder.setHeader(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment); + } + return messageBuilder.build(); + } + } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AbstractDecodingAcknowledgingMessageListener.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AbstractDecodingAcknowledgingMessageListener.java new file mode 100644 index 0000000000..8355f0fecd --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AbstractDecodingAcknowledgingMessageListener.java @@ -0,0 +1,62 @@ +/* + * Copyright 2015 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.listener; + + +import org.springframework.integration.kafka.core.KafkaMessage; +import org.springframework.integration.kafka.core.KafkaMessageMetadata; +import org.springframework.integration.kafka.util.MessageUtils; + +import kafka.serializer.Decoder; + +/** + * Base {@link AcknowledgingMessageListener} implementation that decodes the key and the + * payload using the supplied {@link Decoder}s. + * + * Users of this class must extend it and implement {@code doOnMessage} and must supply + * {@link Decoder} implementations for both the key and the payload. + * + * @author Marius Bogoevici + * @since 1.0.1 + */ +public abstract class AbstractDecodingAcknowledgingMessageListener implements AcknowledgingMessageListener { + + private final Decoder keyDecoder; + + private final Decoder

payloadDecoder; + + public AbstractDecodingAcknowledgingMessageListener(Decoder keyDecoder, Decoder

payloadDecoder) { + this.keyDecoder = keyDecoder; + this.payloadDecoder = payloadDecoder; + } + + @Override + public final void onMessage(KafkaMessage message, Acknowledgment acknowledgment) { + this.doOnMessage(MessageUtils.decodeKey(message, keyDecoder), + MessageUtils.decodePayload(message, payloadDecoder), message.getMetadata(), acknowledgment); + } + + /** + * Process the decoded message + * @param key the message key + * @param payload the message body + * @param metadata the KafkaMessageMetadata + * @param acknowledgment the acknowledgment handle + */ + public abstract void doOnMessage(K key, P payload, KafkaMessageMetadata metadata, Acknowledgment acknowledgment); + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AbstractOffsetManager.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AbstractOffsetManager.java index b27d501d71..9bd5db7e09 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AbstractOffsetManager.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AbstractOffsetManager.java @@ -20,7 +20,9 @@ import java.io.IOException; import java.util.Collection; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import kafka.common.ErrorMapping; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -35,8 +37,6 @@ import org.springframework.integration.kafka.core.PartitionNotFoundException; import org.springframework.integration.kafka.core.Result; import org.springframework.util.Assert; -import kafka.common.ErrorMapping; - /** * Base implementation for {@link OffsetManager}. Subclasses may customize functionality as necessary. * @@ -54,6 +54,8 @@ public abstract class AbstractOffsetManager implements OffsetManager, Disposable protected Map initialOffsets; + protected Map highestUpdatedOffsets = new ConcurrentHashMap(); + public AbstractOffsetManager(ConnectionFactory connectionFactory) { this(connectionFactory, new HashMap()); } @@ -62,7 +64,7 @@ public abstract class AbstractOffsetManager implements OffsetManager, Disposable Assert.notNull(connectionFactory, "A 'connectionFactory' can't be null"); Assert.notNull(initialOffsets, "An initialOffsets can't be null"); this.connectionFactory = connectionFactory; - this.initialOffsets = initialOffsets; + this.initialOffsets = new HashMap(initialOffsets); } public String getConsumerId() { @@ -108,7 +110,11 @@ public abstract class AbstractOffsetManager implements OffsetManager, Disposable */ @Override public synchronized final void updateOffset(Partition partition, long offset) { - doUpdateOffset(partition, offset); + Long highestUpdatedOffset = this.highestUpdatedOffsets.get(partition); + if (highestUpdatedOffset == null || highestUpdatedOffset < offset) { + highestUpdatedOffsets.put(partition, offset); + doUpdateOffset(partition, offset); + } } /** @@ -148,6 +154,7 @@ public abstract class AbstractOffsetManager implements OffsetManager, Disposable for (Partition partition : partitionsToReset) { doRemoveOffset(partition); this.initialOffsets.remove(partition); + this.highestUpdatedOffsets.remove(partition); } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AcknowledgingMessageListener.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AcknowledgingMessageListener.java new file mode 100644 index 0000000000..0ec05e0f15 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/AcknowledgingMessageListener.java @@ -0,0 +1,38 @@ +/* + * Copyright 2015 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.listener; + +import org.springframework.integration.kafka.core.KafkaMessage; + +/** + * Listener for handling incoming Kafka messages, propagating an acknowledgment handle that recipients + * can invoke when the message has been processed. + * + * @author Marius Bogoevici + * @since 1.0.1 + */ +public interface AcknowledgingMessageListener { + + /** + * Executes when a Kafka message is received + * + * @param message the Kafka message to be processed + * @param acknowledgment a handle for acknowledging the message processing + */ + void onMessage(KafkaMessage message, Acknowledgment acknowledgment); + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/Acknowledgment.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/Acknowledgment.java new file mode 100644 index 0000000000..1dda71edc6 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/Acknowledgment.java @@ -0,0 +1,37 @@ +/* + * Copyright 2015 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.listener; + +import org.springframework.integration.kafka.core.KafkaMessage; + +/** + * Handle for acknowledging the processing of a {@link KafkaMessage}. Recipients can store the reference in + * asynchronous scenarios, but the internal state should be assumed transient (i.e. it cannot be serialized + * and deserialized later) + * + * @author Marius Bogoevici + * @since 1.0.1 + */ +public interface Acknowledgment { + + /** + * Invoked when the message for which the acknowledgment has been created has been processed. + * Calling this method implies that all the previous messages in the partition have been processed already. + */ + void acknowledge(); + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java index 62bbffe4a1..c6b494db22 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java @@ -56,7 +56,7 @@ class ConcurrentMessageListenerDispatcher implements Lifecycle { private volatile boolean running; - private final MessageListener delegateListener; + private final Object delegateListener; private final ErrorHandler errorHandler; @@ -68,8 +68,13 @@ class ConcurrentMessageListenerDispatcher implements Lifecycle { private Executor taskExecutor; - public ConcurrentMessageListenerDispatcher(MessageListener delegateListener, ErrorHandler errorHandler, + public ConcurrentMessageListenerDispatcher(Object delegateListener, ErrorHandler errorHandler, Collection partitions, OffsetManager offsetManager, int consumers, int queueSize) { + Assert.isTrue + (delegateListener instanceof MessageListener + || delegateListener instanceof AcknowledgingMessageListener, + "Either a " + MessageListener.class.getName() + " or a " + + AcknowledgingMessageListener.class.getName() + " must be provided"); Assert.notEmpty(partitions, "A set of partitions must be provided"); Assert.isTrue(consumers <= partitions.size(), "The number of consumers must be smaller or equal to the number of partitions"); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/DefaultAcknowledgment.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/DefaultAcknowledgment.java new file mode 100644 index 0000000000..a4c16f930e --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/DefaultAcknowledgment.java @@ -0,0 +1,52 @@ +/* + * Copyright 2015 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.listener; + +import org.springframework.integration.kafka.core.KafkaMessage; +import org.springframework.integration.kafka.core.Partition; + +/** + * Default implementation for an {@link Acknowledgment} that defers to an underlying + * {@link OffsetManager}. + * + * @author Marius Bogoevici + * @since 1.0.1 + */ +public class DefaultAcknowledgment implements Acknowledgment { + + private final OffsetManager offsetManager; + + private final Partition partition; + + private final Long offset; + + public DefaultAcknowledgment(OffsetManager offsetManager, Partition partition, Long offset) { + this.offsetManager = offsetManager; + this.partition = partition; + this.offset = offset; + } + + public DefaultAcknowledgment(OffsetManager offsetManager, KafkaMessage message) { + this(offsetManager, message.getMetadata().getPartition(), message.getMetadata().getNextOffset()); + } + + @Override + public void acknowledge() { + offsetManager.updateOffset(partition, offset); + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java index 0cac19bafd..10be3063ee 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java @@ -31,6 +31,23 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executor; import java.util.concurrent.Executors; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.context.SmartLifecycle; +import org.springframework.integration.kafka.core.BrokerAddress; +import org.springframework.integration.kafka.core.ConnectionFactory; +import org.springframework.integration.kafka.core.ConsumerException; +import org.springframework.integration.kafka.core.FetchRequest; +import org.springframework.integration.kafka.core.KafkaConsumerDefaults; +import org.springframework.integration.kafka.core.KafkaMessage; +import org.springframework.integration.kafka.core.KafkaMessageBatch; +import org.springframework.integration.kafka.core.KafkaTemplate; +import org.springframework.integration.kafka.core.Partition; +import org.springframework.integration.kafka.core.Result; +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; + import com.gs.collections.api.RichIterable; import com.gs.collections.api.block.function.Function; import com.gs.collections.api.block.predicate.Predicate; @@ -49,23 +66,6 @@ import com.gs.collections.impl.list.mutable.FastList; import kafka.common.ErrorMapping; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.context.SmartLifecycle; -import org.springframework.integration.kafka.core.BrokerAddress; -import org.springframework.integration.kafka.core.ConnectionFactory; -import org.springframework.integration.kafka.core.ConsumerException; -import org.springframework.integration.kafka.core.FetchRequest; -import org.springframework.integration.kafka.core.KafkaConsumerDefaults; -import org.springframework.integration.kafka.core.KafkaMessage; -import org.springframework.integration.kafka.core.KafkaMessageBatch; -import org.springframework.integration.kafka.core.KafkaTemplate; -import org.springframework.integration.kafka.core.Partition; -import org.springframework.integration.kafka.core.Result; -import org.springframework.util.Assert; -import org.springframework.util.CollectionUtils; - /** * @author Marius Bogoevici */ @@ -105,7 +105,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { private int queueSize = 1024; - private MessageListener messageListener; + private Object messageListener; private ErrorHandler errorHandler = new LoggingErrorHandler(); @@ -142,11 +142,16 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { this.offsetManager = offsetManager; } - public MessageListener getMessageListener() { + public Object getMessageListener() { return messageListener; } - public void setMessageListener(MessageListener messageListener) { + public void setMessageListener(Object messageListener) { + Assert.isTrue + (messageListener instanceof MessageListener + || messageListener instanceof AcknowledgingMessageListener, + "Either a " + MessageListener.class.getName() + " or a " + + AcknowledgingMessageListener.class.getName() + " must be provided"); this.messageListener = messageListener; } @@ -297,7 +302,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { */ public class FetchTask implements Runnable { - private BrokerAddress brokerAddress; + private final BrokerAddress brokerAddress; public FetchTask(BrokerAddress brokerAddress) { this.brokerAddress = brokerAddress; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/QueueingMessageListenerInvoker.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/QueueingMessageListenerInvoker.java index 8053b76f8d..232cf6a195 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/QueueingMessageListenerInvoker.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/QueueingMessageListenerInvoker.java @@ -21,6 +21,7 @@ import java.util.concurrent.BlockingQueue; import org.springframework.context.Lifecycle; import org.springframework.integration.kafka.core.KafkaMessage; +import org.springframework.util.Assert; /** * Invokes a delegate {@link MessageListener} for all the messages passed to it, storing them @@ -34,16 +35,30 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle { private volatile boolean running = false; - private final MessageListener delegate; + private final MessageListener messageListener; + + private final AcknowledgingMessageListener acknowledgingMessageListener; private final OffsetManager offsetManager; private final ErrorHandler errorHandler; - public QueueingMessageListenerInvoker(int capacity, OffsetManager offsetManager, MessageListener delegate, + public QueueingMessageListenerInvoker(int capacity, OffsetManager offsetManager, Object delegate, ErrorHandler errorHandler) { + if (delegate instanceof MessageListener) { + this.messageListener = (MessageListener) delegate; + this.acknowledgingMessageListener = null; + } + else if (delegate instanceof AcknowledgingMessageListener) { + this.acknowledgingMessageListener = (AcknowledgingMessageListener) delegate; + this.messageListener = null; + } + else { + // it's neither, an exception will be thrown + throw new IllegalArgumentException("Either a " + MessageListener.class.getName() + " or a " + + AcknowledgingMessageListener.class.getName() + " must be provided"); + } this.offsetManager = offsetManager; - this.delegate = delegate; this.errorHandler = errorHandler; this.messages = new ArrayBlockingQueue(capacity, true); } @@ -102,7 +117,12 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle { try { KafkaMessage message = messages.take(); try { - delegate.onMessage(message); + if (messageListener != null) { + messageListener.onMessage(message); + } + else { + acknowledgingMessageListener.onMessage(message, new DefaultAcknowledgment(offsetManager, message)); + } } catch (Exception e) { if (errorHandler != null) { @@ -110,8 +130,10 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle { } } finally { - offsetManager.updateOffset(message.getMetadata().getPartition(), - message.getMetadata().getNextOffset()); + if (messageListener != null) { + offsetManager.updateOffset(message.getMetadata().getPartition(), + message.getMetadata().getNextOffset()); + } } } catch (InterruptedException e) { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaHeaders.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaHeaders.java index eec5496cfe..7d40175401 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaHeaders.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaHeaders.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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. @@ -18,6 +18,7 @@ package org.springframework.integration.kafka.support; /** * @author Artem Bilan + * @author Marius Bogoevici * @since 1.0 */ public abstract class KafkaHeaders { @@ -32,4 +33,8 @@ public abstract class KafkaHeaders { public static final String OFFSET = PREFIX + "_offset"; + public static final String NEXT_OFFSET = PREFIX + "nextOffset"; + + public static final String ACKNOWLEDGMENT = PREFIX + "acknowledgment"; + } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterTests.java index 1723ab2067..ad53de6621 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterTests.java @@ -17,18 +17,18 @@ package org.springframework.integration.kafka.listener; +import static org.hamcrest.CoreMatchers.nullValue; import static org.hamcrest.collection.IsCollectionWithSize.hasSize; import static org.hamcrest.core.IsEqual.equalTo; import static org.junit.Assert.assertThat; import static org.mockito.Mockito.mock; import java.util.ArrayList; +import java.util.Collections; +import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import com.gs.collections.api.multimap.list.MutableListMultimap; -import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap; -import kafka.message.NoCompressionCodec$; import org.junit.Rule; import org.junit.Test; @@ -40,9 +40,16 @@ import org.springframework.integration.kafka.rule.KafkaEmbedded; import org.springframework.integration.kafka.rule.KafkaRule; import org.springframework.integration.kafka.serializer.common.StringDecoder; import org.springframework.integration.kafka.support.KafkaHeaders; +import org.springframework.integration.metadata.SimpleMetadataStore; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import com.gs.collections.api.multimap.list.MutableListMultimap; +import com.gs.collections.impl.list.mutable.FastList; +import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap; + +import kafka.message.NoCompressionCodec$; + /** * @author Marius Bogoevici */ @@ -75,7 +82,7 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen int expectedMessageCount = 100; - final MutableListMultimap receivedData = + final MutableListMultimap receivedData = new SynchronizedPutFastListMultimap(); final CountDownLatch latch = new CountDownLatch(expectedMessageCount); @@ -113,7 +120,7 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessages(100, TEST_TOPIC)); - latch.await((expectedMessageCount/5000) + 1, TimeUnit.MINUTES); + latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES); kafkaMessageListenerContainer.stop(); assertThat(receivedData.valuesView().toList(), hasSize(expectedMessageCount)); @@ -124,4 +131,93 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen } + @Test + @SuppressWarnings("serial") + public void testManualAck() throws Exception { + createTopic(TEST_TOPIC, 5, 1, 1); + + ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory(); + ArrayList readPartitions = new ArrayList(); + for (int i = 0; i < 5; i++) { + readPartitions.add(new Partition(TEST_TOPIC, i)); + } + + final KafkaMessageListenerContainer kafkaMessageListenerContainer = + new KafkaMessageListenerContainer(connectionFactory, + readPartitions.toArray(new Partition[readPartitions.size()])); + MetadataStoreOffsetManager offsetManager = new MetadataStoreOffsetManager(connectionFactory); + SimpleMetadataStore metadataStore = new SimpleMetadataStore(); + offsetManager.setMetadataStore(metadataStore); + kafkaMessageListenerContainer.setOffsetManager(offsetManager); + kafkaMessageListenerContainer.setMaxFetch(100); + kafkaMessageListenerContainer.setConcurrency(2); + + int expectedMessageCount = 100; + + final MutableListMultimap receivedData = + new SynchronizedPutFastListMultimap(); + final List acknowledgments = Collections.synchronizedList(new ArrayList()); + final CountDownLatch latch = new CountDownLatch(expectedMessageCount); + + KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter = + new KafkaMessageDrivenChannelAdapter(kafkaMessageListenerContainer); + + StringDecoder decoder = new StringDecoder(); + kafkaMessageDrivenChannelAdapter.setKeyDecoder(decoder); + kafkaMessageDrivenChannelAdapter.setPayloadDecoder(decoder); + kafkaMessageDrivenChannelAdapter.setBeanFactory(mock(BeanFactory.class)); + kafkaMessageDrivenChannelAdapter.setOutputChannel(new MessageChannel() { + @Override + public boolean send(Message message) { + boolean addedSuccessfully = receivedData.put( + (Integer) message.getHeaders().get(KafkaHeaders.PARTITION_ID), + new KeyedMessageWithOffset( + (String) message.getHeaders().get(KafkaHeaders.MESSAGE_KEY), + (String) message.getPayload(), + (Long) message.getHeaders().get(KafkaHeaders.OFFSET), + Thread.currentThread().getName(), + (Integer) message.getHeaders().get(KafkaHeaders.PARTITION_ID))); + acknowledgments.add(message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class)); + latch.countDown(); + return addedSuccessfully; + } + + + @Override + public boolean send(Message message, long timeout) { + return send(message); + } + }); + + kafkaMessageDrivenChannelAdapter.setAutoCommitOffset(false); + kafkaMessageDrivenChannelAdapter.afterPropertiesSet(); + kafkaMessageDrivenChannelAdapter.start(); + + createStringProducer(NoCompressionCodec$.MODULE$.codec()).send(createMessages(100, TEST_TOPIC)); + + latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES); + kafkaMessageListenerContainer.stop(); + + assertThat(receivedData.valuesView().toList(), hasSize(expectedMessageCount)); + assertThat(latch.getCount(), equalTo(0L)); + System.out.println("All messages received ... checking "); + + validateMessageReceipt(receivedData, 2, 5, 100, expectedMessageCount, readPartitions, 1); + + // at this point, all messages have been processed but not acknowledged + for (Partition readPartition : readPartitions) { + assertThat(metadataStore.get(offsetManager.generateKey(readPartition)), nullValue()); + } + + // Now, we did acknowledge them in the reverse order. + // This way we check that only the highest value was acknowledged + for (Acknowledgment acknowledgment : FastList.newList(acknowledgments).reverseThis()) { + acknowledgment.acknowledge(); + } + + // now they are all acknowledged + for (Partition readPartition : readPartitions) { + assertThat(metadataStore.get(offsetManager.generateKey(readPartition)), equalTo(String.valueOf(20))); + } + } } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/SingleBrokerWithManualAckTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/SingleBrokerWithManualAckTests.java new file mode 100644 index 0000000000..f5345e2463 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/SingleBrokerWithManualAckTests.java @@ -0,0 +1,149 @@ +/* + * Copyright 2015 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.listener; + +import static org.hamcrest.CoreMatchers.nullValue; +import static org.hamcrest.collection.IsCollectionWithSize.hasSize; +import static org.hamcrest.core.IsEqual.equalTo; +import static org.junit.Assert.assertThat; +import static org.springframework.integration.kafka.util.MessageUtils.decodeKey; +import static org.springframework.integration.kafka.util.MessageUtils.decodePayload; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.Rule; +import org.junit.Test; + +import org.springframework.integration.kafka.core.ConnectionFactory; +import org.springframework.integration.kafka.core.KafkaMessage; +import org.springframework.integration.kafka.core.Partition; +import org.springframework.integration.kafka.rule.KafkaEmbedded; +import org.springframework.integration.metadata.SimpleMetadataStore; + +import com.gs.collections.api.multimap.list.MutableListMultimap; +import com.gs.collections.impl.list.mutable.FastList; +import com.gs.collections.impl.multimap.list.SynchronizedPutFastListMultimap; + +import kafka.serializer.StringDecoder; +import kafka.utils.VerifiableProperties; + +/** + * @author Marius Bogoevici + */ +public class SingleBrokerWithManualAckTests extends AbstractMessageListenerContainerTests { + + @Rule + public final KafkaEmbedded kafkaEmbeddedBrokerRule = new KafkaEmbedded(1); + + @Override + public KafkaEmbedded getKafkaRule() { + return kafkaEmbeddedBrokerRule; + } + + @Test(expected = IllegalArgumentException.class) + public void testMessageListenerRequiredIfAutoAckFail() throws Exception { + createTopic(TEST_TOPIC, 5, 1, 1); + ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory(); + ArrayList readPartitions = new ArrayList(); + for (int i = 0; i < 5; i++) { + if (i % 1 == 0) { + readPartitions.add(new Partition(TEST_TOPIC, i)); + } + } + final KafkaMessageListenerContainer kafkaMessageListenerContainer = new KafkaMessageListenerContainer( + connectionFactory, readPartitions.toArray(new Partition[readPartitions.size()])); + kafkaMessageListenerContainer.setMaxFetch(100); + kafkaMessageListenerContainer.setConcurrency(2); + + kafkaMessageListenerContainer.setMessageListener(new Object()); + } + + @Test + public void testLowVolumeLowConcurrency() throws Exception { + createTopic(TEST_TOPIC, 5, 1, 1); + + ConnectionFactory connectionFactory = getKafkaBrokerConnectionFactory(); + ArrayList readPartitions = new ArrayList(); + for (int i = 0; i < 5; i++) { + if (i % 1 == 0) { + readPartitions.add(new Partition(TEST_TOPIC, i)); + } + } + final KafkaMessageListenerContainer kafkaMessageListenerContainer = new KafkaMessageListenerContainer( + connectionFactory, readPartitions.toArray(new Partition[readPartitions.size()])); + kafkaMessageListenerContainer.setMaxFetch(100); + kafkaMessageListenerContainer.setConcurrency(2); + MetadataStoreOffsetManager offsetManager = new MetadataStoreOffsetManager(connectionFactory); + SimpleMetadataStore metadataStore = new SimpleMetadataStore(); + offsetManager.setMetadataStore(metadataStore); + kafkaMessageListenerContainer.setOffsetManager(offsetManager); + + int expectedMessageCount = 100; + + final List acknowledgments = Collections.synchronizedList(new ArrayList()); + + final MutableListMultimap receivedData = + new SynchronizedPutFastListMultimap(); + final CountDownLatch latch = new CountDownLatch(expectedMessageCount); + kafkaMessageListenerContainer.setMessageListener(new AcknowledgingMessageListener() { + @Override + public void onMessage(KafkaMessage message, Acknowledgment acknowledgment) { + StringDecoder decoder = new StringDecoder(new VerifiableProperties()); + receivedData.put(message.getMetadata().getPartition().getId(), + new KeyedMessageWithOffset(decodeKey(message, decoder), decodePayload(message, decoder), + message.getMetadata().getOffset(), Thread.currentThread().getName(), message + .getMetadata().getPartition().getId())); + acknowledgments.add(acknowledgment); + latch.countDown(); + } + }); + + kafkaMessageListenerContainer.start(); + + createStringProducer(0).send(createMessages(100, TEST_TOPIC)); + + latch.await((expectedMessageCount / 5000) + 1, TimeUnit.MINUTES); + kafkaMessageListenerContainer.stop(); + + assertThat(receivedData.valuesView().toList(), hasSize(expectedMessageCount)); + assertThat(latch.getCount(), equalTo(0L)); + System.out.println("All messages received ... checking "); + + validateMessageReceipt(receivedData, 2, 5, 100, expectedMessageCount, readPartitions, 1); + + // at this point, all messages have been processed but not acknowledged + for (Partition readPartition : readPartitions) { + assertThat(metadataStore.get(offsetManager.generateKey(readPartition)), nullValue()); + } + + // now we did acknowledge them in the reverse order. This way we check that only + // the highest value was acknowledged + for (Acknowledgment acknowledgment : FastList.newList(acknowledgments).reverseThis()) { + acknowledgment.acknowledge(); + } + + // now they are all acknowledged + for (Partition readPartition : readPartitions) { + assertThat(metadataStore.get(offsetManager.generateKey(readPartition)), equalTo(String.valueOf(20))); + } + } + +}