为什么KafkaTemplate不关闭tansactional生产者？

Question

    <int-kafka:message-driven-channel-adapter
            listener-container="container"
            auto-startup="true"
            send-timeout="30000"
            channel="channelA"/>
   <bean id="container"  class="org.springframework.kafka.listener.KafkaMessageListenerContainer" parent="kafkaMessageListenerContainerAbstract">
        <constructor-arg>
            <bean class="org.springframework.kafka.listener.ContainerProperties">
                <constructor-arg
                        name="topics"
                        value="test"/>
                <property name="transactionManager" ref="KafkaTransactionManager"/>
            </bean>
        </constructor-arg>
    </bean>
.
.
.
    <int-kafka:outbound-channel-adapter kafka-template="kafkaTemplate"
                                        auto-startup="true"
                                        channel="channelB"
                                        topic="output"/>
      <bean id="dbsenderTemplate" class="org.springframework.kafka.core.KafkaTemplate">
        <constructor-arg>
            <bean class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
                <constructor-arg>
                    <map>
                        <entry key="value.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
                        <entry key="key.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
                        <entry key="bootstrap.servers" value="localhost:9092"/>
                    </map>
                </constructor-arg>
                <property name="transactionIdPrefix" value="mytest-"/>
                <property name="producerPerConsumerPartition" value="false"/>
            </bean>
        </constructor-arg>
    </bean>

启动应用程序的代码如下：

GenericXmlApplicationContext tempContext = new GenericXmlApplicationContext("beans.xml");
tempContext.close();

//POINT A.

try {
  Thread.sleep(60000);
} catch (InterruptedException e) {
   e.printStackTrace();
}

GenericXmlApplicationContext context = new GenericXmlApplicationContext();
context.load("beans.xml");
context.refresh();

//POINT B

在A点，我只是关闭了上下文来检查哪些beans已经关闭，然后Hibernate60秒，以便有时间检查JMX控制台。我注意到，即使上下文是关闭的，但是生产者仍然在JMX中注册。之后，我跟踪了代码，注意到在上下文关闭时，KafkaTemplate调用以下代码：

    public void flush() {
        Producer<K, V> producer = getTheProducer();
        try {
            producer.flush();
        }
        finally {
            closeProducer(producer, inTransaction());
        }
    }

    protected void closeProducer(Producer<K, V> producer, boolean inTx) {
        if (!inTx) {
            producer.close(this.closeTimeout);
        }
    }

这意味着它创建了一个生产者，但是因为它是事务性的，所以它不会被关闭。

if (!this.cache.contains(this)
        && !this.cache.offer(this)) {
        this.delegate.close(closeTimeout);
        }

这使得在关闭上下文时，DefaultKafkaProducerFactory的Destroy方法将清除缓存并物理地关闭生成器。但是在我的情况下，创建了应用程序上下文，但是在消费和生成任何消息之前，上下文是关闭的，只有Kafkatemplate的flush方法在内部调用，强制它创建事务生成器，但不将其放入缓存中。由于我没有启动生产者，而KafkaTemplate是在flush上启动的，所以DefaultKafkaProducerFactory在使用它们之前将它们放在缓存中不是很好吗？

Answer 1

如果此模板操作参与了在模板之外启动的事务，则无法关闭生成器。

即使在关闭时，它也只是“逻辑地”关闭--缓存以供另一个操作重用。

这是一种正确的行为吗？