Kafka Stream 中幂等性和恰好一次的区别 [英] Difference between idempotence and exactly-once in Kafka Stream
问题描述
我正在阅读我所理解的文档，我们可以通过启用 idempotence=true
I was going through document what I understood we can achieve exactly-once transaction with enabling idempotence=true
idempotence:幂等生产者只启用一次生产者针对单个主题.基本上每条消息发送有更强大的保证，不会被复制，以防万一错误
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 中还需要另一个属性恰好一次呢?幂等性和完全一次有什么区别
So if already we have idempotence then why we need another property exactly-once in Kafka Stream? What exactly different between idempotence vs exactly-once
为什么普通的 Kafka Producer 中没有恰好一次的属性?
Why exactly-once property not available in normal Kafka Producer?
推荐答案
在分布式环境中，故障是一种非常常见的情况，随时可能发生.在 Kafka 环境中，broker 可能会崩溃、网络故障、处理失败、消息发布失败或消息消费失败等.这些不同的场景引入了不同类型的数据丢失和重复.
In a distributed environment failure is a very common scenario that can be happened any time. In the Kafka environment, the broker can crash, network failure, failure in processing, failure while publishing message or failure to consume messages, etc. These different scenarios introduced different kinds of data loss and duplication.
失败场景
A(Ack Failed):生产者通过重试>1成功发布消息，但由于失败无法接收确认.在这种情况下，生产者将重试可能引入重复的相同消息.
A(Ack Failed): Producer published message successfully with retry>1 but could not receive acknowledge due to failure. In that case, the Producer will retry the same message that might introduce duplicate.
B(生产者进程在批量消息中失败): 生产者发送一批消息失败，很少发布成功.在这种情况下，一旦生产者重新启动，它将再次重新发布批次中的所有消息，这将在 Kafka 中引入重复.
B(Producer process failed in batch messages): Producer sending a batch of messages it failed with few published success. In that case and once the producer will restart it will again republish all messages from the batch which will introduce duplicate in Kafka.
C(触发并忘记失败) 生产者发布消息，retry=0(触发并忘记).万一发布失败，将不知道并发送下一条消息，这将导致消息丢失.
C(Fire & Forget Failed) Producer published message with retry=0(fire and forget). In case of failure published will not aware and send the next message this will cause the message lost.
D(Consumer failed in batch message) 消费者从 Kafka 收到一批消息并手动提交它们的偏移量 (enable.auto.commit=false).如果消费者在提交给 Kafka 之前失败了，下次消费者将再次消费相同的记录，从而在消费者端复制重复.
D(Consumer failed in batch message) A consumer receives a batch of messages from Kafka and manually commit their offset (enable.auto.commit=false). If consumers failed before committing to Kafka, next time Consumers will consume the same records again which reproduce duplicate on the consumer side.
Exactly-Once 语义
在这种情况下，即使生产者尝试重新发送消息，它也会导致消息只会被消费者发布和消费一次.
In this case, even if a producer tries to resend a message, it leads to the message will be published and consumed by consumers exactly once.
为了在 Kafka 中实现 Exactly-Once 语义，它使用了以下 3 个属性
To achieve Exactly-Once semantic in Kafka, it uses below 3 property
- enable.idempotence=true(地址 a、b 和 c)
- MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=5(生产者每个连接总是有一个进行中的请求)
- isolation.level=read_committed(地址 d)
启用幂等(enable.idempotence=true)
幂等交付使生产者能够准确地向Kafka写入消息在一个主题的生命周期内一次到一个主题的特定分区单个生产者没有数据丢失和每个分区的顺序.
Idempotent delivery enables the producer to write a message to Kafka exactly once to a particular partition of a topic during the lifetime of a single producer without data loss and order per partition.
"注意，启用幂等性要求 MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION 小于或等于 5，RETRIES_CONFIG 大于 0，ACKS_CONFIG 为全部".如果用户未明确设置这些值，则会选择合适的值.如果设置了不兼容的值，则会抛出 ConfigException
为了实现幂等性，Kafka 在生成消息时使用了一个唯一的 id，称为产品 id 或 PID 和序列号.生产者不断增加发布的每条消息的序列号，这些消息映射到唯一的 PID.代理总是将当前序列号与前一个序列号进行比较，如果新序列号不比前一个序列号大 +1，则它会拒绝，这避免了重复，并且如果消息中丢失了更多，则显示相同的时间
To achieve idempotence Kafka uses a unique id which is called product id or PID and sequence number while producing messages. The producer keeps incrementing the sequence number on each message published which map with unique PID. The broker always compare the current sequence number with the previous one and it rejects if the new one is not +1 greater than the previous one which avoids duplication and same time if more than greater show lost in messages
在失败场景中，broker 会将序列号与前一个序列号进行比较，如果序列号没有增加，+1 将拒绝该消息.
In a failure scenario broker will compare the sequence numbers with the previous one and if the sequence not increased +1 will reject the message.
事务(isolation.level)
事务使我们能够以原子方式更新多个主题分区中的数据.交易中包含的所有记录都将被成功保存，或者它们都不会被保存.它允许您在同一个事务中提交您的消费者偏移量以及您已处理的数据，从而允许端到端的恰好一次语义.
Transactions give us the ability to atomically update data in multiple topic partitions. All the records included in a transaction will be successfully saved, or none of them will be. It allows you to commit your consumer offsets in the same transaction along with the data you have processed, thereby allowing end-to-end exactly-once semantics.
生产者不会等待向 Kafka 写入消息，而生产者使用 beginTransaction、commitTransaction 和 abortTransaction(在失败的情况下)消费者使用isolation.level read_committed 或read_uncommitted
The producer doesn't wait to write a message to Kafka whereas the Producer uses beginTransaction, commitTransaction, and abortTransaction(in case of failure) Consumer uses isolation.level either read_committed or read_uncommitted
- read_committed:消费者将始终只读取提交的数据.
- read_uncommitted:按偏移顺序读取所有消息，无需等待用于要提交的事务
如果一个具有隔离级别=read_committed 的消费者到达一个尚未完成的事务的控制消息，它将不会再从该分区传递任何消息，直到生产者提交或中止事务或发生事务超时.事务超时由生产者使用配置transaction.timeout.ms(默认1分钟)决定.
If a consumer with isolation.level=read_committed reaches a control message for a transaction that has not completed, it will not deliver any more messages from this partition until the producer commits or aborts the transaction or a transaction timeout occurs. The transaction timeout is determined by the producer using the configuration transaction.timeout.ms(default 1 minute).
在 Producer & 中恰好一次消费者
在正常情况下，我们有单独的生产者和消费者.生产者必须幂等同时管理事务，因此消费者可以使用隔离级别来只读 read_committed 使整个过程成为原子操作.这保证了生产者将始终与源系统同步.即使生产者崩溃或事务中止，它也始终是一致的，并且将一条消息或一批消息作为一个单元发布一次.
In normal conditions where we have separate producers and consumers. The producer has to idempotent and same time manage transactions so consumers can use isolation.level to read-only read_committed to make the whole process as an atomic operation. This makes a guarantee that the producer will always sync with the source system. Even producer crash or a transaction aborted, it always is consistent and publishes a message or batch of the message as a unit once.
同一个消费者将接收一条消息或一批消息作为一个单元.
The same consumer will either receive a message or batch of the message as a unit once.
在 Exactly-Once 语义中，Producer 和 Consumer 将显示为将作为一个单元运行的原子操作.要么发布并完全消耗一次或中止.
In Exactly-Once semantic Producer along with Consumer will appear as atomic operation which will operate as one unit. Either publish and get consumed once at all or aborted.
在 Kafka Stream 中恰好一次
Kafka Stream 使用来自主题 A 的消息，处理消息并将其发布到主题 B，一旦发布，使用提交(提交主要是在秘密运行)将所有状态存储数据刷新到磁盘.
Kafka Stream consumes messages from topic A, process and publish a message to Topic B and once publish use commit(commit mostly run undercover) to flush all state store data to disk.
Kafka Stream 中的 Exactly-once 是一种读-处理-写模式，可保证此操作将被视为原子操作.由于Kafka Stream同时满足了生产者、消费者和事务的所有需求，因此Kafka Stream带有特殊的参数处理.保证可以精确_once或at_least_once，这使得不必单独处理所有参数变得容易.
Exactly-once in Kafka Stream is a read-process-write pattern that guarantees that this operation will be treated as an atomic operation. Since Kafka Stream caters producer, consumer and transaction all together Kafka Stream comes special parameter processing.guarantee which could exactly_once or at_least_once which make life easy not to handle all parameters separately.
Kafka Streams 自动更新消费者偏移量、本地状态存储、状态存储更改日志主题，并生产以输出所有主题一起.如果这些步骤中的任何一个失败，则所有更改都将回滚.
Kafka Streams atomically updates consumer offsets, local state stores, state store changelog topics, and production to output topics all together. If anyone of these steps fails, all of the changes are rolled back.
processing.guarantee:exactly_once 自动提供以下参数，您无需明确设置
- isolation.level=read_committed
- enable.idempotence=true
- MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=5
这篇关于Kafka Stream 中幂等性和恰好一次的区别的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
