Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1983: Support copy of producer factory

Browse Source 
Fixes https://github.com/spring-projects/spring-kafka/issues/1983

Preserves post processors for of custom factories

Instrumentations attached to ProducerFactory instances are destroyed if the user adds configOverrides.
This behavior is unexpected for the developer, for example it will undo sleuth instrumentation
that is otherwise kept if the new KafkaTemplate(Map) constructor would be used.

* adds copy for all visible factory properties and tests
* moves the copy feature to ProducerFactory
* fixing review remarks
* removes a left over TODO
* implements a generic copy in the KafkaTemplate to avoid breaking changes
* wraps javadoc at col 90
* Clean up for code style

**Cherry-pick to `2.5.x`, `2.6.x` & `2.7.x`**

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java
#	spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.java
#	spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java
This commit is contained in:
Thomas Strauß
2021-10-21 08:10:19 -04:00
committed by Artem Bilan
parent 4a75140174
commit c4787fc892
4 changed files with 250 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java
View File

@@ -108,10 +108,11 @@ import org.springframework.util.StringUtils;
* @author Nakul Mishra
* @author Artem Bilan
* @author Chris Gilbert
* @author Thomas Strauß
*/
public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
implements ProducerFactory<K, V>, ApplicationContextAware,
BeanNameAware, ApplicationListener<ContextStoppedEvent>, DisposableBean {
BeanNameAware, ApplicationListener<ContextStoppedEvent>, DisposableBean {
private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(DefaultKafkaProducerFactory.class));
@@ -360,6 +361,63 @@ public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
this.maxAge = maxAge.toMillis();
}
/**
* Copy properties of the instance and the given properties to create a new producer factory.
* <p>If the {@link org.springframework.kafka.core.DefaultKafkaProducerFactory} makes a
* copy of itself, the transaction id prefix is recovered from the properties. If
* you want to change the ID config, add a new
* {@link org.apache.kafka.clients.producer.ProducerConfig#TRANSACTIONAL_ID_CONFIG}
* key to the override config.</p>
* @param overrideProperties the properties to be applied to the new factory
* @return {@link org.springframework.kafka.core.DefaultKafkaProducerFactory} with
* properties applied
*/
@Override
public ProducerFactory<K, V> copyWithConfigurationOverride(Map<String, Object> overrideProperties) {
Map<String, Object> producerProperties = new HashMap<>(getConfigurationProperties());
producerProperties.putAll(overrideProperties);
producerProperties = ensureExistingTransactionIdPrefixInProperties(producerProperties);
DefaultKafkaProducerFactory<K, V> newFactory =
new DefaultKafkaProducerFactory<>(producerProperties,
getKeySerializerSupplier(),
getValueSerializerSupplier());
newFactory.setPhysicalCloseTimeout((int) getPhysicalCloseTimeout().getSeconds());
newFactory.setProducerPerConsumerPartition(isProducerPerConsumerPartition());
newFactory.setProducerPerThread(isProducerPerThread());
for (ProducerPostProcessor<K, V> templatePostProcessor : getPostProcessors()) {
newFactory.addPostProcessor(templatePostProcessor);
}
for (ProducerFactory.Listener<K, V> templateListener : getListeners()) {
newFactory.addListener(templateListener);
}
return newFactory;
}
/**
* Ensures that the returned properties map contains a transaction id prefix.
* The {@link org.springframework.kafka.core.DefaultKafkaProducerFactory}
* modifies the local properties copy, the txn key is removed and
* stored locally in a property. To make a proper copy of the properties in a
* new factory, the transactionId has to be reinserted prior use.
* The incoming properties are checked for a transactionId key. If none is
* there, the one existing in the factory is added.
* @param producerProperties the properties to be used for the new factory
* @return the producerProperties or a copy with the transaction ID set
*/
private Map<String, Object> ensureExistingTransactionIdPrefixInProperties(Map<String, Object> producerProperties) {
String transactionIdPrefix = getTransactionIdPrefix();
if (StringUtils.hasText(transactionIdPrefix)) {
if (!producerProperties.containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG)) {
Map<String, Object> producerPropertiesWithTxnId = new HashMap<>(producerProperties);
producerPropertiesWithTxnId.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionIdPrefix);
return producerPropertiesWithTxnId;
}
}
return producerProperties;
}
/**
* Add a listener.
* @param listener the listener.
@@ -417,8 +475,8 @@ public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
Assert.isTrue(entry.getValue() instanceof String, () -> "'" + ProducerConfig.TRANSACTIONAL_ID_CONFIG
+ "' must be a String, not a " + entry.getClass().getName());
Assert.isTrue(this.transactionIdPrefix != null
? entry.getValue() != null
: entry.getValue() == null,
? entry.getValue() != null
: entry.getValue() == null,
"Cannot change transactional capability");
this.transactionIdPrefix = (String) entry.getValue();
}
@@ -694,7 +752,7 @@ public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
BlockingQueue<CloseSafeProducer<K, V>> txIdCache = getCache(producerToRemove.txIdPrefix);
if (producerToRemove.epoch != this.epoch.get()
|| (txIdCache != null && !txIdCache.contains(producerToRemove)
&& !txIdCache.offer(producerToRemove))) {
&& !txIdCache.offer(producerToRemove))) {
producerToRemove.closeDelegate(timeout, this.listeners);
return true;
}
@@ -942,7 +1000,7 @@ public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
LOGGER.debug(() -> toString() + " abortTransaction()");
if (this.producerFailed != null) {
LOGGER.debug(() -> "abortTransaction ignored - previous txFailed: " + this.producerFailed.getMessage()
+ ": " + this);
+ ": " + this);
}
else {
try {

67
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java
View File

@@ -75,7 +75,8 @@ import org.springframework.util.concurrent.SettableListenableFuture;
* @author Igor Stepanov
* @author Artem Bilan
* @author Biju Kunjummen
* @author Endika Guti?rrez
* @author Endika Gutierrez
* @author Thomas Strauß
*/
public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationContextAware, BeanNameAware,
ApplicationListener<ContextStoppedEvent>, DisposableBean {
@@ -158,8 +159,16 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationCo
* to occur immediately, regardless of that setting, or if you wish to block until the
* broker has acknowledged receipt according to the producer's {@code acks} property.
* If the configOverrides is not null or empty, a new
* {@link DefaultKafkaProducerFactory} will be created with merged producer properties
* with the overrides being applied after the supplied factory's properties.
* {@link ProducerFactory} will be created using
* {@link org.springframework.kafka.core.ProducerFactory#copyWithConfigurationOverride(java.util.Map)}
* The factory shall apply the overrides after the supplied factory's properties.
* The {@link org.springframework.kafka.core.ProducerPostProcessor}s from the
* original factory are copied over to keep instrumentation alive.
* Registered {@link org.springframework.kafka.core.ProducerFactory.Listener}s are
* also added to the new factory. If the factory implementation does not support
* the copy operation, a generic copy of the ProducerFactory is created which will
* be of type
* DefaultKafkaProducerFactory.
* @param producerFactory the producer factory.
* @param autoFlush true to flush after each send.
* @param configOverrides producer configuration properties to override.
@@ -174,14 +183,7 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationCo
this.micrometerEnabled = KafkaUtils.MICROMETER_PRESENT;
this.customProducerFactory = configOverrides != null && configOverrides.size() > 0;
if (this.customProducerFactory) {
Map<String, Object> configs = new HashMap<>(producerFactory.getConfigurationProperties());
configs.putAll(configOverrides);
DefaultKafkaProducerFactory<K, V> newFactory = new DefaultKafkaProducerFactory<>(configs,
producerFactory.getKeySerializerSupplier(), producerFactory.getValueSerializerSupplier());
newFactory.setPhysicalCloseTimeout((int) producerFactory.getPhysicalCloseTimeout().getSeconds());
newFactory.setProducerPerConsumerPartition(producerFactory.isProducerPerConsumerPartition());
newFactory.setProducerPerThread(producerFactory.isProducerPerThread());
this.producerFactory = newFactory;
this.producerFactory = copyProducerFactoryWithOverrides(producerFactory, configOverrides);
}
else {
this.producerFactory = producerFactory;
@@ -189,6 +191,49 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationCo
this.transactional = this.producerFactory.transactionCapable();
}
private ProducerFactory<K, V> copyProducerFactoryWithOverrides(ProducerFactory<K, V> templateFactory,
Map<String, Object> configOverrides) {
ProducerFactory<K, V> newFactory;
try {
newFactory = templateFactory.copyWithConfigurationOverride(configOverrides);
}
catch (UnsupportedOperationException e) {
newFactory = handleNonCopyableProducerFactory(templateFactory, configOverrides);
}
return newFactory;
}
/**
* This method copies a ProducerFactory that misses the implementation of
* {@link org.springframework.kafka.core.ProducerFactory#copyWithConfigurationOverride(java.util.Map)}.
*
* @param templateFactory the ProducerFactory to copy from
* @param configOverrides new properties to be applied onto the templateFactory properties
* @return a DefaultKafkaProducerFactory configured with configOverrides and all
* public reachable settings of ProducerFactory
*/
private DefaultKafkaProducerFactory<K, V> handleNonCopyableProducerFactory(ProducerFactory<K, V> templateFactory,
Map<String, Object> configOverrides) {
Map<String, Object> producerProperties = new HashMap<>(templateFactory.getConfigurationProperties());
producerProperties.putAll(configOverrides);
DefaultKafkaProducerFactory<K, V> defaultFactory = new DefaultKafkaProducerFactory<>(producerProperties,
templateFactory.getKeySerializerSupplier(),
templateFactory.getValueSerializerSupplier());
defaultFactory.setPhysicalCloseTimeout((int) templateFactory.getPhysicalCloseTimeout().getSeconds());
defaultFactory.setProducerPerConsumerPartition(templateFactory.isProducerPerConsumerPartition());
defaultFactory.setProducerPerThread(templateFactory.isProducerPerThread());
for (ProducerPostProcessor<K, V> templatePostProcessor : templateFactory.getPostProcessors()) {
defaultFactory.addPostProcessor(templatePostProcessor);
}
for (ProducerFactory.Listener<K, V> templateListener : templateFactory.getListeners()) {
defaultFactory.addListener(templateListener);
}
return defaultFactory;
}
@Override
public void setBeanName(String name) {
this.beanName = name;

30
spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2021 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.
@@ -34,6 +34,7 @@ import org.springframework.lang.Nullable;
* @param <V> the value type.
*
* @author Gary Russell
* @author Thomas Strauß
*/
public interface ProducerFactory<K, V> {
@@ -141,7 +142,7 @@ public interface ProducerFactory<K, V> {
/**
* Return true when there is a producer per thread.
* @return the produver per thread.
* @return the producer per thread.
* @since 2.5
*/
default boolean isProducerPerThread() {
@@ -250,6 +251,26 @@ public interface ProducerFactory<K, V> {
default void removeConfig(String configKey) {
}
/**
* Copy the properties of the instance and the given properties to create a new producer factory.
* <p>The copy shall prioritize the override properties over the configured values.
* It is in the responsibility of the factory implementation to make sure the
* configuration of the new factory is identical, complete and correct.</p>
* <p>ProducerPostProcessor and Listeners must stay intact.</p>
* <p>If the factory does not implement this method, an exception will be thrown.</p>
* <p>Note: see
* {@link org.springframework.kafka.core.DefaultKafkaProducerFactory#copyWithConfigurationOverride}</p>
* @param overrideProperties the properties to be applied to the new factory
* @return {@link org.springframework.kafka.core.ProducerFactory} with properties
* applied
* @since 2.5.17
* @see org.springframework.kafka.core.KafkaTemplate#KafkaTemplate(ProducerFactory, java.util.Map)
*/
default ProducerFactory<K, V> copyWithConfigurationOverride(Map<String, Object> overrideProperties) {
throw new UnsupportedOperationException(
"This factory implementation doesn't support creating reconfigured copies.");
}
/**
* Called whenever a producer is added or removed.
*
@@ -271,9 +292,8 @@ public interface ProducerFactory<K, V> {
}
/**
* An exsting producer was removed.
* @param id the producer id (factory bean name and client.id separated by a
* period).
* An existing producer was removed.
* @param id the producer id (factory bean name and client.id separated by a period).
* @param producer the producer.
*/
default void producerRemoved(String id, Producer<K, V> producer) {

114
spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java
View File

@@ -31,6 +31,7 @@ import static org.springframework.kafka.test.assertj.KafkaConditions.value;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
@@ -87,7 +88,8 @@ import org.springframework.util.concurrent.SettableListenableFuture;
* @author Artem Bilan
* @author Igor Stepanov
* @author Biju Kunjummen
* @author Endika Guti?rrez
* @author Endika Gutierrez
* @author Thomas Strauß
*/
@EmbeddedKafka(topics = { KafkaTemplateTests.INT_KEY_TOPIC, KafkaTemplateTests.STRING_KEY_TOPIC })
public class KafkaTemplateTests {
@@ -100,11 +102,26 @@ public class KafkaTemplateTests {
private static Consumer<Integer, String> consumer;
private static final ProducerFactory.Listener<String, String> noopListener = new ProducerFactory.Listener<>() {
@Override
public void producerAdded(String id, Producer<String, String> producer) {
}
@Override
public void producerRemoved(String id, Producer<String, String> producer) {
}
};
private static final ProducerPostProcessor<String, String> noopProducerPostProcessor = processor -> processor;
@BeforeAll
public static void setUp() {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils
.consumerProps("KafkaTemplatetests" + UUID.randomUUID().toString(), "false", embeddedKafka);
.consumerProps("KafkaTemplatetests" + UUID.randomUUID(), "false", embeddedKafka);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
consumer = cf.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer, INT_KEY_TOPIC);
@@ -124,7 +141,7 @@ public class KafkaTemplateTests {
ProxyFactory prox = new ProxyFactory();
prox.setTarget(prod);
@SuppressWarnings("unchecked")
Producer<Integer, String> proxy = (Producer<Integer, String>) prox.getProxy();
Producer<Integer, String> proxy = (Producer<Integer, String>) prox.getProxy();
wrapped.set(proxy);
return proxy;
});
@@ -289,7 +306,7 @@ public class KafkaTemplateTests {
}
PL pl1 = new PL();
PL pl2 = new PL();
CompositeProducerListener<Integer, String> cpl = new CompositeProducerListener<>(new PL[] { pl1, pl2 });
CompositeProducerListener<Integer, String> cpl = new CompositeProducerListener<>(new PL[]{ pl1, pl2 });
template.setProducerListener(cpl);
template.sendDefault("foo");
template.flush();
@@ -340,7 +357,7 @@ public class KafkaTemplateTests {
template.flush();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicReference<SendResult<Integer, String>> theResult = new AtomicReference<>();
future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
future.addCallback(new ListenableFutureCallback<>() {
@Override
public void onSuccess(SendResult<Integer, String> result) {
@@ -374,7 +391,7 @@ public class KafkaTemplateTests {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicReference<SendResult<Integer, String>> theResult = new AtomicReference<>();
AtomicReference<String> value = new AtomicReference<>();
future.addCallback(new KafkaSendCallback<Integer, String>() {
future.addCallback(new KafkaSendCallback<>() {
@Override
public void onSuccess(SendResult<Integer, String> result) {
@@ -439,22 +456,103 @@ public class KafkaTemplateTests {
}
@Test
void testConfigOverrides() {
void testConfigOverridesWithDefaultKafkaProducerFactory() {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
pf.setPhysicalCloseTimeout(6);
pf.setProducerPerConsumerPartition(false);
pf.setProducerPerThread(true);
pf.addPostProcessor(noopProducerPostProcessor);
pf.addListener(noopListener);
Map<String, Object> overrides = new HashMap<>();
overrides.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
overrides.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "TX");
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf, true, overrides);
// modify the overrides map TXNid and clone it again
overrides.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "TX2");
KafkaTemplate<String, String> templateWTX2 = new KafkaTemplate<>(template.getProducerFactory(), true, overrides);
// clone the factory again with empty properties
KafkaTemplate<String, String> templateWTX2_2 = new KafkaTemplate<>(templateWTX2.getProducerFactory(), true,
Collections.singletonMap("dummy", "dont use"));
assertThat(template.getProducerFactory()).isOfAnyClassIn(DefaultKafkaProducerFactory.class);
assertThat(template.getProducerFactory().getConfigurationProperties()
.get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)).isEqualTo(StringSerializer.class);
assertThat(template.getProducerFactory().getPhysicalCloseTimeout()).isEqualTo(Duration.ofSeconds(6));
assertThat(template.getProducerFactory().isProducerPerConsumerPartition()).isFalse();
assertThat(template.getProducerFactory().isProducerPerThread()).isTrue();
assertThat(template.isTransactional()).isTrue();
assertThat(template.getProducerFactory().getListeners()).isEqualTo(pf.getListeners());
assertThat(template.getProducerFactory().getListeners().size()).isEqualTo(1);
assertThat(template.getProducerFactory().getPostProcessors()).isEqualTo(pf.getPostProcessors());
assertThat(template.getProducerFactory().getPostProcessors().size()).isEqualTo(1);
// then: initially we created without TX
assertThat(pf.getTransactionIdPrefix()).isBlank();
// and: we added TX to the first copy factory
assertThat(template.getProducerFactory().getTransactionIdPrefix()).isEqualTo("TX");
// and: we modified TX to TX2 in the second copy factory
assertThat(templateWTX2.getProducerFactory().getTransactionIdPrefix()).isEqualTo("TX2");
// and: we reuse the id from the template (TX2) in the third copy factory
assertThat(templateWTX2_2.getProducerFactory().getTransactionIdPrefix()).isEqualTo("TX2");
}
@Test
void testConfigOverridesWithCustomProducerFactory() {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<String, String> pf = new ProducerFactory<>() {
@Override
public Producer<String, String> createProducer() {
return null;
}
@Override
public List<Listener<String, String>> getListeners() {
return Collections.singletonList(noopListener);
}
@Override
public List<ProducerPostProcessor<String, String>> getPostProcessors() {
return Collections.singletonList(noopProducerPostProcessor);
}
@Override
public Map<String, Object> getConfigurationProperties() {
return Collections.singletonMap(ProducerConfig.ACKS_CONFIG, "all");
}
@Override
public Duration getPhysicalCloseTimeout() {
return Duration.ofSeconds(6);
}
@Override
public boolean isProducerPerConsumerPartition() {
return true;
}
@Override
public boolean isProducerPerThread() {
return false;
}
};
Map<String, Object> overrides = new HashMap<>();
overrides.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
overrides.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "TX");
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf, true, overrides);
assertThat(template.getProducerFactory()).isOfAnyClassIn(DefaultKafkaProducerFactory.class);
assertThat(template.getProducerFactory().getConfigurationProperties()
.get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)).isEqualTo(StringSerializer.class);
assertThat(template.getProducerFactory().getPhysicalCloseTimeout()).isEqualTo(Duration.ofSeconds(6));
assertThat(template.getProducerFactory().isProducerPerConsumerPartition()).isTrue();
assertThat(template.getProducerFactory().isProducerPerThread()).isFalse();
assertThat(template.isTransactional()).isTrue();
assertThat(template.getProducerFactory().getListeners()).isEqualTo(pf.getListeners());
assertThat(template.getProducerFactory().getListeners().size()).isEqualTo(1);
assertThat(template.getProducerFactory().getPostProcessors()).isEqualTo(pf.getPostProcessors());
assertThat(template.getProducerFactory().getPostProcessors().size()).isEqualTo(1);
assertThat(template.getProducerFactory().getTransactionIdPrefix()).isEqualTo("TX");
}
@Test
@@ -481,7 +579,7 @@ public class KafkaTemplateTests {
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
assertThatExceptionOfType(KafkaException.class).isThrownBy(() ->
template.send("missing.topic", "foo"))
template.send("missing.topic", "foo"))
.withCauseExactlyInstanceOf(TimeoutException.class);
pf.destroy();
}