Spring Kafka 和事务

Question

我想将 Spring Kafka 与事务一起使用，但我真的不明白它应该如何配置以及如何工作。

这是我的配置

    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
    props.put(ProducerConfig.RETRIES_CONFIG, String.valueOf(Integer.MAX_VALUE));
    props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);
    props.put(ProducerConfig.ACKS_CONFIG, "all");

此配置用于具有事务 ID 前缀的 DefaultKafkaProducerFactory:

defaultKafkaProducerFactory.setTransactionIdPrefix("my_app.");

问题 1:

我应该如何选择这个交易 ID 前缀？如果我理解正确的话，这个前缀被 spring 用来为每个创建的生产者生成一个事务 id。

为什么我们不能只使用“UUID.randomUUID()”？

问题 2:

如果生产者被销毁，它会生成一个新的交易ID。因此，如果应用程序崩溃，在重新启动时它将重用旧的事务 ID。

这正常吗？？？

问题 3:

我正在使用部署在云上的应用程序，它可以自动向上/向下扩展。这意味着我的前缀无法修复，因为每个实例上的所有生产者都会有冲突的事务 ID。

我应该在其中添加一个随机部分吗？当实例缩小/放大或崩溃并重新启动时，我是否需要恢复相同的前缀？

问题 4:

最后但同样重要的是，我们正在为我们的 Kafka 使用凭据。这似乎不起作用:

Current ACLs for resource `TransactionalId:my_app*`:
    User:CN... has Allow permission for operations: All from hosts: *

在知道我的交易 ID 已生成的情况下，我应该如何设置我的 ACL？

编辑 1

进一步阅读后，如果我理解正确。

如果您有一个从 P0(分区)读取的 C0(消费者)。如果代理开始消费者再平衡。P0 可以分配给另一个消费者 C1。这个消费者 C1 应该使用与之前的 C0 相同的交易 ID 以防止重复(僵尸围栏)？

你如何在 spring-kafka 中实现这一点？事务 ID 似乎与消费者无关，因此与读取的分区无关。

谢谢

Answer 1

1. 由于僵尸防护，您不能使用随机 TID - 如果服务器崩溃，您可能会在主题中有一个部分事务，该事务永远不会完成，并且不会从任何分区使用写入消耗更多内容对于该交易。

2. 出于上述原因，这是设计使然。

3. 同样，您不能随机化；出于上述原因。

例如，Cloud Foundry 有一个指示实例索引的环境变量。如果您使用的云平台不包含类似的东西，您将不得不以某种方式模拟它。然后，在事务 id 中使用它:

spring.kafka.producer.transaction-id-prefix=foo-${instance.index}-

1. ACL - 我无法回答这个问题；我对kafka权限不熟悉；可能最好为此问一个单独的问题。

2. 我认为我们需要向 Spring 添加一些逻辑以确保相同的事务 ID 始终用于特定的主题/分区。

https://github.com/spring-projects/spring-kafka/issues/800#issuecomment-419501929

编辑

自从这个答案 (KIP-447) 以来，情况发生了变化；如果您的经纪人是 2.5.0 或更高版本 - 请参阅。 https://docs.spring.io/spring-kafka/docs/2.5.5.RELEASE/reference/html/#exactly-once和 https://docs.spring.io/spring-kafka/docs/2.6.0-SNAPSHOT/reference/html/#exactly-once