diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java index 48c52a7e..bb38f8cf 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java @@ -18,13 +18,16 @@ package org.springframework.kafka.core; import java.util.Collections; import java.util.HashMap; +import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Map.Entry; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -48,6 +51,7 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.ApplicationListener; import org.springframework.context.event.ContextStoppedEvent; +import org.springframework.kafka.support.TransactionSupport; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -91,6 +95,8 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, private final BlockingQueue> cache = new LinkedBlockingQueue<>(); + private final Map> consumerProducers = new HashMap<>(); + private volatile CloseSafeProducer producer; private Serializer keySerializer; @@ -103,6 +109,8 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, private ApplicationContext applicationContext; + private boolean producerPerConsumerPartition = true; + /** * Construct a factory with the provided configuration. * @param configs the configuration. @@ -170,6 +178,17 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, } } + /** + * Set to false to revert to the previous behavior of a simple incrementing + * trasactional.id suffix for each producer instead of maintaining a producer + * for each group/topic/partition. + * @param producerPerConsumerPartition false to revert. + * @since 1.3.7 + */ + public void setProducerPerConsumerPartition(boolean producerPerConsumerPartition) { + this.producerPerConsumerPartition = producerPerConsumerPartition; + } + /** * Return an unmodifiable reference to the configuration map for this factory. * Useful for cloning to make a similar factory. @@ -203,6 +222,11 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, } producer = this.cache.poll(); } + synchronized (this.consumerProducers) { + this.consumerProducers.forEach( + (k, v) -> ((CloseSafeProducer) v).delegate.close(this.physicalCloseTimeout, TimeUnit.SECONDS)); + this.consumerProducers.clear(); + } } @Override @@ -258,7 +282,12 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, @Override public Producer createProducer() { if (this.transactionIdPrefix != null) { - return createTransactionalProducer(); + if (this.producerPerConsumerPartition) { + return createTransactionalProducerForPartition(); + } + else { + return createTransactionalProducer(); + } } if (this.producer == null) { synchronized (this) { @@ -279,6 +308,37 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, return new KafkaProducer(this.configs, this.keySerializer, this.valueSerializer); } + private Producer createTransactionalProducerForPartition() { + String suffix = TransactionSupport.getTransactionIdSuffix(); + if (suffix == null) { + return createTransactionalProducer(); + } + else { + synchronized (this.consumerProducers) { + if (!this.consumerProducers.containsKey(suffix)) { + Producer newProducer = doCreateTxProducer(suffix, this::removeConsumerProducer); + this.consumerProducers.put(suffix, newProducer); + return newProducer; + } + else { + return this.consumerProducers.get(suffix); + } + } + } + } + + private void removeConsumerProducer(CloseSafeProducer producer) { + synchronized (this.consumerProducers) { + Iterator>> iterator = this.consumerProducers.entrySet().iterator(); + while (iterator.hasNext()) { + if (iterator.next().getValue().equals(producer)) { + iterator.remove(); + break; + } + } + } + } + /** * Subclasses must return a producer from the {@link #getCache()} or a * new raw producer wrapped in a {@link CloseSafeProducer}. @@ -288,18 +348,22 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, protected Producer createTransactionalProducer() { Producer producer = this.cache.poll(); if (producer == null) { - Map configs = new HashMap<>(this.configs); - configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, - this.transactionIdPrefix + this.transactionIdSuffix.getAndIncrement()); - producer = new KafkaProducer(configs, this.keySerializer, this.valueSerializer); - producer.initTransactions(); - return new CloseSafeProducer(producer, this.cache); + return doCreateTxProducer("" + this.transactionIdSuffix.getAndIncrement(), null); } else { return producer; } } + private Producer doCreateTxProducer(String suffix, Consumer> remover) { + Producer producer; + Map configs = new HashMap<>(this.configs); + configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, this.transactionIdPrefix + suffix); + producer = new KafkaProducer(configs, this.keySerializer, this.valueSerializer); + producer.initTransactions(); + return new CloseSafeProducer(producer, this.cache, remover); + } + protected BlockingQueue> getCache() { return this.cache; } @@ -317,16 +381,24 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, private final BlockingQueue> cache; + private final Consumer> removeConsumerProducer; + private volatile boolean txFailed; CloseSafeProducer(Producer delegate) { - this(delegate, null); + this(delegate, null, null); Assert.isTrue(!(delegate instanceof CloseSafeProducer), "Cannot double-wrap a producer"); } CloseSafeProducer(Producer delegate, BlockingQueue> cache) { + this(delegate, cache, null); + } + + CloseSafeProducer(Producer delegate, BlockingQueue> cache, + Consumer> removeConsumerProducer) { this.delegate = delegate; this.cache = cache; + this.removeConsumerProducer = removeConsumerProducer; } @Override @@ -406,6 +478,9 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, + "broker restarted during transaction"); this.delegate.close(); + if (this.removeConsumerProducer != null) { + this.removeConsumerProducer.accept(this); + } } else { synchronized (this) { 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 cefa8137..5e77bc23 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 @@ -67,6 +67,7 @@ import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.LogIfLevelEnabled; import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition; +import org.springframework.kafka.support.TransactionSupport; import org.springframework.kafka.support.serializer.DeserializationException; import org.springframework.kafka.transaction.KafkaAwareTransactionManager; import org.springframework.scheduling.SchedulingAwareRunnable; @@ -1057,6 +1058,8 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener this.logger.trace("Processing " + record); } try { + TransactionSupport.setTransactionIdSuffix( + this.consumerGroupId + "." + record.topic() + "." + record.partition()); this.transactionTemplate.execute(new TransactionCallbackWithoutResult() { @Override @@ -1083,6 +1086,9 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } getAfterRollbackProcessor().process(unprocessed, this.consumer, e, true); } + finally { + TransactionSupport.clearTransactionIdSuffix(); + } } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/TransactionSupport.java b/spring-kafka/src/main/java/org/springframework/kafka/support/TransactionSupport.java new file mode 100644 index 00000000..6afb207c --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/TransactionSupport.java @@ -0,0 +1,46 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.support; + +/** + * Utilities for supporting transactions. + * + * @author Gary Russell + * @since 1.3.7 + * + */ +public final class TransactionSupport { + + private static final ThreadLocal transactionIdSuffix = new ThreadLocal<>(); + + private TransactionSupport() { + super(); + } + + public static void setTransactionIdSuffix(String suffix) { + transactionIdSuffix.set(suffix); + } + + public static String getTransactionIdSuffix() { + return transactionIdSuffix.get(); + } + + public static void clearTransactionIdSuffix() { + transactionIdSuffix.remove(); + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java index b527994e..0f7375f9 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java @@ -68,6 +68,7 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.core.ProducerFactoryUtils; import org.springframework.kafka.event.ConsumerStoppedEvent; import org.springframework.kafka.support.DefaultKafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; @@ -393,6 +394,7 @@ public class TransactionalContainerTests { verify(pf).createProducer(); } + @SuppressWarnings("unchecked") @Test public void testRollbackRecord() throws Exception { logger.info("Start testRollbackRecord"); @@ -413,6 +415,7 @@ public class TransactionalContainerTests { final KafkaTemplate template = new KafkaTemplate<>(pf); final AtomicBoolean failed = new AtomicBoolean(); final CountDownLatch latch = new CountDownLatch(3); + final AtomicReference transactionalId = new AtomicReference<>(); containerProps.setMessageListener((MessageListener) message -> { latch.countDown(); if (failed.compareAndSet(false, true)) { @@ -424,6 +427,9 @@ public class TransactionalContainerTests { if (message.topic().equals(topic1)) { template.send(topic2, "bar"); template.flush(); + transactionalId.set(KafkaTestUtils.getPropertyValue( + ProducerFactoryUtils.getTransactionalResourceHolder(pf).getProducer(), + "delegate.transactionManager.transactionalId", String.class)); } }); @@ -466,8 +472,10 @@ public class TransactionalContainerTests { assertThat(records.count()).isEqualTo(0); // depending on timing, the position might include the offset representing the commit in the log assertThat(consumer.position(new TopicPartition(topic1, 0))).isGreaterThanOrEqualTo(1L); + assertThat(transactionalId.get()).startsWith("rr.group.txTopic"); logger.info("Stop testRollbackRecord"); pf.destroy(); + assertThat(KafkaTestUtils.getPropertyValue(pf, "consumerProducers", Map.class)).isEmpty(); consumer.close(); } diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index d892582c..82b6d44e 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -256,7 +256,12 @@ Spring for Apache Kafka adds support in several ways. Transactions are enabled by providing the `DefaultKafkaProducerFactory` with a `transactionIdPrefix`. In that case, instead of managing a single shared `Producer`, the factory maintains a cache of transactional producers. When the user `close()` s a producer, it is returned to the cache for reuse instead of actually being closed. -The `transactional.id` property of each producer is `transactionIdPrefix` + `n`, where `n` starts with `0` and is incremented for each new producer. +The `transactional.id` property of each producer is `transactionIdPrefix` + `n`, where `n` starts with `0` and is incremented for each new producer, unless the transaction is started by a listener container with a record-based listener. +In that case, the `transactional.id` is `...`; this is to properly support fencing zombies https://www.confluent.io/blog/transactions-apache-kafka/[as described here]. +This new behavior was added in versions 1.3.7, 2.0.6, 2.1.10, and 2.2.0. +If you wish to revert to the previous behavior, set the `producerPerConsumerPartition` property on the `DefaultKafkaProducerFactory` to `false`. + +NOTE: While transactions are supported with batch listeners, zombie fencing cannot be supported because a batch may contain records from multiple topics/partitions. ====== KafkaTransactionManager diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 0d3ab5e4..a65c4880 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -79,3 +79,9 @@ See <> for more information. The streams configuration bean must now be a simple `Properties` object instead of a `StreamsConfig`. See <> for more information. + + +==== Transactional Id + +When a transaction is started by the listener container, the `transactional.id` is now the `transactionIdPrefix` appended with `..`. +This is to allow proper fencing of zombies https://www.confluent.io/blog/transactions-apache-kafka/[as described here].