扩展 Kafka Connect SinkTask 并从给定的偏移量开始消费

Question

我想扩展 SinkTask 以创建我自己的接收器连接器。

如果我在刷新期间保存了偏移量，并且下次启动接收器连接器时我想从我保存的偏移量中恢复读取，那么正确的方法是什么？

我尝试使用覆盖 initialize(SinkTaskContext context) 的 SinkTaskContext 来分配我自己的偏移量：

@Override
public void initialize(SinkTaskContext context) {
  HashMap<TopicPartition, Long> offsetMap = new HashMap<>();
  ...
  context.offset(offsetMap);
}

但这不起作用，因为分区尚未分配。我遇到了异常。

然后我是否应该将上下文（来自 initialize()）保存到一个全局变量中，然后在方法 open(Collection<TopicPartition> partitions) 中使用它为其分配偏移量（从 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 ?
}

Answer 1

在 open() 期间重置偏移量应该 是正确的方法，但由于 bug 仍未解决，目前无法正确处理。

目前的解决方法是处理 put() 中的重置偏移量。这可能有点违反直觉，但由于您正在管理自己的偏移量，因此您实际上可以根据需要忽略数据。当您收到第一个 put() 电话时，您可以处理加载偏移量并重置它们。所有后续数据都将来自您在重置时指定的偏移量。这就是 HDFS connector 当前实现其恰好一次交付的方式。（不幸的是，这是一个很好的例子，说明了如何获得恰好一次，但代码相对复杂。）事实上，由于 HDFS 连接器驱动了 Kafka Connect 中的偏移量管理功能，因此它不会进行重置关于重新平衡正是在实施中遗漏了这一点。

扩展 Kafka Connect SinkTask 并从给定的偏移量开始消费

Extend Kafka Connect SinkTask and start consuming from given offsets

apache-kafka

apache-kafka-connect