Kafka 与 extractly-once

[Jimmy Wong]

Hi，all：
请教一下，我设置 checkpoint 的时间是 5 分钟，如果在这 5 分钟之内，某个 task 挂了，然后又重新拉起。我是不是可以理解为这时候从 checkpoint 的数据获得的是这 5 分钟之前的 Kafka offset，但是这 5 分钟之内的消息已经消费，流向下游。重新拉起之后，source 重放，那么这时候这 5 分钟的数据会再次被消费麽？如果再次消费，那么怎么保证 Extractly-Once 呢？
| |
Jimmy
|
|
[hidden email]
|
签名由网易邮箱大师定制

[jasine chen]

你好，应该是配合kafka 事务，在checkpoint的时候做事务提交，下游只读取commit的消息就能保证exactly-once，当然，会丧失一定的时效性

Jasine Chen
[hidden email]
Beijing
On Sep 9, 2019, 11:50 AM +0800, Jimmy Wong <[hidden email]>, wrote:

> Hi，all：
> 请教一下，我设置 checkpoint 的时间是 5 分钟，如果在这 5 分钟之内，某个 task 挂了，然后又重新拉起。我是不是可以理解为这时候从 checkpoint 的数据获得的是这 5 分钟之前的 Kafka offset，但是这 5 分钟之内的消息已经消费，流向下游。重新拉起之后，source 重放，那么这时候这 5 分钟的数据会再次被消费麽？如果再次消费，那么怎么保证 Extractly-Once 呢？
> | |
> Jimmy
> |
> |
> [hidden email]
> |
> 签名由网易邮箱大师定制
>

[chang chan]

消息队列本身很难保证消息不重复
exactly once 可以用  消息队列的 at least once + 后端幂等消费来实现
另外不建议使用 kafka 事务, 会拉低消息消费的速度

Jimmy Wong <[hidden email]> 于2019年9月9日周一 上午11:50写道：

> Hi，all：
> 请教一下，我设置 checkpoint 的时间是 5 分钟，如果在这 5 分钟之内，某个 task 挂了，然后又重新拉起。我是不是可以理解为这时候从
> checkpoint 的数据获得的是这 5 分钟之前的 Kafka offset，但是这 5
> 分钟之内的消息已经消费，流向下游。重新拉起之后，source 重放，那么这时候这 5 分钟的数据会再次被消费麽？如果再次消费，那么怎么保证
> Extractly-Once 呢？
> | |
> Jimmy
> |
> |
> [hidden email]
> |
> 签名由网易邮箱大师定制
>
>

[Jimmy Wong]

HI，能详细说下 “后端幂等消费” 的方案麽？








在 2019-09-09 14:37:55，"chang chan" <[hidden email]> 写道：

>消息队列本身很难保证消息不重复
>exactly once 可以用  消息队列的 at least once + 后端幂等消费来实现
>另外不建议使用 kafka 事务, 会拉低消息消费的速度
>
>Jimmy Wong <[hidden email]> 于2019年9月9日周一 上午11:50写道：
>
>> Hi，all：
>> 请教一下，我设置 checkpoint 的时间是 5 分钟，如果在这 5 分钟之内，某个 task 挂了，然后又重新拉起。我是不是可以理解为这时候从
>> checkpoint 的数据获得的是这 5 分钟之前的 Kafka offset，但是这 5
>> 分钟之内的消息已经消费，流向下游。重新拉起之后，source 重放，那么这时候这 5 分钟的数据会再次被消费麽？如果再次消费，那么怎么保证
>> Extractly-Once 呢？
>> | |
>> Jimmy
>> |
>> |
>> [hidden email]
>> |
>> 签名由网易邮箱大师定制
>>
>>

[Jimmy Wong]

