我想扩展SinkTask
以创建自己的接收器连接器。
如果我在刷新期间保存偏移量,并且下次启动接收器连接器时我想从保存的偏移量中恢复读取,那么正确的方法是什么?
我尝试使用SinkTaskContext
被覆盖initialize(SinkTaskContext context)
的来分配我自己的偏移量:
@Override
public void initialize(SinkTaskContext context) {
HashMap<TopicPartition, Long> offsetMap = new HashMap<>();
...
context.offset(offsetMap);
}
但这不起作用,因为尚未分配分区。我遇到了一个例外。
然后我是否应该将上下文 (from initialize()
) 保存到全局变量中,然后使用它在方法内部分配偏移量open(Collection<TopicPartition> partitions)
(overridden from SinkTask
),就像我在里面做的那样initialize
?例如:
@Override
public void open(Collection<TopicPartition> partitions) {
HashMap<TopicPartition, Long> offsetMapNew = new HashMap<>();
for (TopicPartition tp : partitions) // for each partition assigned
{
Long offset = myOffsetMap.get(tp.topic() + "-" + tp.partition());
if (offset == null) { offset = 0l; } // 0 Long
offsetMapNew.put(tp, offset);
}
mySavedTaskContext.offset(offsetMapNew); // sync offsets ?
}