apache-kafka - Kafka Stream 中幂等性和恰好一次的区别

Question

我正在阅读我所理解的文档，我们可以通过启用 idempotence=true 实现一次交易

idempotence: The Idempotent producer enables exactly once for a producer against a single topic. Basically each single message send has stonger guarantees and will not be duplicated in case there's an error

因此，如果我们已经拥有幂等性，那么为什么我们需要在 Kafka Stream 中恰好一次的另一个属性？幂等性与完全一次之间有什么区别

为什么在普通的 Kafka Producer 中没有一次性属性？

Answer 1

在分布式环境中，故障是一种非常常见的情况，随时可能发生。在 Kafka 环境中，broker 可能会崩溃、网络故障、处理失败、消息发布失败或消息消费失败等。
这些不同的场景引入了不同类型的数据丢失和重复。
故障场景
A(确认失败): 生产者成功发布消息，重试> 1，但由于失败而无法接收确认。在这种情况下，生产者将重试可能引入重复的相同消息。

B(生产者进程在批量消息中失败): 生产者发送一批消息失败，很少发布成功。在这种情况下，一旦生产者重新启动，它将再次重新发布批次中的所有消息，这将在 Kafka 中引入重复。

C(Fire & Forget Failed) 生产者发布消息，retry=0(fire and忘记)。万一发布失败，将不知道并发送下一条消息，这将导致消息丢失。

D(Consumer failed in batch message) 消费者从Kafka收到一批消息并手动提交他们的偏移量(enable.auto.commit=false)。如果消费者在提交给 Kafka 之前失败了，下一次消费者将再次消费相同的记录，从而在消费者端复制重复。

恰好一次语义

In this case, even if a producer tries to resend a message, it leadsto the message will be published and consumed by consumers exactly once.

为了在 Kafka 中实现 Exactly-Once 语义，它使用以下 3 个属性

enable.idempotence=true (address a, b & c)

MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=5(生产者每个连接总是有一个正在进行的请求)

isolation.level=read_committed(地址d)

启用幂等(enable.idempotence=true)

Idempotent delivery enables the producer to write a message to Kafka exactlyonce to a particular partition of a topic during the lifetime of asingle producer without data loss and order per partition.

“请注意，启用幂等性要求 MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION 小于或等于 5，RETRIES_CONFIG 大于 0 且 ACKS_CONFIG 为 'all'。如果用户未明确设置这些值，则将选择合适的值。如果不兼容的值设置，一个 ConfigException 将被抛出”
为了实现幂等性，Kafka 在生成消息时使用了一个唯一的 id，称为产品 id 或 PID 和序列号。生产者不断增加发布的每条消息的序列号，这些消息映射到唯一的 PID。代理总是将当前序列号与前一个序列号进行比较，如果新序列号不比前一个序列号大 +1，则它会拒绝，这避免了重复，并且如果消息中丢失了更多，则显示相同的时间

在失败场景中，broker 会将序列号与前一个序列号进行比较，如果序列号没有增加，+1 将拒绝该消息。

事务(isolation.level)
事务使我们能够以原子方式更新多个主题分区中的数据。交易中包含的所有记录都将被成功保存，或者它们都不会被保存。它允许您在同一个事务中提交您的消费者偏移量以及您已处理的数据，从而允许端到端的恰好一次语义。
生产者不会等待向 Kafka 写入消息，而生产者使用 beginTransaction、commitTransaction 和 abortTransaction(在失败的情况下)
消费者使用隔离级别 read_committed 或 read_uncommitted

read_committed:消费者将始终只读取提交的数据。

read_uncommitted:按偏移顺序读取所有消息，无需等待
对于要提交的事务

如果具有隔离级别=read_committed 的消费者到达尚未完成的事务的控制消息，则在生产者提交或中止事务或发生事务超时之前，它将不会从该分区传递更多消息。事务超时由生产者使用配置transaction.timeout.ms(默认1分钟)确定。
在生产者和消费者中恰好一次
在正常情况下，我们有单独的生产者和消费者。生产者必须幂等同时管理事务，因此消费者可以使用隔离级别来只读 read_committed 使整个过程成为原子操作。
这保证了生产者将始终与源系统同步。即使生产者崩溃或事务中止，它也始终是一致的，并且将一条消息或一批消息作为一个单元发布一次。
同一个消费者将接收一条消息或一批消息作为一个单元一次。

In Exactly-Once semantic Producer along with Consumer will appear asatomic operation which will operate as one unit. Either publish andget consumed once at all or aborted.

在 Kafka Stream 中恰好一次
Kafka Stream 使用来自主题 A 的消息，处理消息并将其发布到主题 B，一旦发布，使用提交(提交主要是在 secret 运行)将所有状态存储数据刷新到磁盘。
Kafka Stream 中的 Exactly-once 是一种读-处理-写模式，可保证此操作将被视为原子操作。由于Kafka Stream同时满足生产者、消费者和事务，因此Kafka Stream带有特殊的参数processing.guarantee，它可以精确_once或at_least_once，这使得不必单独处理所有参数变得容易。

Kafka Streams atomically updates consumer offsets, local state stores,state store changelog topics, and production to output topics alltogether. If anyone of these steps fails, all of the changes arerolled back.

processing.guarantee:exactly_once 自动提供以下参数，您无需明确设置

isolation.level=read_committed

enable.idempotence=true

MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=5