go - Kafka : Sarama, 幂等性和 transactional.id

Question

是否Shopify/sarama提供类似于 transactional.id 的选项在 JVM API 中？
该库支持幂等( Config.Producer.Idemponent ，类似于 enable.idempotence )，但我不明白如何在没有 transactional.id 的情况下使用它.
如果我错了，请纠正我，Sarama 中缺少有关这些选项的文档。但是根据 JVM 文档，没有标识符的幂等性将受到单个生产者 session 的限制。换句话说，当生产者失败并重新启动时，我们将失去保证。
我在源代码和一些测试( for example )中找到了相关属性，但不明白如何在外部使用它们。

Answer 1

Shopify/sarama为 Kafka Exactly Once (Idempotency) 提供启用幂等的生产者。但是对于下面的配置设置需要在那里。
来自 Shopify/sarama/config.go

    if c.Producer.Idempotent {
        if !c.Version.IsAtLeast(V0_11_0_0) {
            return ConfigurationError("Idempotent producer requires Version >= V0_11_0_0")
        }
        if c.Producer.Retry.Max == 0 {
            return ConfigurationError("Idempotent producer requires Producer.Retry.Max >= 1")
        }
        if c.Producer.RequiredAcks != WaitForAll {
            return ConfigurationError("Idempotent producer requires Producer.RequiredAcks to be WaitForAll")
        }
        if c.Net.MaxOpenRequests > 1 {
            return ConfigurationError("Idempotent producer requires Net.MaxOpenRequests to be 1")
        }
    }

在 Shopify/sarama他们是如何做到这一点的，有一个 producerEpoch身份证号 AsyncProducer的 transactionManager .您可以引用 Shopify/sarama/async_producer.go 中的文件.此 Id 使用生产者初始化进行初始化，并在成功生成每条消息时递增。阅读 bumpEpoch()函数在 async_producer.go 中看到文件。
这是该生产者与代理 session 的序列 ID，它与每条消息一起发送。消息发布成功时递增。
阅读 this example .它描述了幂等性的工作原理。
您对生产者 session 事实是正确的。这正是曾经为单个生产者 session 所 promise 的。在序列失败后立即重新启动生产者时，可能会有重复。

When producer restarts, new PID gets assigned. So the idempotency is promised only for a single producer session. Even though producer retries requests on failures, each message is persisted in the log exactly once. There can still be duplicates depending on the source where the producer is getting data. Kafka won’t take care of the duplicate data received by the producer. So, in some cases, you may require an additional de-duplication system.