diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBindingRebalanceListener.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBindingRebalanceListener.java index a856362ad..4bcc62a49 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBindingRebalanceListener.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBindingRebalanceListener.java @@ -55,11 +55,16 @@ public interface KafkaBindingRebalanceListener { /** * Invoked when partitions are initially assigned or after a rebalance. Applications - * might only want to perform seek operations on an initial assignment. + * might only want to perform seek operations on an initial assignment. While the + * 'initial' argument is true for each thread (when concurrency is greater than 1), + * implementations should keep track of exactly which partitions have been sought. + * There is a race in that a rebalance could occur during startup and so a topic/ + * partition that has been sought on one thread may be re-assigned to another + * thread and you may not wish to re-seek it at that time. * @param bindingName the name of the binding. * @param consumer the consumer. * @param partitions the partitions. - * @param initial true if this is the initial assignment. + * @param initial true if this is the initial assignment on the current thread. */ default void onPartitionsAssigned(String bindingName, Consumer consumer, Collection partitions, boolean initial) { diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index b70966455..7e137edf1 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -28,9 +28,9 @@ import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Predicate; import java.util.regex.Pattern; @@ -582,6 +582,7 @@ public class KafkaMessageChannelBinder extends public void setupRebalanceListener( final ExtendedConsumerProperties extendedConsumerProperties, final ContainerProperties containerProperties) { + Assert.isTrue(!extendedConsumerProperties.getExtension().isResetOffsets(), "'resetOffsets' cannot be set when a KafkaBindingRebalanceListener is provided"); final String bindingName = bindingNameHolder.get(); @@ -591,7 +592,7 @@ public class KafkaMessageChannelBinder extends containerProperties .setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() { - private boolean initial = true; + private final ThreadLocal initialAssignment = new ThreadLocal<>(); @Override public void onPartitionsRevokedBeforeCommit(Consumer consumer, @@ -613,11 +614,15 @@ public class KafkaMessageChannelBinder extends public void onPartitionsAssigned(Consumer consumer, Collection partitions) { try { + Boolean initial = this.initialAssignment.get(); + if (initial == null) { + initial = Boolean.TRUE; + } userRebalanceListener.onPartitionsAssigned(bindingName, - consumer, partitions, this.initial); + consumer, partitions, initial); } finally { - this.initial = false; + this.initialAssignment.set(Boolean.FALSE); } } @@ -664,20 +669,22 @@ public class KafkaMessageChannelBinder extends boolean resetOffsets = extendedConsumerProperties.getExtension().isResetOffsets(); final Object resetTo = consumerFactory.getConfigurationProperties() .get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG); - final AtomicBoolean initialAssignment = new AtomicBoolean(true); if (!"earliest".equals(resetTo) && !"latest".equals(resetTo)) { logger.warn("no (or unknown) " + ConsumerConfig.AUTO_OFFSET_RESET_CONFIG + " property cannot reset"); resetOffsets = false; } if (groupManagement && resetOffsets) { - containerProperties - .setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() { + Set sought = ConcurrentHashMap.newKeySet(); + containerProperties.setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() { @Override public void onPartitionsRevokedBeforeCommit( Consumer consumer, Collection tps) { - // no op + + if (logger.isInfoEnabled()) { + logger.info("Partitions revoked: " + tps); + } } @Override @@ -687,14 +694,23 @@ public class KafkaMessageChannelBinder extends } @Override - public void onPartitionsAssigned(Consumer consumer, - Collection tps) { - if (initialAssignment.getAndSet(false)) { + public void onPartitionsAssigned(Consumer consumer, Collection tps) { + if (logger.isInfoEnabled()) { + logger.info("Partitions assigned: " + tps); + } + List toSeek = tps.stream() + .filter(tp -> { + boolean shouldSeek = !sought.contains(tp); + sought.add(tp); + return shouldSeek; + }) + .collect(Collectors.toList()); + if (toSeek.size() > 0) { if ("earliest".equals(resetTo)) { - consumer.seekToBeginning(tps); + consumer.seekToBeginning(toSeek); } else if ("latest".equals(resetTo)) { - consumer.seekToEnd(tps); + consumer.seekToEnd(toSeek); } } } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 2abf13a02..f69976e20 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -33,6 +33,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.IntStream; import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.kafka.clients.admin.AdminClient; @@ -1522,9 +1523,9 @@ public class KafkaBinderTests extends ExtendedProducerProperties producerProperties = createProducerProperties(); - ((GenericApplicationContext) this.applicationContext).registerBean("pkExtractor", + this.applicationContext.registerBean("pkExtractor", PartitionTestSupport.class, () -> new PartitionTestSupport()); - ((GenericApplicationContext) this.applicationContext).registerBean("pkSelector", + this.applicationContext.registerBean("pkSelector", PartitionTestSupport.class, () -> new PartitionTestSupport()); producerProperties.setPartitionKeyExtractorName("pkExtractor"); producerProperties.setPartitionSelectorName("pkSelector"); @@ -1650,9 +1651,9 @@ public class KafkaBinderTests extends Binder binder = getBinder(); ExtendedProducerProperties properties = createProducerProperties(); properties.setHeaderMode(HeaderMode.none); - ((GenericApplicationContext) this.applicationContext).registerBean("pkExtractor", + this.applicationContext.registerBean("pkExtractor", RawKafkaPartitionTestSupport.class, () -> new RawKafkaPartitionTestSupport()); - ((GenericApplicationContext) this.applicationContext).registerBean("pkSelector", + this.applicationContext.registerBean("pkSelector", RawKafkaPartitionTestSupport.class, () -> new RawKafkaPartitionTestSupport()); properties.setPartitionKeyExtractorName("pkExtractor"); properties.setPartitionSelectorName("pkSelector"); @@ -3016,6 +3017,81 @@ public class KafkaBinderTests extends } } + @Test + @SuppressWarnings("unchecked") + public void testResetOffsets() throws Exception { + Binding producerBinding = null; + Binding consumerBinding = null; + try { + String testPayload = "test"; + + ExtendedProducerProperties producerProperties = createProducerProperties(); + + DirectChannel moduleOutputChannel = createBindableChannel("output", + createProducerBindingProperties(producerProperties)); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.setConcurrency(2); + consumerProperties.setInstanceCount(5); // 10 partitions across 2 threads + consumerProperties.getExtension().setResetOffsets(true); + + DirectChannel moduleInputChannel = createBindableChannel("input", + createConsumerBindingProperties(consumerProperties)); + + String testTopicName = "existing" + System.currentTimeMillis(); + KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties(); + configurationProperties.setAutoAddPartitions(true); + Binder binder = getBinder(configurationProperties); + producerBinding = binder.bindProducer(testTopicName, moduleOutputChannel, + producerProperties); + + consumerBinding = binder.bindConsumer(testTopicName, "testReset", + moduleInputChannel, consumerProperties); + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + IntStream.range(0, 10).forEach(i -> moduleOutputChannel.send(MessageBuilder.withPayload(testPayload) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) + .setHeader(KafkaHeaders.PARTITION_ID, i) + .build())); + CountDownLatch latch1 = new CountDownLatch(10); + CountDownLatch latch2 = new CountDownLatch(20); + AtomicReference> inboundMessageRef = new AtomicReference<>(); + AtomicInteger received = new AtomicInteger(); + moduleInputChannel.subscribe(message1 -> { + try { + inboundMessageRef.set((Message) message1); + } + finally { + received.incrementAndGet(); + latch1.countDown(); + latch2.countDown(); + } + }); + assertThat(latch1.await(10, TimeUnit.SECONDS)).as("Failed to receive messages").isTrue(); + consumerBinding.unbind(); + consumerBinding = binder.bindConsumer(testTopicName, "testReset", + moduleInputChannel, consumerProperties); + assertThat(latch2.await(10, TimeUnit.SECONDS)).as("Failed to receive message").isTrue(); + binder.bindConsumer(testTopicName + "-x", "testReset", + moduleInputChannel, consumerProperties).unbind(); // cause another rebalance + assertThat(received.get()).as("Unexpected reset").isEqualTo(20); + + assertThat(inboundMessageRef.get()).isNotNull(); + assertThat(inboundMessageRef.get().getPayload()).isEqualTo("test".getBytes()); + assertThat(inboundMessageRef.get().getHeaders()).containsEntry("contentType", + MimeTypeUtils.TEXT_PLAIN); + } + finally { + if (producerBinding != null) { + producerBinding.unbind(); + } + if (consumerBinding != null) { + consumerBinding.unbind(); + } + } + } + + private final class FailingInvocationCountingMessageHandler implements MessageHandler {