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
This commit is contained in:
Gary Russell
committed by
Artem Bilan
Artem Bilan
parent
e36d458c35
commit
669e48f1ee
@@ -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<K, V> implements ProducerFactory<K, V>,
|
||||
}
|
||||
|
||||
private CloseSafeProducer<K, V> doCreateTxProducer(String suffix, Consumer<CloseSafeProducer<K, V>> remover) {
|
||||
Producer<K, V> producer;
|
||||
Map<String, Object> configs = new HashMap<>(this.configs);
|
||||
configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, this.transactionIdPrefix + suffix);
|
||||
producer = new KafkaProducer<K, V>(configs, this.keySerializer, this.valueSerializer);
|
||||
producer.initTransactions();
|
||||
return new CloseSafeProducer<K, V>(producer, this.cache, remover);
|
||||
Producer<K, V> newProducer;
|
||||
Map<String, Object> newProducerConfigs = new HashMap<>(this.configs);
|
||||
newProducerConfigs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, this.transactionIdPrefix + suffix);
|
||||
newProducer = new KafkaProducer<K, V>(newProducerConfigs, this.keySerializer, this.valueSerializer);
|
||||
newProducer.initTransactions();
|
||||
return new CloseSafeProducer<K, V>(newProducer, this.cache, remover,
|
||||
(String) newProducerConfigs.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG));
|
||||
}
|
||||
|
||||
protected BlockingQueue<CloseSafeProducer<K, V>> getCache() {
|
||||
@@ -343,6 +345,8 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
|
||||
|
||||
private final Consumer<CloseSafeProducer<K, V>> removeConsumerProducer;
|
||||
|
||||
private final String txId;
|
||||
|
||||
private volatile boolean txFailed;
|
||||
|
||||
CloseSafeProducer(Producer<K, V> delegate) {
|
||||
@@ -356,9 +360,17 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
|
||||
|
||||
CloseSafeProducer(Producer<K, V> delegate, BlockingQueue<CloseSafeProducer<K, V>> cache,
|
||||
Consumer<CloseSafeProducer<K, V>> removeConsumerProducer) {
|
||||
|
||||
this(delegate, cache, removeConsumerProducer, null);
|
||||
}
|
||||
|
||||
CloseSafeProducer(Producer<K, V> delegate, @Nullable BlockingQueue<CloseSafeProducer<K, V>> cache,
|
||||
@Nullable Consumer<CloseSafeProducer<K, V>> removeConsumerProducer, @Nullable String txId) {
|
||||
|
||||
this.delegate = delegate;
|
||||
this.cache = cache;
|
||||
this.removeConsumerProducer = removeConsumerProducer;
|
||||
this.txId = txId;
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -393,10 +405,16 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
|
||||
|
||||
@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<K, V> implements ProducerFactory<K, V>,
|
||||
@Override
|
||||
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> 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<K, V> implements ProducerFactory<K, V>,
|
||||
|
||||
@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<K, V> implements ProducerFactory<K, V>,
|
||||
|
||||
@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 : "")
|
||||
+ "]";
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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<String, Object> configs = Collections.singletonMap(ConsumerConfig.CLIENT_ID_CONFIG, "foo");
|
||||
DefaultKafkaConsumerFactory<String, String> factory = new DefaultKafkaConsumerFactory<String, String>(configs) {
|
||||
DefaultKafkaConsumerFactory<String, String> factory =
|
||||
new DefaultKafkaConsumerFactory<String, String>(configs) {
|
||||
|
||||
@Override
|
||||
protected KafkaConsumer<String, String> createKafkaConsumer(Map<String, Object> configs) {
|
||||
@@ -47,4 +75,87 @@ public class DefaultKafkaConsumerFactoryTests {
|
||||
factory.createConsumer("-1");
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Test
|
||||
public void testNestedTxProducerIsCached() throws Exception {
|
||||
Map<String, Object> producerProps = KafkaTestUtils.producerProps(this.embeddedKafka);
|
||||
producerProps.put(ProducerConfig.RETRIES_CONFIG, 1);
|
||||
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(producerProps);
|
||||
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
|
||||
DefaultKafkaProducerFactory<Integer, String> pfTx = new DefaultKafkaProducerFactory<>(producerProps);
|
||||
pfTx.setTransactionIdPrefix("fooTx.");
|
||||
KafkaTemplate<Integer, String> templateTx = new KafkaTemplate<>(pfTx);
|
||||
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("txCache1Group", "false", this.embeddedKafka);
|
||||
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
|
||||
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
|
||||
ContainerProperties containerProps = new ContainerProperties("txCache1");
|
||||
CountDownLatch latch = new CountDownLatch(1);
|
||||
containerProps.setMessageListener((MessageListener<Integer, String>) r -> {
|
||||
templateTx.executeInTransaction(t -> t.send("txCacheSendFromListener", "bar"));
|
||||
templateTx.executeInTransaction(t -> t.send("txCacheSendFromListener", "baz"));
|
||||
latch.countDown();
|
||||
});
|
||||
KafkaTransactionManager<Integer, String> tm = new KafkaTransactionManager<>(pfTx);
|
||||
containerProps.setTransactionManager(tm);
|
||||
KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(cf,
|
||||
containerProps);
|
||||
container.start();
|
||||
try {
|
||||
ListenableFuture<SendResult<Integer, String>> 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<String, Object> producerProps = KafkaTestUtils.producerProps(this.embeddedKafka);
|
||||
producerProps.put(ProducerConfig.RETRIES_CONFIG, 1);
|
||||
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(producerProps);
|
||||
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
|
||||
DefaultKafkaProducerFactory<Integer, String> pfTx = new DefaultKafkaProducerFactory<>(producerProps);
|
||||
pfTx.setTransactionIdPrefix("fooTx.");
|
||||
KafkaTemplate<Integer, String> templateTx = new KafkaTemplate<>(pfTx);
|
||||
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("txCache2Group", "false", this.embeddedKafka);
|
||||
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
|
||||
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
|
||||
ContainerProperties containerProps = new ContainerProperties("txCache2");
|
||||
CountDownLatch latch = new CountDownLatch(1);
|
||||
containerProps.setMessageListener((MessageListener<Integer, String>) r -> {
|
||||
templateTx.send("txCacheSendFromListener", "bar");
|
||||
templateTx.send("txCacheSendFromListener", "baz");
|
||||
latch.countDown();
|
||||
});
|
||||
KafkaTransactionManager<Integer, String> tm = new KafkaTransactionManager<>(pfTx);
|
||||
containerProps.setTransactionManager(tm);
|
||||
KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(cf,
|
||||
containerProps);
|
||||
container.start();
|
||||
try {
|
||||
ListenableFuture<SendResult<Integer, String>> 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 {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user