From c4787fc8927921cfd637cc86dcd5dc8af2d31204 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Thomas=20Strau=C3=9F?= <49093997+thst71@users.noreply.github.com> Date: Thu, 21 Oct 2021 08:10:19 -0400 Subject: [PATCH] GH-1983: Support copy of producer factory 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 --- .../core/DefaultKafkaProducerFactory.java | 68 ++++++++++- .../kafka/core/KafkaTemplate.java | 67 ++++++++-- .../kafka/core/ProducerFactory.java | 30 ++++- .../kafka/core/KafkaTemplateTests.java | 114 ++++++++++++++++-- 4 files changed, 250 insertions(+), 29 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 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(); }