请教如何在flink job 里自定义kafka source function 及时提交offset

[jasonwu]

你好，

我有这样一个业务场景，在checkpoint disable的情况下，不想用auto.commit 功能在等待interval时间后提交offsets, 想要在flink job sink 完成后及时提交offset， 我阅读了一下flinkKafkaConsumer的源码，没有找到实现方式，麻烦请教大神有什么建议的实现方式吗？ 或者还有什么其他的建议？ 谢谢！  附件中是我写的测试代码。

53行我想new 一个kafkaConsumer出来，调用commitSync()方法但不管用。

KafkaConsumer<?, ?> kafkaConsumer = new KafkaConsumer(properties);
kafkaConsumer.commitSync();

[HideOnBushKi]

我猜测，要是可以在invoke的时候，拿到sink当前对应的barriar的话，从这个barriar取出offset，或者在richSinkFunction时，调用getRuntimeContext()上下文对象，从这个上下文中，拿到当前的task
metrics，从而取出task的offset



--
Sent from: http://apache-flink.147419.n8.nabble.com/