Hi，我的理解是这样的：这个问题是由于 source 的重放，导致 Flink 内部就重复计算，并且会传到下游，最终反映在结果上，或者说这种操作不能保证内部 Extractly-Once？比如在 [8:00,8:05） 这 5 分钟之内，在 8:03 某个 task 挂了，然后又重新拉起。这时候从 checkpoint 的恢复获得的是这 8:00 分钟之前的 Kafka offset，但是 [8:00,8:03) 之内的消息已经消费，流向下游。重新拉起之后，由于某种原因 source 重放，那么这时候 [8:00,8:03) 的数据会再次被消费，并且会发往下游。






在 2019-09-09 16:01:48，"[hidden email]" <[hidden email]> 写道：

sink 的精确一次需要外部系统的支持的, 比如 kafka 的事务性producer, 社区有一篇文章讲的很好, 可以看一下 https://ververica.cn/developers/exactly-once/
 
[hidden email]
 
发件人： Jimmy Wong
发送时间： 2019-09-09 11:50
收件人： [hidden email]
主题： Kafka 与 extractly-once
Hi，all：
请教一下，我设置 checkpoint 的时间是 5 分钟，如果在这 5 分钟之内，某个 task 挂了，然后又重新拉起。我是不是可以理解为这时候从 checkpoint 的数据获得的是这 5 分钟之前的 Kafka offset，但是这 5 分钟之内的消息已经消费，流向下游。重新拉起之后，source 重放，那么这时候这 5 分钟的数据会再次被消费麽？如果再次消费，那么怎么保证 Extractly-Once 呢？
| |
Jimmy
|
|
[hidden email]
|
签名由网易邮箱大师定制

[Yan Gao]

Hi，
拉取过的消息，成功消费完更新offset完成消费。但可能存在的情况是，拉取过来的消息，没有成功消费下task挂掉了（offset没有提交），后面是会重再重新拉取进行消费的。Kafka的Extractly-Once是有作用范围的（
客户端从拉取消息，到消费完成，成功提交位移），要整个系统都实现精确一次语义，*flink拉取消息到成功提交位移这部分逻辑*
需要支持或实现幂等的，即能够处理*可能的多次拉取到成功提交位移前*这种情形。

Jimmy Wong <[hidden email]> 于2019年9月9日周一 下午5:44写道：

> Hi，我的理解是这样的：这个问题是由于 source 的重放，导致 Flink
> 内部就重复计算，并且会传到下游，最终反映在结果上，或者说这种操作不能保证内部 Extractly-Once？比如在 [8:00,8:05） 这 5
> 分钟之内，在 8:03 某个 task 挂了，然后又重新拉起。这时候从 checkpoint 的恢复获得的是这 8:00 分钟之前的 Kafka
> offset，但是 [8:00,8:03) 之内的消息已经消费，流向下游。重新拉起之后，由于某种原因 source 重放，那么这时候
> [8:00,8:03) 的数据会再次被消费，并且会发往下游。
>
>
>
>
>
>
> 在 2019-09-09 16:01:48，"[hidden email]" <[hidden email]> 写道：
>
> sink 的精确一次需要外部系统的支持的, 比如 kafka 的事务性producer, 社区有一篇文章讲的很好, 可以看一下
> https://ververica.cn/developers/exactly-once/
>
> [hidden email]
>
> 发件人： Jimmy Wong
> 发送时间： 2019-09-09 11:50
> 收件人： [hidden email]
> 主题： Kafka 与 extractly-once
> Hi，all：
> 请教一下，我设置 checkpoint 的时间是 5 分钟，如果在这 5 分钟之内，某个 task 挂了，然后又重新拉起。我是不是可以理解为这时候从
> checkpoint 的数据获得的是这 5 分钟之前的 Kafka offset，但是这 5
> 分钟之内的消息已经消费，流向下游。重新拉起之后，source 重放，那么这时候这 5 分钟的数据会再次被消费麽？如果再次消费，那么怎么保证
> Extractly-Once 呢？
> | |
> Jimmy
> |
> |
> [hidden email]
> |
> 签名由网易邮箱大师定制
>
>