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 f788a329..aa1ba715 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 @@ -108,10 +108,11 @@ import org.springframework.util.StringUtils; * @author Nakul Mishra * @author Artem Bilan * @author Chris Gilbert + * @author Thomas Strauß */ public class DefaultKafkaProducerFactory extends KafkaResourceFactory implements ProducerFactory, ApplicationContextAware, - BeanNameAware, ApplicationListener, DisposableBean { + BeanNameAware, ApplicationListener, DisposableBean { private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(DefaultKafkaProducerFactory.class)); @@ -360,6 +361,63 @@ public class DefaultKafkaProducerFactory extends KafkaResourceFactory this.maxAge = maxAge.toMillis(); } + /** + * Copy properties of the instance and the given properties to create a new producer factory. + *

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.

+ * @param overrideProperties the properties to be applied to the new factory + * @return {@link org.springframework.kafka.core.DefaultKafkaProducerFactory} with + * properties applied + */ + @Override + public ProducerFactory copyWithConfigurationOverride(Map overrideProperties) { + Map producerProperties = new HashMap<>(getConfigurationProperties()); + producerProperties.putAll(overrideProperties); + producerProperties = ensureExistingTransactionIdPrefixInProperties(producerProperties); + DefaultKafkaProducerFactory newFactory = + new DefaultKafkaProducerFactory<>(producerProperties, + getKeySerializerSupplier(), + getValueSerializerSupplier()); + newFactory.setPhysicalCloseTimeout((int) getPhysicalCloseTimeout().getSeconds()); + newFactory.setProducerPerConsumerPartition(isProducerPerConsumerPartition()); + newFactory.setProducerPerThread(isProducerPerThread()); + for (ProducerPostProcessor templatePostProcessor : getPostProcessors()) { + newFactory.addPostProcessor(templatePostProcessor); + } + for (ProducerFactory.Listener 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 ensureExistingTransactionIdPrefixInProperties(Map producerProperties) { + String transactionIdPrefix = getTransactionIdPrefix(); + if (StringUtils.hasText(transactionIdPrefix)) { + if (!producerProperties.containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG)) { + Map 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 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 extends KafkaResourceFactory BlockingQueue> 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 extends KafkaResourceFactory LOGGER.debug(() -> toString() + " abortTransaction()"); if (this.producerFailed != null) { LOGGER.debug(() -> "abortTransaction ignored - previous txFailed: " + this.producerFailed.getMessage() - + ": " + this); + + ": " + this); } else { try { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java index 10370a06..16a60bfe 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java @@ -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 implements KafkaOperations, ApplicationContextAware, BeanNameAware, ApplicationListener, DisposableBean { @@ -158,8 +159,16 @@ public class KafkaTemplate implements KafkaOperations, 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 implements KafkaOperations, ApplicationCo this.micrometerEnabled = KafkaUtils.MICROMETER_PRESENT; this.customProducerFactory = configOverrides != null && configOverrides.size() > 0; if (this.customProducerFactory) { - Map configs = new HashMap<>(producerFactory.getConfigurationProperties()); - configs.putAll(configOverrides); - DefaultKafkaProducerFactory 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 implements KafkaOperations, ApplicationCo this.transactional = this.producerFactory.transactionCapable(); } + private ProducerFactory copyProducerFactoryWithOverrides(ProducerFactory templateFactory, + Map configOverrides) { + + ProducerFactory 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 handleNonCopyableProducerFactory(ProducerFactory templateFactory, + Map configOverrides) { + + Map producerProperties = new HashMap<>(templateFactory.getConfigurationProperties()); + producerProperties.putAll(configOverrides); + DefaultKafkaProducerFactory defaultFactory = new DefaultKafkaProducerFactory<>(producerProperties, + templateFactory.getKeySerializerSupplier(), + templateFactory.getValueSerializerSupplier()); + defaultFactory.setPhysicalCloseTimeout((int) templateFactory.getPhysicalCloseTimeout().getSeconds()); + defaultFactory.setProducerPerConsumerPartition(templateFactory.isProducerPerConsumerPartition()); + defaultFactory.setProducerPerThread(templateFactory.isProducerPerThread()); + for (ProducerPostProcessor templatePostProcessor : templateFactory.getPostProcessors()) { + defaultFactory.addPostProcessor(templatePostProcessor); + } + for (ProducerFactory.Listener templateListener : templateFactory.getListeners()) { + defaultFactory.addListener(templateListener); + } + return defaultFactory; + } + @Override public void setBeanName(String name) { this.beanName = name; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.java index d57eac0a..95b90bb4 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.java @@ -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 the value type. * * @author Gary Russell + * @author Thomas Strauß */ public interface ProducerFactory { @@ -141,7 +142,7 @@ public interface ProducerFactory { /** * 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 { default void removeConfig(String configKey) { } + /** + * Copy the properties of the instance and the given properties to create a new producer factory. + *

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.

+ *

ProducerPostProcessor and Listeners must stay intact.

+ *

If the factory does not implement this method, an exception will be thrown.

+ *

Note: see + * {@link org.springframework.kafka.core.DefaultKafkaProducerFactory#copyWithConfigurationOverride}

+ * @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 copyWithConfigurationOverride(Map 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 { } /** - * 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 producer) { diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java index a3989ce1..c88fadba 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java @@ -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 consumer; + private static final ProducerFactory.Listener noopListener = new ProducerFactory.Listener<>() { + + @Override + public void producerAdded(String id, Producer producer) { + } + + @Override + public void producerRemoved(String id, Producer producer) { + } + + }; + + private static final ProducerPostProcessor noopProducerPostProcessor = processor -> processor; + + @BeforeAll public static void setUp() { embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils - .consumerProps("KafkaTemplatetests" + UUID.randomUUID().toString(), "false", embeddedKafka); + .consumerProps("KafkaTemplatetests" + UUID.randomUUID(), "false", embeddedKafka); DefaultKafkaConsumerFactory 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 proxy = (Producer) prox.getProxy(); + Producer proxy = (Producer) prox.getProxy(); wrapped.set(proxy); return proxy; }); @@ -289,7 +306,7 @@ public class KafkaTemplateTests { } PL pl1 = new PL(); PL pl2 = new PL(); - CompositeProducerListener cpl = new CompositeProducerListener<>(new PL[] { pl1, pl2 }); + CompositeProducerListener 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> theResult = new AtomicReference<>(); - future.addCallback(new ListenableFutureCallback>() { + future.addCallback(new ListenableFutureCallback<>() { @Override public void onSuccess(SendResult result) { @@ -374,7 +391,7 @@ public class KafkaTemplateTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicReference> theResult = new AtomicReference<>(); AtomicReference value = new AtomicReference<>(); - future.addCallback(new KafkaSendCallback() { + future.addCallback(new KafkaSendCallback<>() { @Override public void onSuccess(SendResult result) { @@ -439,22 +456,103 @@ public class KafkaTemplateTests { } @Test - void testConfigOverrides() { + void testConfigOverridesWithDefaultKafkaProducerFactory() { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); pf.setPhysicalCloseTimeout(6); pf.setProducerPerConsumerPartition(false); pf.setProducerPerThread(true); + pf.addPostProcessor(noopProducerPostProcessor); + pf.addListener(noopListener); Map overrides = new HashMap<>(); overrides.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); overrides.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "TX"); KafkaTemplate template = new KafkaTemplate<>(pf, true, overrides); + // modify the overrides map TXNid and clone it again + overrides.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "TX2"); + KafkaTemplate templateWTX2 = new KafkaTemplate<>(template.getProducerFactory(), true, overrides); + // clone the factory again with empty properties + KafkaTemplate 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 senderProps = KafkaTestUtils.producerProps(embeddedKafka); + ProducerFactory pf = new ProducerFactory<>() { + + @Override + public Producer createProducer() { + return null; + } + + @Override + public List> getListeners() { + return Collections.singletonList(noopListener); + } + + @Override + public List> getPostProcessors() { + return Collections.singletonList(noopProducerPostProcessor); + } + + @Override + public Map 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 overrides = new HashMap<>(); + overrides.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + overrides.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "TX"); + KafkaTemplate 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 template = new KafkaTemplate<>(pf, true); assertThatExceptionOfType(KafkaException.class).isThrownBy(() -> - template.send("missing.topic", "foo")) + template.send("missing.topic", "foo")) .withCauseExactlyInstanceOf(TimeoutException.class); pf.destroy(); }