Kafka connect的offset commit机制

1. 说明

版本说明：

Confluent ： 2.0.0

Kafka ： 0.9.0.0

准确的说，Kafka connector 是一个双向的通道：

kafka->target ，对应任务称为SinkTask;

source->kafka， 对应任务称为SourceTask；

我比较关注的是kafka->hdfs，本文重点是SinkTask的offset commit机制。

2. 迷惑性的配置

在connect的配置文件中有这么一项配置：“offset.storage.topic”。首先声明，在confluent document里面对这个字段的解释就是扯淡的。

offset.storage.topic - the Kafka topic to store connector offset state in. Although this topic can be auto-created if your cluster has auto topic creation enabled, it is highly recommended that you create it before starting the Kafka Connect cluster. To support large Kafka Connect clusters, this topic should have a large number of partitions (e.g. 25 or 50, just like Kafka’s built-in __offsets topic) and highly replicated (3x or more).

看了这段说明，你立马会觉得这就是kafka connect用来存放offset的。BUT，它和SinkTask没多大关系，和hdfsSinkTask没有半毛钱关系。翻阅史书典籍（代码注释），终于得到了详细的说明。

出处：kafka-2.0.0\connect\runtime\src\main\java\org\apache\kafka\connect\storage\OffsetStorageWriter.java

* <p>
* Connect uses an OffsetStorage implementation to save state about the current progress of
* source (import to Kafka) jobs, which may have many input partitions and "offsets" may not be as
* simple as they are for Kafka partitions or files. Offset storage is not required for sink jobs
* because they can use Kafka's native offset storage (or the sink data store can handle offset
* storage to achieve exactly once semantics).
* </p>

大概是这么个意思：

1 . 主要是给SourceTask用的，用来记录Kafka connector对源数据的消费状态，这里的源就不是指kafka了，可能是阿猫阿狗什么的；

2 . SinkTask用不到，因为可以直接用kafka的offset管理机制记录消费状态；

只跑SinkTask时，通过kafka可以看到这个topic里面是没有数据的。

xxx$ ./bin/kafka-run-class kafka.tools.GetOffsetShell --topic connect-offsets --broker-list localhost:9092 --time -1
connect-offsets:0:0

3. kafka connector的offset commit机制

首先将这两个概念剥开：kafka connect和kafka hdfs connect。是confluent下两个项目。kafka connect书写了整个处理框架， hdfs connect专注于目标为hdfs情况下的处理。

3.1 kafka connect 处理过程

SinkTask的处理逻辑集中在WorkerSinkTaskThread.java中。重点关注commitOffsets方法。基本过程是：

调用task的flush机制记录offset。这个task根据目标文件系统不同，可能有种多样，比如hdfsSinkTask、FileStreamSinkTask等；

异步向kafka Server commit offset，注意commit是基于partition的；

3.2 kafka hdfs connect 处理过程

准确的说kafka hdfs connect在offset commit这事上什么都没有做。

@Override
public void flush(Map<TopicPartition, OffsetAndMetadata> offsets) {
// Do nothing as the connector manages the offset
}

也就是说在offset这事上另有高招，什么高招？另起一篇再说咯。