如何在Kafka接收器连接器中手动提交偏移量

Question

我有一个Kafka接收器任务，通过put（）方法收听Kafka主题
但我不想自动提交偏移量，因为一旦从Kafka取出记录，我就有一些处理逻辑
从Kafka获取记录后，如果处理成功，则只有我想提交偏移量，否则它应该再次从同一偏移量读取。

我可以在Kafka consumer中看到方法commitSync（），但在Sink Connector中找不到替代方法。

Answer 1

添加此属性：（"enable.auto.commit"，"false"）

使可能汽车commit的默认值为true，第二个属性为auto。犯罪间隔ms的默认值为5000

Answer 2

接收Kafka连接器提交

如果选项（enable.auto.commit）为False，则根据下面的选项（offset.flush.interval.ms）每60秒自动提交一次。如果put（）方法中没有错误，它将正常提交。

offset.flush.interval.ms
Interval at which to try committing offsets for tasks.

Type: long
Default: 60000
Importance: low

在Sink Kafka中管理偏移量

Kafka Connect应提交通过预提交传递给连接器的所有偏移量。但是，如果预提交返回一组空的偏移量，那么Kafka Connect将不会记录任何偏移量。在此处输入链接描述

SinkTask.java

/**
 * Pre-commit hook invoked prior to an offset commit.
 *
 * The default implementation simply invokes {@link #flush(Map)} and is thus able to assume all {@code currentOffsets} are committable.
 *
 * @param currentOffsets the current offset state as of the last call to {@link #put(Collection)}},
 *                       provided for convenience but could also be determined by tracking all offsets included in the {@link SinkRecord}s
 *                       passed to {@link #put}.
 *
 * @return an empty map if Connect-managed offset commits are not desired, otherwise a map of committable offsets by topic-partition.
 */
public Map<TopicPartition, OffsetAndMetadata> preCommit(Map<TopicPartition, OffsetAndMetadata> currentOffsets) {
    flush(currentOffsets);
    return currentOffsets;
}

或

SinkTaskContext.java

/**
 * Request an offset commit. Sink tasks can use this to minimize the potential for redelivery
 * by requesting an offset commit as soon as they flush data to the destination system.
 *
 * This is a hint to the runtime and no timing guarantee should be assumed.
 */
void requestCommit();