对于事务性和非事务性Kafka生产者，具有不同配置的相同KafkaTemplate

Question

通过 allowNon 事务属性，可以使用相同的 KafkaTemplate 来创建事务性和非事务性生产者。这两种生产者将使用来自Kafka生产者工厂的相同配置。

在某些特定情况下，可能需要为这两种生产者使用不同的配置。

下面的例子展示了一个具体的用例

• KafkaTemplate-非事务性生产者-

问题可以通过拥有2个工厂和/或2个模板来解决，但为了避免样板代码和复杂性，我们的想法是保持相同的模板。

目前有没有支持这种用例的方法？

如果没有，有个支持就太好了。通过tx前缀，可以覆盖事务性生产者的默认配置条目。

• KafkaTemplate-非事务性生产者-

如果您对这些问题有任何反馈，我们将不胜感激。

Answer 1

同意你的观点，对于大多数用例来说，这已经足够复杂了。

在我们的用例中，一些层构建在KafkaTemplate之上，该模板是隐藏的，从用户的角度来看可以认为是内部的。解决方案可能是在执行操作时检查我们是否在事务中。这是KafkaTemplate#inTransaction已经给出的内容

为了避免这种情况，可以在DefaultProducerFactory中进行轻微的内部返工，如下所示

// private -> protected
// new rawConfigs parameters
protected CloseSafeProducer<K, V> doCreateTxProducer(String prefix, String suffix,
        BiPredicate<CloseSafeProducer<K, V>, Duration> remover, Map<String, Object> rawConfigs) {
    // same as now
}

这样做将提供一种在需要时自定义配置的方法。

出于一致性原因，也可以对创建KafkaProducer（已受保护，只是一个新的参数）方法执行相同的操作。

如果你同意的话，我会非常乐意打开一个功能请求并为其删除一个PR。如果没有，我们可以考虑回答的问题

再次感谢您的快速回复。

Answer 2

已经有一个允许重写生成器工厂配置的构造函数。

    /**
     * Create an instance using the supplied producer factory and autoFlush setting.
     * <p>
     * Set autoFlush to {@code true} if you have configured the producer's
     * {@code linger.ms} to a non-default value and wish send operations on this template
     * 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 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.
     * @since 2.5
     * @see Producer#flush()
     */
    public KafkaTemplate(ProducerFactory<K, V> producerFactory, boolean autoFlush,
            @Nullable Map<String, Object> configOverrides) {