From 669e48f1ee8a5f76337d32f251c922b10f4385cb Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 17 Dec 2018 14:50:56 -0500 Subject: [PATCH] GH-908: Don't cache dedicated consumer producers Fixes https://github.com/spring-projects/spring-kafka/issues/908 Zombie-fenced producers, running on container threads with topic/partition/group transactional ids must not be added to he general producer cache for use by other arbitrary producer operations. These producers are maintained in their own cache, keyed by the transactional id. Add tests to verify these producers are not cached and that a producer used within a nested transaction is added to the cache. **cherry-pick to 2.1.x, 2.0.x; backport PR will be published for 1.3.x after review/merge** # Conflicts: # spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java # Conflicts: # spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java --- .../core/DefaultKafkaProducerFactory.java | 85 ++++++++++--- .../DefaultKafkaConsumerFactoryTests.java | 115 +++++++++++++++++- 2 files changed, 179 insertions(+), 21 deletions(-) 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 87eee40d..6f532e30 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 @@ -48,6 +48,7 @@ import org.apache.kafka.common.serialization.Serializer; import org.springframework.beans.factory.DisposableBean; import org.springframework.context.Lifecycle; import org.springframework.kafka.support.TransactionSupport; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; /** @@ -304,12 +305,13 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, } private CloseSafeProducer 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); + Producer newProducer; + Map newProducerConfigs = new HashMap<>(this.configs); + newProducerConfigs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, this.transactionIdPrefix + suffix); + newProducer = new KafkaProducer(newProducerConfigs, this.keySerializer, this.valueSerializer); + newProducer.initTransactions(); + return new CloseSafeProducer(newProducer, this.cache, remover, + (String) newProducerConfigs.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG)); } protected BlockingQueue> getCache() { @@ -343,6 +345,8 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, private final Consumer> removeConsumerProducer; + private final String txId; + private volatile boolean txFailed; CloseSafeProducer(Producer delegate) { @@ -356,9 +360,17 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, CloseSafeProducer(Producer delegate, BlockingQueue> cache, Consumer> removeConsumerProducer) { + + this(delegate, cache, removeConsumerProducer, null); + } + + CloseSafeProducer(Producer delegate, @Nullable BlockingQueue> cache, + @Nullable Consumer> removeConsumerProducer, @Nullable String txId) { + this.delegate = delegate; this.cache = cache; this.removeConsumerProducer = removeConsumerProducer; + this.txId = txId; } @Override @@ -393,10 +405,16 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, @Override public void beginTransaction() throws ProducerFencedException { + if (logger.isDebugEnabled()) { + logger.debug("beginTransaction: " + this); + } try { this.delegate.beginTransaction(); } catch (RuntimeException e) { + if (logger.isErrorEnabled()) { + logger.error("beginTransaction failed: " + this, e); + } this.txFailed = true; throw e; } @@ -405,15 +423,22 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, @Override public void sendOffsetsToTransaction(Map offsets, String consumerGroupId) throws ProducerFencedException { + this.delegate.sendOffsetsToTransaction(offsets, consumerGroupId); } @Override public void commitTransaction() throws ProducerFencedException { + if (logger.isDebugEnabled()) { + logger.debug("commitTransaction: " + this); + } try { this.delegate.commitTransaction(); } catch (RuntimeException e) { + if (logger.isErrorEnabled()) { + logger.error("commitTransaction failed: " + this, e); + } this.txFailed = true; throw e; } @@ -421,10 +446,16 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, @Override public void abortTransaction() throws ProducerFencedException { + if (logger.isDebugEnabled()) { + logger.debug("abortTransaction: " + this); + } try { this.delegate.abortTransaction(); } catch (RuntimeException e) { + if (logger.isErrorEnabled()) { + logger.error("Abort failed: " + this, e); + } this.txFailed = true; throw e; } @@ -432,34 +463,50 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, @Override public void close() { + close(0, null); + } + + @Override + public void close(long timeout, @Nullable TimeUnit unit) { if (this.cache != null) { if (this.txFailed) { - logger.warn("Error during transactional operation; producer removed from cache; possible cause: " - + "broker restarted during transaction"); - - this.delegate.close(); + if (logger.isWarnEnabled()) { + logger.warn("Error during transactional operation; producer removed from cache; possible cause: " + + "broker restarted during transaction: " + this); + } + if (unit == null) { + this.delegate.close(); + } + else { + this.delegate.close(timeout, unit); + } if (this.removeConsumerProducer != null) { this.removeConsumerProducer.accept(this); } } else { - synchronized (this) { - if (!this.cache.contains(this)) { - this.cache.offer(this); + if (this.removeConsumerProducer == null) { // dedicated consumer producers are not cached + synchronized (this) { + if (!this.cache.contains(this) + && !this.cache.offer(this)) { + if (unit == null) { + this.delegate.close(); + } + else { + this.delegate.close(timeout, unit); + } + } } } } } } - @Override - public void close(long timeout, TimeUnit unit) { - close(); - } - @Override public String toString() { - return "CloseSafeProducer [delegate=" + this.delegate + "]"; + return "CloseSafeProducer [delegate=" + this.delegate + "" + + (this.txId != null ? ", txId=" + this.txId : "") + + "]"; } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java index b1cee31c..22d5eb55 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-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. @@ -20,22 +20,50 @@ import static org.assertj.core.api.Assertions.assertThat; import java.util.Collections; import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.producer.ProducerConfig; import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.listener.KafkaMessageListenerContainer; +import org.springframework.kafka.listener.MessageListener; +import org.springframework.kafka.listener.config.ContainerProperties; +import org.springframework.kafka.support.SendResult; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.kafka.transaction.KafkaTransactionManager; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.util.concurrent.ListenableFuture; /** * @author Gary Russell * @since 1.0.6 * */ +@EmbeddedKafka(topics = { "txCache1", "txCache2", "txCacheSendFromListener" }, + brokerProperties = { + "transaction.state.log.replication.factor=1", + "transaction.state.log.min.isr=1" } +) +@RunWith(SpringRunner.class) public class DefaultKafkaConsumerFactoryTests { + @Autowired + private KafkaEmbedded embeddedKafka; + @Test public void testClientId() { Map configs = Collections.singletonMap(ConsumerConfig.CLIENT_ID_CONFIG, "foo"); - DefaultKafkaConsumerFactory factory = new DefaultKafkaConsumerFactory(configs) { + DefaultKafkaConsumerFactory factory = + new DefaultKafkaConsumerFactory(configs) { @Override protected KafkaConsumer createKafkaConsumer(Map configs) { @@ -47,4 +75,87 @@ public class DefaultKafkaConsumerFactoryTests { factory.createConsumer("-1"); } + @SuppressWarnings("unchecked") + @Test + public void testNestedTxProducerIsCached() throws Exception { + Map producerProps = KafkaTestUtils.producerProps(this.embeddedKafka); + producerProps.put(ProducerConfig.RETRIES_CONFIG, 1); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(producerProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + DefaultKafkaProducerFactory pfTx = new DefaultKafkaProducerFactory<>(producerProps); + pfTx.setTransactionIdPrefix("fooTx."); + KafkaTemplate templateTx = new KafkaTemplate<>(pfTx); + Map consumerProps = KafkaTestUtils.consumerProps("txCache1Group", "false", this.embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + ContainerProperties containerProps = new ContainerProperties("txCache1"); + CountDownLatch latch = new CountDownLatch(1); + containerProps.setMessageListener((MessageListener) r -> { + templateTx.executeInTransaction(t -> t.send("txCacheSendFromListener", "bar")); + templateTx.executeInTransaction(t -> t.send("txCacheSendFromListener", "baz")); + latch.countDown(); + }); + KafkaTransactionManager tm = new KafkaTransactionManager<>(pfTx); + containerProps.setTransactionManager(tm); + KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, + containerProps); + container.start(); + try { + ListenableFuture> future = template.send("txCache1", "foo"); + future.get(); + assertThat(KafkaTestUtils.getPropertyValue(pf, "cache", BlockingQueue.class)).hasSize(0); + assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue(); + assertThat(KafkaTestUtils.getPropertyValue(pfTx, "cache", BlockingQueue.class)).hasSize(1); + } + finally { + container.stop(); + pf.destroy(); + pfTx.destroy(); + } + } + + @SuppressWarnings("unchecked") + @Test + public void testContainerTxProducerIsNotCached() throws Exception { + Map producerProps = KafkaTestUtils.producerProps(this.embeddedKafka); + producerProps.put(ProducerConfig.RETRIES_CONFIG, 1); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(producerProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + DefaultKafkaProducerFactory pfTx = new DefaultKafkaProducerFactory<>(producerProps); + pfTx.setTransactionIdPrefix("fooTx."); + KafkaTemplate templateTx = new KafkaTemplate<>(pfTx); + Map consumerProps = KafkaTestUtils.consumerProps("txCache2Group", "false", this.embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + ContainerProperties containerProps = new ContainerProperties("txCache2"); + CountDownLatch latch = new CountDownLatch(1); + containerProps.setMessageListener((MessageListener) r -> { + templateTx.send("txCacheSendFromListener", "bar"); + templateTx.send("txCacheSendFromListener", "baz"); + latch.countDown(); + }); + KafkaTransactionManager tm = new KafkaTransactionManager<>(pfTx); + containerProps.setTransactionManager(tm); + KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, + containerProps); + container.start(); + try { + ListenableFuture> future = template.send("txCache2", "foo"); + future.get(); + assertThat(KafkaTestUtils.getPropertyValue(pf, "cache", BlockingQueue.class)).hasSize(0); + assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue(); + assertThat(KafkaTestUtils.getPropertyValue(pfTx, "cache", BlockingQueue.class)).hasSize(0); + } + finally { + container.stop(); + pf.destroy(); + pfTx.destroy(); + } + } + + @Configuration + public static class Config { + + } + }