1

我想扩展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 ?
}
4

1 回答 1

0

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

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

于 2016-07-23T20:56:59.767 回答