From 06616f73eda4ea7f341c3d43cf211d6f4eb54211 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 2 Jun 2016 17:26:44 -0400 Subject: [PATCH] GH-89: Add Support for Topic Seek Fixes GH-89 (https://github.com/spring-projects/spring-kafka/issues/89) * Introduce `TopicPartitionInitialOffset`, where it utilizes `TopicPartition` and `Long initialOffset` * The `initialOffset` can be: - `null` - do nothing; - positive (including `0`) - absolute offset - negative - the offset relative to the current last offset of the partition: `consumer.seekToEnd() + initialOffset` * Rework everything around to rely on a new `TopicPartitionInitialOffset` abstraction * The logic in the `KafkaMessageListenerContainer.ListenerConsumer.initPartitionsIfNeeded()` reworked to in favor of a new abstraction * remove redundant `recentOffset` * Reflect new `TopicPartitionInitialOffset` in the docs Add `@PartitionOffset` support for the `@TopicPartition` Polishing --- ...kaListenerAnnotationBeanPostProcessor.java | 78 ++++++++++++---- .../kafka/annotation/PartitionOffset.java | 48 ++++++++++ .../kafka/annotation/TopicPartition.java | 9 ++ .../config/AbstractKafkaListenerEndpoint.java | 17 ++-- ...ncurrentKafkaListenerContainerFactory.java | 7 +- .../kafka/config/KafkaListenerEndpoint.java | 5 +- .../ConcurrentMessageListenerContainer.java | 19 ++-- .../KafkaMessageListenerContainer.java | 56 ++++++----- .../listener/config/ContainerProperties.java | 35 ++----- .../support/TopicPartitionInitialOffset.java | 93 +++++++++++++++++++ .../EnableKafkaIntegrationTests.java | 3 +- ...ncurrentMessageListenerContainerTests.java | 48 +++++++--- src/reference/asciidoc/kafka.adoc | 20 ++-- 13 files changed, 324 insertions(+), 114 deletions(-) create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/annotation/PartitionOffset.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/support/TopicPartitionInitialOffset.java diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java index 092a3f6d..5dd9ccba 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java @@ -56,6 +56,7 @@ import org.springframework.kafka.config.KafkaListenerEndpointRegistrar; import org.springframework.kafka.config.KafkaListenerEndpointRegistry; import org.springframework.kafka.config.MethodKafkaListenerEndpoint; import org.springframework.kafka.config.MultiMethodKafkaListenerEndpoint; +import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory; import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; @@ -85,6 +86,7 @@ import org.springframework.util.StringUtils; * @author Stephane Nicoll * @author Juergen Hoeller * @author Gary Russell + * @author Artem Bilan * * @see KafkaListener * @see EnableKafka @@ -100,7 +102,7 @@ public class KafkaListenerAnnotationBeanPostProcessor /** * The bean name of the default {@link org.springframework.kafka.config.KafkaListenerContainerFactory}. */ - static final String DEFAULT_KAFKA_LISTENER_CONTAINER_FACTORY_BEAN_NAME = "kafkaListenerContainerFactory"; + public static final String DEFAULT_KAFKA_LISTENER_CONTAINER_FACTORY_BEAN_NAME = "kafkaListenerContainerFactory"; private final Set> nonAnnotatedClasses = Collections.newSetFromMap(new ConcurrentHashMap, Boolean>(64)); @@ -402,15 +404,15 @@ public class KafkaListenerAnnotationBeanPostProcessor } } - private org.apache.kafka.common.TopicPartition[] resolveTopicPartitions(KafkaListener kafkaListener) { + private TopicPartitionInitialOffset[] resolveTopicPartitions(KafkaListener kafkaListener) { TopicPartition[] topicPartitions = kafkaListener.topicPartitions(); - List result = new ArrayList<>(); + List result = new ArrayList<>(); if (topicPartitions.length > 0) { for (TopicPartition topicPartition : topicPartitions) { result.addAll(resolveTopicPartitionsList(topicPartition)); } } - return result.toArray(new org.apache.kafka.common.TopicPartition[result.size()]); + return result.toArray(new TopicPartitionInitialOffset[result.size()]); } private String[] resolveTopics(KafkaListener kafkaListener) { @@ -444,18 +446,62 @@ public class KafkaListenerAnnotationBeanPostProcessor return pattern; } - private List resolveTopicPartitionsList(TopicPartition topicPartition) { + private List resolveTopicPartitionsList(TopicPartition topicPartition) { Object topic = resolveExpression(topicPartition.topic()); Assert.state(topic instanceof String, "topic in @TopicPartition must resolve to a String, not " + topic.getClass()); Assert.state(StringUtils.hasText((String) topic), "topic in @TopicPartition must not be empty"); String[] partitions = topicPartition.partitions(); - Assert.state(partitions.length > 0, - "At least one partition required in @TopicPartition for topic '" + topic + "'"); - List result = new ArrayList<>(); - if (partitions.length > 0) { - for (int i = 0; i < partitions.length; i++) { - resolvePartitionAsInteger((String) topic, resolveExpression(partitions[i]), result); + PartitionOffset[] partitionOffsets = topicPartition.partitionOffsets(); + Assert.state(partitions.length > 0 || partitionOffsets.length > 0, + "At least one 'partition' or 'partitionOffset' required in @TopicPartition for topic '" + topic + "'"); + List result = new ArrayList<>(); + for (int i = 0; i < partitions.length; i++) { + resolvePartitionAsInteger((String) topic, resolveExpression(partitions[i]), result); + } + + for (PartitionOffset partitionOffset : partitionOffsets) { + Object partitionValue = resolveExpression(partitionOffset.partition()); + Integer partition; + if (partitionValue instanceof String) { + Assert.state(StringUtils.hasText((String) partitionValue), + "partition in @PartitionOffset for topic '" + topic + "' cannot be empty"); + partition = Integer.valueOf((String) partitionValue); + } + else if (partitionValue instanceof Integer) { + partition = (Integer) partitionValue; + } + else { + throw new IllegalArgumentException(String.format( + "@PartitionOffset for topic '%s' can't resolve '%s' as an Integer or String, resolved to '%s'", + topic, partitionOffset.partition(), partitionValue.getClass())); + } + + Object initialOffsetValue = resolveExpression(partitionOffset.initialOffset()); + Long initialOffset; + if (initialOffsetValue instanceof String) { + Assert.state(StringUtils.hasText((String) initialOffsetValue), + "'initialOffset' in @PartitionOffset for topic '" + topic + "' cannot be empty"); + initialOffset = Long.valueOf((String) initialOffsetValue); + } + else if (initialOffsetValue instanceof Long) { + initialOffset = (Long) initialOffsetValue; + } + else { + throw new IllegalArgumentException(String.format( + "@PartitionOffset for topic '%s' can't resolve '%s' as an Long or String, resolved to '%s'", + topic, partitionOffset.initialOffset(), initialOffsetValue.getClass())); + } + + TopicPartitionInitialOffset topicPartitionOffset = + new TopicPartitionInitialOffset((String) topic, partition, initialOffset); + if (!result.contains(topicPartitionOffset)) { + result.add(topicPartitionOffset); + } + else { + throw new IllegalArgumentException( + String.format("@TopicPartition can't have the same partition configuration twice: [%s]", + topicPartitionOffset)); } } return result; @@ -484,7 +530,7 @@ public class KafkaListenerAnnotationBeanPostProcessor @SuppressWarnings("unchecked") private void resolvePartitionAsInteger(String topic, Object resolvedValue, - List result) { + List result) { if (resolvedValue instanceof String[]) { for (Object object : (String[]) resolvedValue) { resolvePartitionAsInteger(topic, object, result); @@ -493,15 +539,15 @@ public class KafkaListenerAnnotationBeanPostProcessor else if (resolvedValue instanceof String) { Assert.state(StringUtils.hasText((String) resolvedValue), "partition in @TopicPartition for topic '" + topic + "' cannot be empty"); - result.add(new org.apache.kafka.common.TopicPartition(topic, Integer.valueOf((String) resolvedValue))); + result.add(new TopicPartitionInitialOffset(topic, Integer.valueOf((String) resolvedValue))); } else if (resolvedValue instanceof Integer[]) { for (Integer partition : (Integer[]) resolvedValue) { - result.add(new org.apache.kafka.common.TopicPartition(topic, partition)); + result.add(new TopicPartitionInitialOffset(topic, partition)); } } else if (resolvedValue instanceof Integer) { - result.add(new org.apache.kafka.common.TopicPartition(topic, (Integer) resolvedValue)); + result.add(new TopicPartitionInitialOffset(topic, (Integer) resolvedValue)); } else if (resolvedValue instanceof Iterable) { for (Object object : (Iterable) resolvedValue) { @@ -510,7 +556,7 @@ public class KafkaListenerAnnotationBeanPostProcessor } else { throw new IllegalArgumentException(String.format( - "@KafKaListener can't resolve '%s' as an Integer or String", resolvedValue)); + "@KafKaListener for topic '%s' can't resolve '%s' as an Integer or String", topic, resolvedValue)); } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/PartitionOffset.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/PartitionOffset.java new file mode 100644 index 00000000..7cd680e4 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/annotation/PartitionOffset.java @@ -0,0 +1,48 @@ +/* + * Copyright 2016 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.kafka.annotation; + +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Used to add partition/initial offset information to a {@code KafkaListener}. + * + * @author Artem Bilan + */ +@Target({}) +@Retention(RetentionPolicy.RUNTIME) +public @interface PartitionOffset { + + /** + * The partition within the topic to listen on. + * Property place holders and SpEL expressions are supported, + * which must resolve to Integer (or String that can be parsed as Integer). + * @return partition within the topic. + */ + String partition(); + + /** + * The initial offset of the {@link #partition()}. + * Property place holders and SpEL expressions are supported, + * which must resolve to Long (or String that can be parsed as Long). + * @return initial offset. + */ + String initialOffset(); + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/TopicPartition.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/TopicPartition.java index d9054b7e..617de0fe 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/annotation/TopicPartition.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/annotation/TopicPartition.java @@ -24,6 +24,7 @@ import java.lang.annotation.Target; * Used to add topic/partition information to a {@code KafkaListener}. * * @author Gary Russell + * @author Artem Bilan * */ @Target({}) @@ -40,6 +41,7 @@ public @interface TopicPartition { /** * The partitions within the topic. + * Partitions specified here can't be duplicated in {@link #partitionOffsets()}. * @return the partitions within the topic. Property place * holders and SpEL expressions are supported, which must * resolve to Integers (or Strings that can be parsed as @@ -47,4 +49,11 @@ public @interface TopicPartition { */ String[] partitions() default {}; + /** + * The partitions with initial offsets within the topic. + * Partitions specified here can't be duplicated in the {@link #partitions()}. + * @return the {@link PartitionOffset} array. + */ + PartitionOffset[] partitionOffsets() default {}; + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java index 5664fdc8..8a3f2a75 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java @@ -22,8 +22,6 @@ import java.util.Collection; import java.util.Collections; import java.util.regex.Pattern; -import org.apache.kafka.common.TopicPartition; - import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; @@ -39,6 +37,7 @@ import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapte import org.springframework.kafka.listener.adapter.RecordFilterStrategy; import org.springframework.kafka.listener.adapter.RetryingAcknowledgingMessageListenerAdapter; import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter; +import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.kafka.support.converter.MessageConverter; import org.springframework.retry.RecoveryCallback; import org.springframework.retry.support.RetryTemplate; @@ -65,7 +64,7 @@ public abstract class AbstractKafkaListenerEndpoint private Pattern topicPattern; - private final Collection topicPartitions = new ArrayList<>(); + private final Collection topicPartitions = new ArrayList<>(); private BeanFactory beanFactory; @@ -117,7 +116,7 @@ public abstract class AbstractKafkaListenerEndpoint * Set the topics to use. Either these or 'topicPattern' or 'topicPartitions' * should be provided, but not a mixture. * @param topics to set. - * @see #setTopicPartitions(TopicPartition...) + * @see #setTopicPartitions(TopicPartitionInitialOffset...) * @see #setTopicPattern(Pattern) */ public void setTopics(String... topics) { @@ -136,14 +135,14 @@ public abstract class AbstractKafkaListenerEndpoint } /** - * Set the topics to use. Either these or 'topicPattern' - * or 'topicPartitions' + * Set the topicPartitions to use. + * Either this or 'topic' or 'topicPattern' * should be provided, but not a mixture. * @param topicPartitions to set. * @see #setTopics(String...) * @see #setTopicPattern(Pattern) */ - public void setTopicPartitions(TopicPartition... topicPartitions) { + public void setTopicPartitions(TopicPartitionInitialOffset... topicPartitions) { Assert.notNull(topicPartitions, "'topics' must not be null"); this.topicPartitions.clear(); this.topicPartitions.addAll(Arrays.asList(topicPartitions)); @@ -154,7 +153,7 @@ public abstract class AbstractKafkaListenerEndpoint * @return the topicPartitions for this endpoint. */ @Override - public Collection getTopicPartitions() { + public Collection getTopicPartitions() { return Collections.unmodifiableCollection(this.topicPartitions); } @@ -162,7 +161,7 @@ public abstract class AbstractKafkaListenerEndpoint * Set the topic pattern to use. Cannot be used with * topics or topicPartitions. * @param topicPattern the pattern - * @see #setTopicPartitions(TopicPartition...) + * @see #setTopicPartitions(TopicPartitionInitialOffset...) * @see #setTopics(String...) */ public void setTopicPattern(Pattern topicPattern) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/ConcurrentKafkaListenerContainerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/config/ConcurrentKafkaListenerContainerFactory.java index 0223d4f6..6404c268 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/ConcurrentKafkaListenerContainerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/ConcurrentKafkaListenerContainerFactory.java @@ -18,10 +18,9 @@ package org.springframework.kafka.config; import java.util.Collection; -import org.apache.kafka.common.TopicPartition; - import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; import org.springframework.kafka.listener.config.ContainerProperties; +import org.springframework.kafka.support.TopicPartitionInitialOffset; /** * A {@link KafkaListenerContainerFactory} implementation to build a @@ -54,10 +53,10 @@ public class ConcurrentKafkaListenerContainerFactory @Override protected ConcurrentMessageListenerContainer createContainerInstance(KafkaListenerEndpoint endpoint) { - Collection topicPartitions = endpoint.getTopicPartitions(); + Collection topicPartitions = endpoint.getTopicPartitions(); if (!topicPartitions.isEmpty()) { ContainerProperties properties = new ContainerProperties( - topicPartitions.toArray(new TopicPartition[topicPartitions.size()])); + topicPartitions.toArray(new TopicPartitionInitialOffset[topicPartitions.size()])); return new ConcurrentMessageListenerContainer(getConsumerFactory(), properties); } else { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpoint.java b/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpoint.java index d30b0f61..a65b9a6e 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpoint.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpoint.java @@ -19,9 +19,8 @@ package org.springframework.kafka.config; import java.util.Collection; import java.util.regex.Pattern; -import org.apache.kafka.common.TopicPartition; - import org.springframework.kafka.listener.MessageListenerContainer; +import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.kafka.support.converter.MessageConverter; /** @@ -59,7 +58,7 @@ public interface KafkaListenerEndpoint { * Return the topicPartitions for this endpoint. * @return the topicPartitions for this endpoint. */ - Collection getTopicPartitions(); + Collection getTopicPartitions(); /** * Return the topicPattern for this endpoint. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java index 258c086f..f3ad5b11 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java @@ -26,6 +26,7 @@ import org.apache.kafka.common.TopicPartition; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.listener.config.ContainerProperties; +import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.util.Assert; /** @@ -42,6 +43,7 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Murali Reddy * @author Jerome Mirc + * @author Artem Bilan */ public class ConcurrentMessageListenerContainer extends AbstractMessageListenerContainer { @@ -52,15 +54,14 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis private int concurrency = 1; /** - * Construct an instance with the supplied configuration properties and specific - * topics/partitions - when using this constructor, {@link ContainerProperties#setRecentOffset(long) - * recentOffset} can be specified. + * Construct an instance with the supplied configuration properties. * The topic partitions are distributed evenly across the delegate * {@link KafkaMessageListenerContainer}s. * @param consumerFactory the consumer factory. * @param containerProperties the container properties. */ - public ConcurrentMessageListenerContainer(ConsumerFactory consumerFactory, ContainerProperties containerProperties) { + public ConcurrentMessageListenerContainer(ConsumerFactory consumerFactory, + ContainerProperties containerProperties) { super(containerProperties); Assert.notNull(consumerFactory, "A ConsumerFactory must be provided"); this.consumerFactory = consumerFactory; @@ -97,7 +98,7 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis protected void doStart() { if (!isRunning()) { ContainerProperties containerProperties = getContainerProperties(); - TopicPartition[] topicPartitions = containerProperties.getTopicPartitions(); + TopicPartitionInitialOffset[] topicPartitions = containerProperties.getTopicPartitions(); if (topicPartitions != null && this.concurrency > topicPartitions.length) { this.logger.warn("When specific partitions are provided, the concurrency must be less than or " @@ -128,19 +129,19 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis } } - private TopicPartition[] partitionSubset(ContainerProperties containerProperties, int i) { - TopicPartition[] topicPartitions = containerProperties.getTopicPartitions(); + private TopicPartitionInitialOffset[] partitionSubset(ContainerProperties containerProperties, int i) { + TopicPartitionInitialOffset[] topicPartitions = containerProperties.getTopicPartitions(); if (this.concurrency == 1) { return topicPartitions; } else { int numPartitions = topicPartitions.length; if (numPartitions == this.concurrency) { - return new TopicPartition[] { topicPartitions[i] }; + return new TopicPartitionInitialOffset[] { topicPartitions[i] }; } else { int perContainer = numPartitions / this.concurrency; - TopicPartition[] subset; + TopicPartitionInitialOffset[] subset; if (i == this.concurrency - 1) { subset = Arrays.copyOfRange(topicPartitions, i * perContainer, topicPartitions.length); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index b6477636..5e80a28c 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -16,6 +16,7 @@ package org.springframework.kafka.listener; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; @@ -52,6 +53,7 @@ import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.event.ListenerContainerIdleEvent; import org.springframework.kafka.listener.config.ContainerProperties; import org.springframework.kafka.support.Acknowledgment; +import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -75,7 +77,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private final ConsumerFactory consumerFactory; - private final TopicPartition[] topicPartitions; + private final TopicPartitionInitialOffset[] topicPartitions; private ListenerConsumer listenerConsumer; @@ -92,24 +94,23 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener */ public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, ContainerProperties containerProperties) { - this(consumerFactory, containerProperties, (TopicPartition[]) null); + this(consumerFactory, containerProperties, (TopicPartitionInitialOffset[]) null); } /** * Construct an instance with the supplied configuration properties and specific - * topics/partitions - when using this constructor, - * {@link ContainerProperties#setRecentOffset(long) recentOffset} can be specified. + * topics/partitions/initialOffsets. * @param consumerFactory the consumer factory. * @param containerProperties the container properties. * @param topicPartitions the topics/partitions; duplicates are eliminated. */ public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, - ContainerProperties containerProperties, TopicPartition... topicPartitions) { + ContainerProperties containerProperties, TopicPartitionInitialOffset... topicPartitions) { super(containerProperties); Assert.notNull(consumerFactory, "A ConsumerFactory must be provided"); this.consumerFactory = consumerFactory; if (topicPartitions != null) { - this.topicPartitions = Arrays.asList(topicPartitions).toArray(new TopicPartition[topicPartitions.length]); + this.topicPartitions = Arrays.copyOf(topicPartitions, topicPartitions.length); } else { this.topicPartitions = containerProperties.getTopicPartitions(); @@ -124,7 +125,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener */ public Collection getAssignedPartitions() { if (this.listenerConsumer.definedPartitions != null) { - return Collections.unmodifiableCollection(this.listenerConsumer.definedPartitions); + return Collections.unmodifiableCollection(this.listenerConsumer.definedPartitions.keySet()); } else if (this.listenerConsumer.assignedPartitions != null) { return Collections.unmodifiableCollection(this.listenerConsumer.assignedPartitions); @@ -163,9 +164,9 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener (getBeanName() == null ? "" : getBeanName()) + "-kafka-listener-"); getContainerProperties().setListenerTaskExecutor(listenerExecutor); } - this.listenerConsumer = new ListenerConsumer(this.listener, this.acknowledgingMessageListener, - getContainerProperties().getRecentOffset()); - this.listenerConsumerFuture = getContainerProperties().getConsumerTaskExecutor() + this.listenerConsumer = new ListenerConsumer(this.listener, this.acknowledgingMessageListener); + this.listenerConsumerFuture = getContainerProperties() + .getConsumerTaskExecutor() .submitListenable(this.listenerConsumer); } @@ -225,8 +226,6 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private final AcknowledgingMessageListener acknowledgingMessageListener; - private final long recentOffset; - private final boolean autoCommit = KafkaMessageListenerContainer.this.consumerFactory.isAutoCommit(); private final boolean isManualAck = this.containerProperties.getAckMode().equals(AckMode.MANUAL); @@ -246,7 +245,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private final ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher(); - private volatile Collection definedPartitions; + private volatile Map definedPartitions; private ConsumerRecords unsent; @@ -267,8 +266,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener */ private boolean paused; - private ListenerConsumer(MessageListener listener, AcknowledgingMessageListener ackListener, - long recentOffset) { + private ListenerConsumer(MessageListener listener, AcknowledgingMessageListener ackListener) { Assert.state(!this.isAnyManualAck || !this.autoCommit, "Consumer cannot be configured for auto commit for ackMode " + this.containerProperties.getAckMode()); final Consumer consumer = KafkaMessageListenerContainer.this.consumerFactory.createConsumer(); @@ -347,15 +345,17 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } } else { - List topicPartitions = Arrays - .asList(KafkaMessageListenerContainer.this.topicPartitions); - this.definedPartitions = topicPartitions; - consumer.assign(topicPartitions); + List topicPartitions = + Arrays.asList(KafkaMessageListenerContainer.this.topicPartitions); + this.definedPartitions = new HashMap<>(topicPartitions.size()); + for (TopicPartitionInitialOffset topicPartition : topicPartitions) { + this.definedPartitions.put(topicPartition.topicPartition(), topicPartition.initialOffset()); + } + consumer.assign(new ArrayList<>(this.definedPartitions.keySet())); } this.consumer = consumer; this.listener = listener; this.acknowledgingMessageListener = ackListener; - this.recentOffset = recentOffset; } private void startInvoker() { @@ -639,11 +639,17 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener * When using auto assignment (subscribe), the ConsumerRebalanceListener is not * called until we poll() the consumer. */ - if (this.recentOffset > 0) { - this.consumer.seekToEnd( - this.definedPartitions.toArray(new TopicPartition[this.definedPartitions.size()])); - for (TopicPartition topicPartition : this.definedPartitions) { - long newOffset = this.consumer.position(topicPartition) - this.recentOffset; + for (Entry entry : this.definedPartitions.entrySet()) { + TopicPartition topicPartition = entry.getKey(); + Long offset = entry.getValue(); + if (offset != null) { + long newOffset = offset; + + if (offset < 0) { + this.consumer.seekToEnd(topicPartition); + newOffset = this.consumer.position(topicPartition) + offset; + } + this.consumer.seek(topicPartition, newOffset); if (this.logger.isDebugEnabled()) { this.logger.debug("Reset " + topicPartition + " to offset " + newOffset); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java index 4b88c986..8d939771 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java @@ -22,7 +22,6 @@ import java.util.regex.Pattern; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.OffsetCommitCallback; -import org.apache.kafka.common.TopicPartition; import org.springframework.core.task.AsyncListenableTaskExecutor; import org.springframework.kafka.listener.AbstractMessageListenerContainer; @@ -31,12 +30,14 @@ import org.springframework.kafka.listener.AcknowledgingMessageListener; import org.springframework.kafka.listener.ErrorHandler; import org.springframework.kafka.listener.LoggingErrorHandler; import org.springframework.kafka.listener.MessageListener; +import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.util.Assert; /** * Contains runtime properties for a listener container. * * @author Gary Russell + * @author Artem Bilan */ public class ContainerProperties { @@ -57,9 +58,9 @@ public class ContainerProperties { private final Pattern topicPattern; /** - * Topics/partitions. + * Topics/partitions/initial offsets. */ - private final TopicPartition[] topicPartitions; + private final TopicPartitionInitialOffset[] topicPartitions; /** * The ack mode to use when auto ack (in the configuration properties) is false. @@ -142,13 +143,6 @@ public class ContainerProperties { */ private long shutdownTimeout = DEFAULT_SHUTDOWN_TIMEOUT; - /** - * The offset to this number of records back from the latest when starting. - * Overrides any consumer properties (earliest, latest). Only applies when - * explicit topic/partition assignment is provided. - */ - private long recentOffset; - /** * A user defined {@link ConsumerRebalanceListener} implementation. */ @@ -185,12 +179,12 @@ public class ContainerProperties { this.topicPartitions = null; } - public ContainerProperties(TopicPartition... topicPartitions) { + public ContainerProperties(TopicPartitionInitialOffset... topicPartitions) { this.topics = null; this.topicPattern = null; Assert.notEmpty(topicPartitions, "An array of topicPartitions must be provided"); this.topicPartitions = new LinkedHashSet<>(Arrays.asList(topicPartitions)) - .toArray(new TopicPartition[topicPartitions.length]); + .toArray(new TopicPartitionInitialOffset[topicPartitions.length]); } /** @@ -310,16 +304,6 @@ public class ContainerProperties { this.shutdownTimeout = shutdownTimeout; } - /** - * Set the offset to this number of records back from the latest when starting. - * Overrides any consumer properties (earliest, latest). Only applies when - * explicit topic/partition assignment is provided. - * @param recentOffset the offset from the latest; default 0. - */ - public void setRecentOffset(long recentOffset) { - this.recentOffset = recentOffset; - } - /** * Set the user defined {@link ConsumerRebalanceListener} implementation. * @param consumerRebalanceListener the {@link ConsumerRebalanceListener} instance @@ -380,7 +364,7 @@ public class ContainerProperties { return this.topicPattern; } - public TopicPartition[] getTopicPartitions() { + public TopicPartitionInitialOffset[] getTopicPartitions() { return this.topicPartitions; } @@ -432,10 +416,6 @@ public class ContainerProperties { return this.shutdownTimeout; } - public long getRecentOffset() { - return this.recentOffset; - } - public ConsumerRebalanceListener getConsumerRebalanceListener() { return this.consumerRebalanceListener; } @@ -455,4 +435,5 @@ public class ContainerProperties { public boolean isAckOnError() { return this.ackOnError; } + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/TopicPartitionInitialOffset.java b/spring-kafka/src/main/java/org/springframework/kafka/support/TopicPartitionInitialOffset.java new file mode 100644 index 00000000..c6649669 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/TopicPartitionInitialOffset.java @@ -0,0 +1,93 @@ +/* + * Copyright 2016 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.kafka.support; + +import java.util.Objects; + +import org.apache.kafka.common.TopicPartition; + +/** + * A configuration container to represent a topic name, partition number and, optionally, + * an initial offset for it. The initial offset can be: + *
    + *
  • {@code null} - do nothing;
    • + *
  • positive (including {@code 0}) - seek to the absolute offset within the partition; + *
    • + *
  • negative - seek to the offset relative to the current last offset within the + * partition: {@code consumer.seekToEnd() + initialOffset}.
    • + *
+ * Offsets are applied when the container is {@code start()}ed. + * + * @author Artem Bilan + */ +public class TopicPartitionInitialOffset { + + private final TopicPartition topicPartition; + + private final Long initialOffset; + + public TopicPartitionInitialOffset(String topic, int partition) { + this(topic, partition, null); + } + + public TopicPartitionInitialOffset(String topic, int partition, Long initialOffset) { + this.topicPartition = new TopicPartition(topic, partition); + this.initialOffset = initialOffset; + } + + public TopicPartition topicPartition() { + return this.topicPartition; + } + + public int partition() { + return this.topicPartition.partition(); + } + + public String topic() { + return this.topicPartition.topic(); + } + + public Long initialOffset() { + return this.initialOffset; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + TopicPartitionInitialOffset that = (TopicPartitionInitialOffset) o; + return Objects.equals(this.topicPartition, that.topicPartition); + } + + @Override + public int hashCode() { + return this.topicPartition.hashCode(); + } + + @Override + public String toString() { + return "TopicPartitionInitialOffset{" + + "topicPartition=" + this.topicPartition + + ", initialOffset=" + this.initialOffset + + '}'; + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index cf709d4b..4c06b484 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java @@ -453,7 +453,8 @@ public class EnableKafkaIntegrationTests { @KafkaListener(id = "fiz", topicPartitions = { @TopicPartition(topic = "annotated5", partitions = { "#{'${foo:0,1}'.split(',')}" }), - @TopicPartition(topic = "annotated6", partitions = { "0", "1" }) + @TopicPartition(topic = "annotated6", partitions = "0", + partitionOffsets = @PartitionOffset(partition = "${xxx:1}", initialOffset = "${yyy:0}")) }) public void listen5(ConsumerRecord record) { this.record = record; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java index 6c0c7e8c..15132727 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java @@ -58,6 +58,7 @@ import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode; import org.springframework.kafka.listener.config.ContainerProperties; +import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.KafkaTestUtils; @@ -220,7 +221,7 @@ public class ConcurrentMessageListenerContainerTests { public void testDefinedPartitions() throws Exception { this.logger.info("Start auto parts"); final Map props = KafkaTestUtils.consumerProps("test3", "true", embeddedKafka); - TopicPartition topic1Partition0 = new TopicPartition(topic3, 0); + TopicPartitionInitialOffset topic1Partition0 = new TopicPartitionInitialOffset(topic3, 0); final CountDownLatch initialConsumersLatch = new CountDownLatch(2); @@ -256,7 +257,7 @@ public class ConcurrentMessageListenerContainerTests { container1.setBeanName("b1"); container1.start(); - TopicPartition topic1Partition1 = new TopicPartition(topic3, 1); + TopicPartitionInitialOffset topic1Partition1 = new TopicPartitionInitialOffset(topic3, 1); ContainerProperties container2Props = new ContainerProperties(topic1Partition1); ConcurrentMessageListenerContainer container2 = new ConcurrentMessageListenerContainer<>(cf, container2Props); @@ -305,10 +306,11 @@ public class ConcurrentMessageListenerContainerTests { props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); cf = new DefaultKafkaConsumerFactory<>(props); // reset minus one + topic1Partition0 = new TopicPartitionInitialOffset(topic3, 0, -1L); + topic1Partition1 = new TopicPartitionInitialOffset(topic3, 1, -1L); ContainerProperties container4Props = new ContainerProperties(topic1Partition0, topic1Partition1); resettingContainer = new ConcurrentMessageListenerContainer<>(cf, container4Props); resettingContainer.setBeanName("b4"); - container4Props.setRecentOffset(1); final CountDownLatch latch4 = new CountDownLatch(2); final AtomicReference receivedMessage = new AtomicReference<>(); container4Props.setMessageListener((MessageListener) message -> { @@ -322,6 +324,28 @@ public class ConcurrentMessageListenerContainerTests { assertThat(receivedMessage.get()).isIn("baz", "qux"); assertThat(latch4.getCount()).isEqualTo(0L); + // reset plus one + template.sendDefault(0, 0, "FOO"); + template.sendDefault(1, 2, "BAZ"); + template.flush(); + + topic1Partition0 = new TopicPartitionInitialOffset(topic3, 0, 1L); + topic1Partition1 = new TopicPartitionInitialOffset(topic3, 1, 1L); + ContainerProperties container5Props = new ContainerProperties(topic1Partition0, topic1Partition1); + resettingContainer = new ConcurrentMessageListenerContainer<>(cf, container5Props); + resettingContainer.setBeanName("b4"); + final CountDownLatch latch5 = new CountDownLatch(4); + final List messages = new ArrayList<>(); + container5Props.setMessageListener((MessageListener) message -> { + ConcurrentMessageListenerContainerTests.this.logger.info("auto part 1: " + message); + messages.add(message.value()); + latch5.countDown(); + }); + resettingContainer.start(); + assertThat(latch5.await(60, TimeUnit.SECONDS)).isTrue(); + resettingContainer.stop(); + assertThat(messages).contains("baz", "qux", "FOO", "BAZ"); + this.logger.info("Stop auto parts"); } @@ -461,14 +485,14 @@ public class ConcurrentMessageListenerContainerTests { @Test @SuppressWarnings("unchecked") public void testConcurrencyWithPartitions() { - TopicPartition[] topic1PartitionS = new TopicPartition[]{ - new TopicPartition(topic1, 0), - new TopicPartition(topic1, 1), - new TopicPartition(topic1, 2), - new TopicPartition(topic1, 3), - new TopicPartition(topic1, 4), - new TopicPartition(topic1, 5), - new TopicPartition(topic1, 6) + TopicPartitionInitialOffset[] topic1PartitionS = new TopicPartitionInitialOffset[]{ + new TopicPartitionInitialOffset(topic1, 0), + new TopicPartitionInitialOffset(topic1, 1), + new TopicPartitionInitialOffset(topic1, 2), + new TopicPartitionInitialOffset(topic1, 3), + new TopicPartitionInitialOffset(topic1, 4), + new TopicPartitionInitialOffset(topic1, 5), + new TopicPartitionInitialOffset(topic1, 6) }; ConsumerFactory cf = mock(ConsumerFactory.class); Consumer consumer = mock(Consumer.class); @@ -495,7 +519,7 @@ public class ConcurrentMessageListenerContainerTests { assertThat(containers.size()).isEqualTo(3); for (int i = 0; i < 3; i++) { assertThat(KafkaTestUtils.getPropertyValue(containers.get(i), "topicPartitions", - TopicPartition[].class).length).isEqualTo(i < 2 ? 2 : 3); + TopicPartitionInitialOffset[].class).length).isEqualTo(i < 2 ? 2 : 3); } container.stop(); } diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index d39317ee..35cefe13 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -149,31 +149,32 @@ public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, ContainerProperties containerProperties) public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, - ContainerProperties containerProperties, TopicPartition... topicPartitions) + ContainerProperties containerProperties, TopicPartitionInitialOffset... topicPartitions) ---- Each takes a `ConsumerFactory` and information about topics and partitions, as well as other configuration in a `ContainerProperties` object. -The second constructor is used by the `ConcurrentMessageListenerContainer` (see below) to distribute `TopicPartitions` across the consumer instances. +The second constructor is used by the `ConcurrentMessageListenerContainer` (see below) to distribute `TopicPartitionInitialOffset` across the consumer instances. `ContainerProperties` has the following constructors: [source, java] ---- -public ContainerProperties(TopicPartition... topicPartitions) +public ContainerProperties(TopicPartitionInitialOffset... topicPartitions) public ContainerProperties(String... topics) public ContainerProperties(Pattern topicPattern) ---- -The first takes an array of `TopicPartition` arguments to explicitly instruct the container which partitions to use -(using the consumer `assign()` method). +The first takes an array of `TopicPartitionInitialOffset` arguments to explicitly instruct the container which partitions to use +(using the consumer `assign()` method), and with an optional initial offset: a positive value is an absolute offset; a negative value is relative to the current last offset within a partition. +The offsets are applied when the container is started. The second takes an array of topics and Kafka allocates the partitions based on the `group.id` property - distributing partitions across the group. The third uses a regex `Pattern` to select the topics. -Refer to the javadocs for `ContainerProperties` for more information about the various properties that can be set. +Refer to the JavaDocs for `ContainerProperties` for more information about the various properties that can be set. ====== ConcurrentMessageListenerContainer @@ -290,19 +291,22 @@ public Map consumerConfigs() { Notice that to set container properties, you must use the `getContainerProperties()` method on the factory. It is used as a template for the actual properties injected into the container. -You can also configure POJO listeners with explicit topics and partitions: +You can also configure POJO listeners with explicit topics and partitions (and, optionally, their initial offsets): [source, java] ---- @KafkaListener(id = "bar", topicPartitions = { @TopicPartition(topic = "topic1", partitions = { "0", "1" }), - @TopicPartition(topic = "topic2", partitions = { "0", "1" }) + @TopicPartition(topic = "topic2", partitions = "0", + partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100")) }) public void listen(ConsumerRecord record) { ... } ---- +Each partition can be specified in the `partitions` or `partitionOffsets` attribute, but not both. + When using manual `AckMode`, the listener can also be provided with the `Acknowledgment`; this example also shows how to use a different container factory.