我在Flink应用程序中使用了flink-connector-kafka，语义设置为EXACTLY_ONCE，我看到下面的日志表明Kafka生产者已经关闭并重新连接：

Closing the Kafka producer with timeoutMillis = 0 ms.
Proceeding to force close the producer since pending requests could not be completed within timeout 0 ms.

查看源代码，我发现了来自生产者提交函数的close调用。提交函数调用finally块中的recycleTransactionalProducer，recycleTransactionalProducer函数调用close函数，后者打印日志。为什么Kafka生产者每次提交都被关闭？
软件包中的源代码：
org.apache.flink.streaming.connectors.kafka;
org.apache.kafka.clients.producer;

@Override
protected void commit(FlinkKafkaProducer.KafkaTransactionState transaction) {
    if (transaction.isTransactional()) {
        try {
            transaction.producer.commitTransaction();
        } finally {
            recycleTransactionalProducer(transaction.producer);
        }
    }
}

private void recycleTransactionalProducer(FlinkKafkaInternalProducer<byte[], byte[]> producer) {
    availableTransactionalIds.add(producer.getTransactionalId());
    producer.flush();
    producer.close(Duration.ofSeconds(0));
}

private void close(Duration timeout, boolean swallowException) {
    long timeoutMs = timeout.toMillis();
    if (timeoutMs < 0)
        throw new IllegalArgumentException("The timeout cannot be negative.");
    log.info("Closing the Kafka producer with timeoutMillis = {} ms.", timeoutMs);