apache-kafka - 使用幂等 Kafka Producer 时的排序保证

Question

我在我的应用程序中使用 Kafka 1.0.1，并且我已经开始使用 0.11 中引入的 Idempotent Producer 功能，并且在使用 Idempontent 功能时我无法理解排序保证。

我的制片人的配置是:
enable.idempotence = truemax.in.flight.requests.per.connection = 5retries = 50acks = all
根据文档:

重试

Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error. Note that this retry is no different than if the client resent the record upon receiving the error. Allowing retries without setting max.in.flight.requests.per.connection to 1 will potentially change the ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second succeeds, then the records in the second batch may appear first.

enable.idempotence

When set to 'true', the producer will ensure that exactly one copy of each message is written in the stream. If 'false', producer retries due to broker failures, etc., may write duplicates of the retried message in the stream. Note that enabling idempotence requires max.in.flight.requests.per.connection to be less than or equal to 5, retries to be greater than 0 and acks must be 'all'. If these values are not explicitly set by the user, suitable values will be chosen. If incompatible values are set, a ConfigException will be thrown.

我的配置似乎符合要求，但它们似乎没有对齐。

我的另一个问题与 OutOfOrderSequenceException 有关:
根据文档，如果我收到此异常，则意味着生产者有可能出现故障。但是如果我的生产者配置了 max.in.flight.requests.per.connection = 5假设第二个请求出现了乱序异常，那么以下所有已经在运行的请求会发生什么？这是否意味着我肯定有问题？

Answer 1

在 KafkaProducer 中启用幂等性时，顺序得到保证。
即使你有 max.in.flight.requests.per.connection大于或等于 1 的幂等 KafkaProducer 仍将确保 TopicPartition 内的排序。与 max.in.flight 相关的“重试”描述仅在禁用幂等性时才适用。
幂等的 KafkaProducer 使用内部递增的序列号来确保排序多达 5 个 in.flight 请求，如 Pull-Request #3743 中所述。 :

"[...] we retain the record metadata for 5 older batches."

此外， enable.idempotence 上的文档通知最多有 5 个飞行请求。否则你会得到一个 ConfigException .
更多细节在 KAFKA-5494以及 Design for max.in.flight > 1 with idempotence enabled. 上的相应文档步骤 5 和 6 将澄清您的问题:
有了上述假设，解决方案如下:

我们跟踪发送到分区的最后一个确认序列。这会在每次成功 ack 时更新，因此应该始终增加。

我们跟踪给定分区的批处理绑定(bind)的下一个序列。

当批处理耗尽时，我们分配了 nextSequence。我们还将 nextSequence 增加一个批处理的记录计数。

如果生产请求成功，我们将最后确认的序列设置为批处理的最后一个序列。

如果生产请求失败，成功的飞行批处理也将失败并出现 OutOfOrderSequenceException。

因此，如果批处理的序列号不是最后一个确认序列的后继，并且如果它失败并出现 OutOfOrderSequenceException，我们认为这是可重试的。

当一个批处理重新排队时，我们会在将其插入队列之前删除生产者 ID 和序列号。

当第一个飞行批处理失败(无论出于何种原因)时，我们将 nextSequence 重置为 lastAckdSequence + 1。

因此，如果一个批处理致命失败，则后续批处理的序列号在重试时将不同。这很好，因为之前的失败是 OutOfSequence异常，这绝对意味着请求被拒绝。

"[...] if my producer is configured with max.in.flight.requests.per.connection = 5 and let's say that the second request got the out of order exception, what happens to all of the following requests that are already in flight? will this mean that I am for sure out of order?"

如第 5 步和第 6 步所述，所有后续的进行中请求也将失败，并出现可重试 OutOfOrderSequenceExcpetion。 .如 retries大于 0幂等的 KafkaProducer 将能够保持订单。