KafkaConnect提供了一个接口OffsetBackingStore用于存储每个Connector当前订阅源实例的位点信息offset（例如MySql的binlog），offset在KafkaConnect中是一个key-value的值，具体由每个Connector自己去定义，OffsetBackingStore只会支持读写字节类型操作（实际上是ByteBuffer）,对于ByteBuffer的序列化和反序列需要Connector自己实现。
public interface OffsetBackingStore {
/* 启动BackingStore */
void start();
/* 停止BackingStore */
void stop();
/* 根据指定Key获取对应的value */
Future<Map<ByteBuffer, ByteBuffer>> get(Collection<ByteBuffer> keys);
/* 存储key和对应的value */
Future<Void> set(Map<ByteBuffer, ByteBuffer> values, Callback<Void> callback);
/* key-value的配置类，这里的Worker是Connect的配置文件中的配置 */
void configure(WorkerConfig config);
}从上面的接口参数可以看出，OffsetBackingStore需要支持批量读写key-value操作，是考虑到减少网络请求次数。由于OffsetBackingStore是一个共享资源，可能由许多与单个任务相关联的OffsetStorage实例使用，因此调用者必须确保键包含关于连接器的信息，以便共享名称空间不会导致键冲突。
OffsetBackingStore默认有基于内存的MemoryOffsetBackingStore，用于分布式基于Kafka的KafkaOffsetBackingStore的实现，以及基于文件的FileOffsetBackingStore，本次会介绍比较常用的基于Kafka的实现。
注意不同的connect集群，不能使用相同的config、status和offset三个topic，但是可以复用同一个Kafka。
1、KafkaOffsetBackingStore
KafkaOffsetBackingStore是利用kafka上的topic存储connector中的offset信息，以便下次connector重新起来时可以读取并继续订阅源集群数据，使用时需要事先调用configure方法初始化配置
// WorkerConfig表示配置文件传递过来的配置
@Override
public void configure(final WorkerConfig config) {
// 获取offset.topic
String topic = config.getString(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG);
if (topic == null || topic.trim().length() == 0)
throw new ConfigException("Offset storage topic must be specified");
// 获取Kafka的集群ID,用于后续的监控
String clusterId = ConnectUtils.lookupKafkaClusterId(config);
data = new HashMap<>();
Map<String, Object> originals = config.originals();
// 设置producer的配置,主要包括序列化和反序列key和value,以及发送超时时间
Map<String, Object> producerProps = new HashMap<>(originals);
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE);
ConnectUtils.addMetricsContextProperties(producerProps, config, clusterId);
Map<String, Object> consumerProps = new HashMap<>(originals);
// 设置consumer的配置,需要保证和producer的序列化和反序列一致
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
ConnectUtils.addMetricsContextProperties(consumerProps, config, clusterId);
Map<String, Object> adminProps = new HashMap<>(originals);
ConnectUtils.addMetricsContextProperties(adminProps, config, clusterId);
Supplier<TopicAdmin> adminSupplier;
if (topicAdminSupplier != null) {
adminSupplier = topicAdminSupplier;
} else {
// Create our own topic admin supplier that we'll close when we're stopped
ownTopicAdmin = new SharedTopicAdmin(adminProps);
adminSupplier = ownTopicAdmin;
}
// 这里是取出offset.topic的其他配置
Map<String, Object> topicSettings = config instanceof DistributedConfig
? ((DistributedConfig) config).offsetStorageTopicSettings()
: Collections.emptyMap();
// 这里是用于创建Kafka Connect Offset.topic的配置
// 该Topic要么人工手动创建,否则由connect帮你创建
// 关键是需要配置topic的cleanup.policy=compact
NewTopic topicDescription = TopicAdmin.defineTopic(topic)
.config(topicSettings) // first so that we override user-supplied settings as needed
.compacted()
.partitions(config.getInt(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG))
.replicationFactor(config.getShort(DistributedConfig.OFFSET_STORAGE_REPLICATION_FACTOR_CONFIG))
.build();
// 初始化offsetLog
offsetLog = createKafkaBasedLog(topic, producerProps, consumerProps, consumedCallback, topicDescription, adminSupplier);
}代码中表示connect会自动帮我们创建offset.topic，用于存储connector中的offset。其中当connect帮我们创建时，offset.topic要求cleanup.policy=compact，该参数的如图所示：
日志的compact操作会根据Key将消息聚合，永久保留最后一次出现时的数据。这样无论什么时候消费消息，都能拿到每个Key的最新版本的数据。这样就可以把connector发送的offset永久存储到topic中，并自动进行聚合操作。聚合操作不是实时发生的，所以这里OffsetBackingStore需要判断如果Key重复的话，自动拿偏移量最大的，或者说最新的数据。
回来看下KafkaOffsetBackingStore的start、stop、get和set三个方法，这三个方法逻辑比较简单，主要部分都交给KafkaBaseLog实现
@Override
public void start() {
log.info("Starting KafkaOffsetBackingStore");
offsetLog.start();
log.info("Finished reading offsets topic and starting KafkaOffsetBackingStore");
}
@Override
public void stop() {
log.info("Stopping KafkaOffsetBackingStore");
try {
offsetLog.stop();
} finally {
if (ownTopicAdmin != null) {
ownTopicAdmin.close();
}
}
log.info("Stopped KafkaOffsetBackingStore");
}
private HashMap<ByteBuffer, ByteBuffer> data;
@Override
public Future<Map<ByteBuffer, ByteBuffer>> get(final Collection<ByteBuffer> keys) {
// 这里是创建KafkaBaseLog返回时的回调方法
ConvertingFutureCallback<Void, Map<ByteBuffer, ByteBuffer>> future = new ConvertingFutureCallback<Void, Map<ByteBuffer, ByteBuffer>>() {
@Override
public Map<ByteBuffer, ByteBuffer> convert(Void result) {
Map<ByteBuffer, ByteBuffer> values = new HashMap<>();
for (ByteBuffer key : keys)
values.put(key, data.get(key));
return values;
}
};
// 这里注释提到，KafkaOffsetBackingStore本身一旦connector比较多的情况下
// offset.topic可能会存在比较多的offset消息,compact的清理策略本身不会实
// 时发生,所以这里的从头读到尾时一个比较耗时的操作,只应该调用在connect
// 刚开始启动的时候
offsetLog.readToEnd(future);
return future;
}
@Override
public Future<Void> set(final Map<ByteBuffer, ByteBuffer> values, final Callback<Void> callback) {
SetCallbackFuture producerCallback = new SetCallbackFuture(values.size(), callback);
for (Map.Entry<ByteBuffer, ByteBuffer> entry : values.entrySet()) {
ByteBuffer key = entry.getKey();
ByteBuffer value = entry.getValue();
offsetLog.send(key == null ? null : key.array(), value == null ? null : value.array(), producerCallback);
}
return producerCallback;
}KafkaOffsetBackingStore核心使用了KafkaBaseLog该类，下面来看下该类相关操作。
2、KafkaBaseLog
KafkaBaseLog提供了一个通用的基于Kafka的分布式日志存储。该存储支持共享和压缩。这个类在后台线程中运行一个消费者来持续跟踪目标主题，同时提供一个生产者用于发送offset、config等不同数据类型的消息到Kafka中。（除了上面提到的offset存储实现KafkaOffsetBackingStore，connect内部还有同样基于KafkaBaseLog的status实现KafkaStatusBackingStore，以及config的实现KafkaConfigBackingStore）
先来看KafkaBaseLog.start方法。
public void start() {
log.info("Starting KafkaBasedLog with topic " + topic);
// Create the topic admin client and initialize the topic ...
// TopicAdmin,用于创建指定的topic
admin = topicAdminSupplier.get(); // may be null
initializer.accept(admin);
// 创建KafkaBaeLog的producer和consumer
// 关于producer,需要保证以下两点:
// 1、默认下需要保证所有replication都ack后才返回,所以ACKS_CONFIG=all;
// 2、为了避免消息发送乱序（网络问题）,MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1;
// 关于消费者创建,需要保证如下:
// 1、需要保证每次启动时从头开始消费,所以AUTO_OFFSET_RESET_CONFIG=earliest;
// 2、禁止自动提交commit,ENABLE_AUTO_COMMIT_CONFIG=false
producer = createProducer();
consumer = createConsumer();
List<TopicPartition> partitions = new ArrayList<>();
// 获取当前KafkaBaseLog指定的topic下的所有分区
List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
long started = time.nanoseconds();
long sleepMs = 100;
while (partitionInfos == null && time.nanoseconds() - started < CREATE_TOPIC_TIMEOUT_NS) {
time.sleep(sleepMs);
sleepMs = Math.min(2 * sleepMs, MAX_SLEEP_MS);
partitionInfos = consumer.partitionsFor(topic);
}
if (partitionInfos == null)
throw new ConnectException("Could not look up partition metadata for offset backing store topic in" +
" allotted period. This could indicate a connectivity issue, unavailable topic partitions, or if" +
" this is your first use of the topic it may have taken too long to create.");
for (PartitionInfo partition : partitionInfos)
partitions.add(new TopicPartition(partition.topic(), partition.partition()));
partitionCount = partitions.size();
// 将所有的分区都交给consumer去消费,这样如果每个connect都是采用kafka分布式存储的话
// 那么不同的connect都有各自的KafkaBaseLog,相当于每个connect都会拿到同一份topic上的消息
consumer.assign(partitions);
// 消费者指定offset为topic的开始
consumer.seekToBeginning(partitions);
// 调用方法从头读到尾
readToLogEnd(true);
// 启动一个线程去实时消费topic中的数据
thread = new WorkThread();
thread.start();
log.info("Finished reading KafkaBasedLog for topic " + topic);
log.info("Started KafkaBasedLog for topic " + topic);
}readToLogEnd可以说时KafkaBaseLog最关键的方法，用于从队列中获取全部的消息。
private void readToLogEnd(boolean shouldRetry) {
// 获取consumer中所有的分区
Set<TopicPartition> assignment = consumer.assignment();
// 获取consumer分配的分区当前最大的offset
Map<TopicPartition, Long> endOffsets = readEndOffsets(assignment, shouldRetry);
log.trace("Reading to end of log offsets {}", endOffsets);
while (!endOffsets.isEmpty()) {
Iterator<Map.Entry<TopicPartition, Long>> it = endOffsets.entrySet().iterator();
while (it.hasNext()) {
Map.Entry<TopicPartition, Long> entry = it.next();
TopicPartition topicPartition = entry.getKey();
long endOffset = entry.getValue();
// 获取consumer关于当前分区的最大消费位点
long lastConsumedOffset = consumer.position(topicPartition);
// 如果消费者关于当前分区的offset等于或大于（两个offset获取有时差，可能有新数据导致）分区本身的最大offset
// 则说明已经消费者已经读取了topic上的全部数据，无需重复读取
if (lastConsumedOffset >= endOffset) {
log.trace("Read to end offset {} for {}", endOffset, topicPartition);
it.remove();
} else {
// 需要一直读取topic上的数据
log.trace("Behind end offset {} for {}; last-read offset is {}",
endOffset, topicPartition, lastConsumedOffset);
poll(Integer.MAX_VALUE);
break;
}
}
}
}
// poll方法由消费者持续读取topic上的数据,并调用consumerCallback回调
private void poll(long timeoutMs) {
try {
ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(timeoutMs));
for (ConsumerRecord<K, V> record : records)
consumedCallback.onCompletion(null, record);
} catch (WakeupException e) {
// Expected on get() or stop(). The calling code should handle this
throw e;
} catch (KafkaException e) {
log.error("Error polling: " + e);
}
}至于这里对于消费消息的callback，则可以由外部去定义，例如在KafkaOffsetBackingStore中，则是每消费数据则修改内部的data对象。
private final Callback<ConsumerRecord<byte[], byte[]>> consumedCallback = new Callback<ConsumerRecord<byte[], byte[]>>() {
@Override
public void onCompletion(Throwable error, ConsumerRecord<byte[], byte[]> record) {
ByteBuffer key = record.key() != null ? ByteBuffer.wrap(record.key()) : null;
ByteBuffer value = record.value() != null ? ByteBuffer.wrap(record.value()) : null;
// 每消费一条数据则写入内部的Map中,因此消息中同样的key只会保留最新的一份
data.put(key, value);
}
};KafkaBaseLog除了会在启动的时候去消费topic中的数据，还会起一个线程去实时消费，达到类似追踪的效果
private class WorkThread extends Thread {
public WorkThread() {
super("KafkaBasedLog Work Thread - " + topic);
}
@Override
public void run() {
try {
while (true) {
int numCallbacks;
synchronized (KafkaBasedLog.this) {
if (stopRequested)
break;
numCallbacks = readLogEndOffsetCallbacks.size();
}
// 如果callback为空,就是没有回调，也不用再轮询了。
if (numCallbacks > 0) {
try {
// 读取topic中的数据
readToLogEnd(false);
} catch (TimeoutException e) {
continue;
} catch (RetriableException | org.apache.kafka.connect.errors.RetriableException e) {
continue;
} catch (WakeupException e) {
continue;
}
}
synchronized (KafkaBasedLog.this) {
for (int i = 0; i < numCallbacks; i++) {
Callback<Void> cb = readLogEndOffsetCallbacks.poll();
cb.onCompletion(null, null);
}
}
try {
poll(Integer.MAX_VALUE);
} catch (WakeupException e) {
continue;
}
}
} catch (Throwable t) {
log.error("Unexpected exception in {}", this, t);
}
}
}这样在KafkaConnect中，就可以根据connector提供key拿到对应的offset信息。
最后再来看下KafkaBaseLog发送消息的逻辑
public void send(K key, V value) {
send(key, value, null);
}
public void send(K key, V value, org.apache.kafka.clients.producer.Callback callback) {
producer.send(new ProducerRecord<>(topic, key, value), callback);
}可以看到KafkaBaseLog本身只负责消息的发送和接受，并不处理key、value的序列化和反序列化，相关序列化部分由外部的调用方去实现。
3、总结
Kafka本身也可以作为一个数据库去永久存储消息，只需要把当前topic的cleanup.policy设置为compact即可（后面想到啥再回来总结(●'◡'●)）。