上篇·flink 1.4利用kafka0.11实现完整的一致性语义

发表于 | 分类于 | | |
字数统计 3k | 阅读时长 15

flink kafka-connector0.10版本分析，与1.4版本中kafka11对比

引言

官方文档在出了1.4之后特意发表了一篇blog，通过以下这两个条件实现了真正意义上的exactly once语义

  1. kafka producer0.11的事务性
  2. two phase commit protocol

我们先看0.10版本的kafka-connector的行为逻辑.

Kafka-Connector10

kafkaConsumer10

FlinkKafkaConsumerBase

这个抽象类实现了CheckpointedFunction, 这个接口的描述：

1
2
3
4
* This is the core interface for <i>stateful transformation functions</i>, meaning functions
* that maintain state across individual stream records.
* While more lightweight interfaces exist as shortcuts for various types of state, this interface offer the
* greatest flexibility in managing both <i>keyed state</i> and <i>operator state</i>.

这个接口中主要要去做两件事情：

1
2
3
4
5
//每一次做checkpoint的时候被调用
void snapshotState(FunctionSnapshotContext context) throws Exception;

//初始化每一个并发的实例的时候被调用
void initializeState(FunctionInitializationContext context) throws Exception;

初始化函数的调用时机是在open之前的：

1
2
3
//StreamTask.java
initializeState();
openAllOperators();

在初始化的函数中提供了一个FunctionSnapshotContext

1
void snapshotState(FunctionSnapshotContext context) throws Exception;

让你既可以注册一个KeyedStateStore，也可以注册一个OperatorStateStore

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public interface ManagedInitializationContext {

	/**
	 * Returns true, if state was restored from the snapshot of a previous execution. This returns always false for
	 * stateless tasks.
	 */
	boolean isRestored();

	/**
	 * Returns an interface that allows for registering operator state with the backend.
	 */
	OperatorStateStore getOperatorStateStore();

	/**
	 * Returns an interface that allows for registering keyed state with the backend.
	 */
	KeyedStateStore getKeyedStateStore();

}

我们可以看到kafka10是怎么利用这个CheckpointedFunction来管理记录内部offset的呢？

initializeState

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
//初始化过程
public final void initializeState(FunctionInitializationContext context) throws Exception {

		// we might have been restored via restoreState() which restores from legacy operator state
		if (!restored) {
			restored = context.isRestored();
		}

		OperatorStateStore stateStore = context.getOperatorStateStore();
		offsetsStateForCheckpoint = stateStore.getSerializableListState(DefaultOperatorStateBackend.DEFAULT_OPERATOR_STATE_NAME);
		//如果是在重启恢复的过程中
		if (context.isRestored()) {
			if (restoredState == null) {
				restoredState = new HashMap<>();
				for (Tuple2<KafkaTopicPartition, Long> kafkaOffset : offsetsStateForCheckpoint.get()) {
				//partition和相应的offset数
					restoredState.put(kafkaOffset.f0, kafkaOffset.f1);
				}

				LOG.info("Setting restore state in the FlinkKafkaConsumer.");
				if (LOG.isDebugEnabled()) {
					LOG.debug("Using the following offsets: {}", restoredState);
				}
			}
		} else {
			LOG.info("No restore state for FlinkKafkaConsumer.");
		}
	}

那么我们看到恢复或者初始化的时候将信息保存到了一组HashMap中，但是这个Map怎么作用于消费阶段呢？

1
2
//将从状态中获取到的列表赋予给消费的列表
subscribedPartitionsToStartOffsets = restoredState;

snapshot

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@Override
public final void snapshotState(FunctionSnapshotContext context) throws Exception {
	if (!running) {
		LOG.debug("snapshotState() called on closed source");
	} else {

		offsetsStateForCheckpoint.clear();

		final AbstractFetcher<?, ?> fetcher = this.kafkaFetcher;
		if (fetcher == null) {
			// fetcher还没有初始化，返回上一次恢复的partition和offset，也就是这里会有个bug。我提了个issue:[FLINK-8869][2]
			for (Map.Entry<KafkaTopicPartition, Long> subscribedPartition : subscribedPartitionsToStartOffsets.entrySet()) {
				offsetsStateForCheckpoint.add(Tuple2.of(subscribedPartition.getKey(), subscribedPartition.getValue()));
			}
			//
			if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
				// the map cannot be asynchronously updated, because only one checkpoint call can happen
				// on this function at a time: either snapshotState() or notifyCheckpointComplete()
				pendingOffsetsToCommit.put(context.getCheckpointId(), restoredState);
			}
		} else {
			HashMap<KafkaTopicPartition, Long> currentOffsets = fetcher.snapshotCurrentState();

			if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
				// the map cannot be asynchronously updated, because only one checkpoint call can happen
				// on this function at a time: either snapshotState() or notifyCheckpointComplete()
				pendingOffsetsToCommit.put(context.getCheckpointId(), currentOffsets);
			}

			for (Map.Entry<KafkaTopicPartition, Long> kafkaTopicPartitionLongEntry : currentOffsets.entrySet()) {
				offsetsStateForCheckpoint.add(
						Tuple2.of(kafkaTopicPartitionLongEntry.getKey(), kafkaTopicPartitionLongEntry.getValue()));
			}
		}

		if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
			// truncate the map of pending offsets to commit, to prevent infinite growth
			while (pendingOffsetsToCommit.size() > MAX_NUM_PENDING_CHECKPOINTS) {
				pendingOffsetsToCommit.remove(0);
			}
		}
	}
}

offsetCommitMode

如上述代码中的offsetCommitMode,主要有以下几种

1
2
3
DISABLED,
ON_CHECKPOINTS, // 完成一次checkpoint后向kafka提交消费offset，只有这种模式下才需要我们手动去提交offset到kafka
KAFKA_PERIODIC;

这个配置主要改变了commitOffset回kafka的时机. 首先在snapshot的时候会将对应的checkpointId和相应的offset的列表放入pendingOffsetsToCommit, 在checkpoint完成后回调notifyCheckpointComplete，这里面主要完成了offset的commit工作。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
			// only one commit operation must be in progress
			if (LOG.isDebugEnabled()) {
				LOG.debug("Committing offsets to Kafka/ZooKeeper for checkpoint " + checkpointId);
			}

			try {
				final int posInMap = pendingOffsetsToCommit.indexOf(checkpointId);
				if (posInMap == -1) {
					LOG.warn("Received confirmation for unknown checkpoint id {}", checkpointId);
					return;
				}

				@SuppressWarnings("unchecked")
				HashMap<KafkaTopicPartition, Long> offsets =
					(HashMap<KafkaTopicPartition, Long>) pendingOffsetsToCommit.remove(posInMap);

				// remove older checkpoints in map
				for (int i = 0; i < posInMap; i++) {
					pendingOffsetsToCommit.remove(0);
				}

				if (offsets == null || offsets.size() == 0) {
					LOG.debug("Checkpoint state was empty.");
					return;
				}

				fetcher.commitInternalOffsetsToKafka(offsets, offsetCommitCallback);
			} catch (Exception e) {
				if (running) {
					throw e;
				}
				// else ignore exception if we are no longer running
			}

消费partition分配问题

  • 在从上次点恢复的情况下是直接从state中获取应该读取哪一个partition，offset。如果并发度改变了会做出什么样的反馈呢?会正确做出rescale吗

  • 第一次进行读取的时候会初始化处当前task（并发度）所需要订阅的partition

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
protected static Map<KafkaTopicPartition, Long> initializeSubscribedPartitionsToStartOffsets(
			List<KafkaTopicPartition> kafkaTopicPartitions, //topic的所有partition
			int indexOfThisSubtask, // 当前task的维度
			int numParallelSubtasks, // 总的并发度
			StartupMode startupMode, // 从哪个offset消费的模式（最新，最老，指定offset）
			Map<KafkaTopicPartition, Long> specificStartupOffsets) {

		Map<KafkaTopicPartition, Long> subscribedPartitionsToStartOffsets = new HashMap<>(kafkaTopicPartitions.size());

		for (KafkaTopicPartition kafkaTopicPartition : kafkaTopicPartitions) {
			// only handle partitions that this subtask should subscribe to（选取当前subtask所需要订阅的partition）
			if (KafkaTopicPartitionAssigner.assign(kafkaTopicPartition, numParallelSubtasks) == indexOfThisSubtask) {
				if (startupMode != StartupMode.SPECIFIC_OFFSETS) {
					//StateSentinel都是一串随机的负数占位符(都是一个标记在KafkaConsumerThread中进行判断)
					subscribedPartitionsToStartOffsets.put(kafkaTopicPartition, startupMode.getStateSentinel());
				} else {
					if (specificStartupOffsets == null) {
						throw new IllegalArgumentException(
							"Startup mode for the consumer set to " + StartupMode.SPECIFIC_OFFSETS +
								", but no specific offsets were specified");
					}

					Long specificOffset = specificStartupOffsets.get(kafkaTopicPartition);
					if (specificOffset != null) {
						// since the specified offsets represent the next record to read, we subtract
						// it by one so that the initial state of the consumer will be correct
						// 这里需要减去1
						subscribedPartitionsToStartOffsets.put(kafkaTopicPartition, specificOffset - 1);
					} else {
						subscribedPartitionsToStartOffsets.put(kafkaTopicPartition, KafkaTopicPartitionStateSentinel.GROUP_OFFSET);
					}
				}
			}
		}

		return subscribedPartitionsToStartOffsets;
}
1
2
3
4
5
6
7
8
9
10
11
public static int assign(KafkaTopicPartition partition, int numParallelSubtasks) {
	int startIndex = ((partition.getTopic().hashCode() * 31) & 0x7FFFFFFF) % numParallelSubtasks;

	// here, the assumption is that the id of Kafka partitions are always ascending
	// starting from 0, and therefore can be used directly as the offset clockwise from the start index
	// 这里看出：每个Partition只会分配到一个subtask来消费
	// 1. partition > parallel 一个subtask会订阅多个partition
	// 2. partition < parallel 有subtask会是空闲的
	// 3. startIndex由topic名字计算得出
	return (startIndex + partition.getPartition()) % numParallelSubtasks;
}

消费模型kafka10

在完成partition订阅之后，就要开始真正的run方法了，FlinkKafkaConsumer也是实现自SouceFunction，因此主要的逻辑也都是在run方法中实现。 主要逻辑：

kafkaConsumerThread和Kafka10Fetcher通过Handover交互,我觉得这段代码写的很不错，可以好好学习下。可以形象的比作在接力跑：kafkaConsumerThread通过真正的消费线程消费放入一个HandOver，再由kafkaFetcher去poll，完成整个消费过程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
// we need only do work, if we actually have partitions assigned
if (!subscribedPartitionsToStartOffsets.isEmpty()) {

	// create the fetcher that will communicate with the Kafka brokers
	final AbstractFetcher<T, ?> fetcher = createFetcher(
			sourceContext,
			subscribedPartitionsToStartOffsets,
			periodicWatermarkAssigner,
			punctuatedWatermarkAssigner,
			(StreamingRuntimeContext) getRuntimeContext(),
			offsetCommitMode);

	// publish the reference, for snapshot-, commit-, and cancel calls
	// IMPORTANT: We can only do that now, because only now will calls to
	//            the fetchers 'snapshotCurrentState()' method return at least
	//            the restored offsets
	this.kafkaFetcher = fetcher;
	if (!running) {
		return;
	}
	
	// (3) run the fetcher' main work method
	// 主要工作方法
	fetcher.runFetchLoop();
}
else {
	// this source never completes, so emit a Long.MAX_VALUE watermark
	// to not block watermark forwarding
	// 发送最大的watermark就不会block住下游的watermark更新
	sourceContext.emitWatermark(new Watermark(Long.MAX_VALUE));

	// wait until this is canceled
	final Object waitLock = new Object();
	while (running) {
		try {
			//noinspection SynchronizationOnLocalVariableOrMethodParameter
			synchronized (waitLock) {
				waitLock.wait();
			}
		}
		catch (InterruptedException e) {
			if (!running) {
				// restore the interrupted state, and fall through the loop
				// 打断当前线程
				Thread.currentThread().interrupt();
			}
		}
	}
}
1
2
3
4
5
// Handover的描述， Handover代码可以再好好学习下。
* The Handover is a utility to hand over data (a buffer of records) and exception from a
* <i>producer</i> thread to a <i>consumer</i> thread. It effectively behaves like a
* "size one blocking queue", with some extras around exception reporting, closing, and
* waking up thread without {@link Thread#interrupt() interrupting} threads.

KafkaFetcher

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
public void runFetchLoop() throws Exception {
	try {
		final Handover handover = this.handover;

		// kick off the actual Kafka consumer
		// 启动真正的消费线程
		consumerThread.start();

		while (running) {
			// this blocks until we get the next records
			// it automatically re-throws exceptions encountered in the fetcher thread
			// 在handover中获取真正的数据，并抛出其他线程中的异常
			final ConsumerRecords<byte[], byte[]> records = handover.pollNext();

			// get the records for each topic partition
			// subscribedPartitionStates维护的是每个partition的状态（partition，KPH（partition的描述，依据版本可能不同），offset, committedOffset, watermark）
			for (KafkaTopicPartitionState<TopicPartition> partition : subscribedPartitionStates()) {

				List<ConsumerRecord<byte[], byte[]>> partitionRecords =
						records.records(partition.getKafkaPartitionHandle());

				for (ConsumerRecord<byte[], byte[]> record : partitionRecords) {
					final T value = deserializer.deserialize(
							record.key(), record.value(),
							record.topic(), record.partition(), record.offset());

					if (deserializer.isEndOfStream(value)) {
						// end of stream signaled
						running = false;
						break;
					}

					// emit the actual record. this also updates offset state atomically
					// and deals with timestamps and watermark generation
					// 这里会进入真正的通过sourceContext发送数据的代码 如下
					emitRecord(value, partition, record.offset(), record);
				}
			}
		}
	}
	finally {
		// this signals the consumer thread that no more work is to be done
		consumerThread.shutdown();
	}

	// on a clean exit, wait for the runner thread
	try {
		consumerThread.join();
	}
	catch (InterruptedException e) {
		// may be the result of a wake-up interruption after an exception.
		// we ignore this here and only restore the interruption state
		Thread.currentThread().interrupt();
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
protected void emitRecordWithTimestamp(
		T record, KafkaTopicPartitionState<KPH> partitionState, long offset, long timestamp) throws Exception {

	if (record != null) {
		if (timestampWatermarkMode == NO_TIMESTAMPS_WATERMARKS) {
			// fast path logic, in case there are no watermarks generated in the fetcher

			// emit the record, using the checkpoint lock to guarantee
			// atomicity of record emission and offset state update
			synchronized (checkpointLock) {
				sourceContext.collectWithTimestamp(record, timestamp);
				partitionState.setOffset(offset);
			}
		} else if (timestampWatermarkMode == PERIODIC_WATERMARKS) {
		    // 更新partitionstate中的watermark状态
			emitRecordWithTimestampAndPeriodicWatermark(record, partitionState, offset, timestamp);
		} else {
			emitRecordWithTimestampAndPunctuatedWatermark(record, partitionState, offset, timestamp);
		}
	} else {
		// if the record is null, simply just update the offset state for partition
		synchronized (checkpointLock) {
			partitionState.setOffset(offset);
		}
	}
}

在设置了kafkaTimestampassigner之后就会进行一个定时任务向下游发送watermark，值为所有partition维护的最小值：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
public void onProcessingTime(long timestamp) throws Exception {

	long minAcrossAll = Long.MAX_VALUE;
	for (KafkaTopicPartitionStateWithPeriodicWatermarks<?, ?> state : allPartitions) {
		
		// we access the current watermark for the periodic assigners under the state
		// lock, to prevent concurrent modification to any internal variables
		final long curr;
		//noinspection SynchronizationOnLocalVariableOrMethodParameter
		// 这个锁是防止别的线程修改其他变量
		synchronized (state) {
			curr = state.getCurrentWatermarkTimestamp();
		}
		
		minAcrossAll = Math.min(minAcrossAll, curr);
	}
	
	// emit next watermark, if there is one
	if (minAcrossAll > lastWatermarkTimestamp) {
		lastWatermarkTimestamp = minAcrossAll;
		emitter.emitWatermark(new Watermark(minAcrossAll));
	}
	
	// schedule the next watermark
	timerService.registerTimer(timerService.getCurrentProcessingTime() + interval, this);
}

KafkaConsumerThread

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
if (records == null) {
				try {
					records = consumer.poll(pollTimeout);
				}
				catch (WakeupException we) {
					continue;
				}
			}

			try {
				handover.produce(records);
				records = null;
			}
			catch (Handover.WakeupException e) {
				// fall through the loop
			}

主要做的工作就是从consumer消费数据塞入handover，等待拉取

Handover

桥接模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public class KafkaConsumerCallBridge {

	public void assignPartitions(KafkaConsumer<?, ?> consumer, List<TopicPartition> topicPartitions) throws Exception {
		consumer.assign(topicPartitions);
	}

	public void seekPartitionToBeginning(KafkaConsumer<?, ?> consumer, TopicPartition partition) {
		consumer.seekToBeginning(partition);
	}

	public void seekPartitionToEnd(KafkaConsumer<?, ?> consumer, TopicPartition partition) {
		consumer.seekToEnd(partition);
	}

}

解决08 09 10 版本的api不兼容问题

Kafkaproducer10

###initializeState

什么都不做

snapshot

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@Override
public void snapshotState(FunctionSnapshotContext ctx) throws Exception {
	// check for asynchronous errors and fail the checkpoint if necessary
	checkErroneous();

	if (flushOnCheckpoint) {
		// flushing is activated: We need to wait until pendingRecords is 0
		flush();
		synchronized (pendingRecordsLock) {
			if (pendingRecords != 0) {
				throw new IllegalStateException("Pending record count must be zero at this point: " + pendingRecords);
			}

			// if the flushed requests has errors, we should propagate it also and fail the checkpoint
			checkErroneous();
		}
	}
}

这里涉及到一个flushOnCheckPoint的问题，再调用producer.flush期间，producer会将所有没写入的，在buffer中的数据刷盘，然后调用commitCallBack，这就保证了ckpt之后数据不会丢的问题。

主要工作方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
public void invoke(IN next) throws Exception {
	// propagate asynchronous errors
	checkErroneous();

	byte[] serializedKey = schema.serializeKey(next);
	byte[] serializedValue = schema.serializeValue(next);
	// 每条元素可以自己自己要写到的topic
	String targetTopic = schema.getTargetTopic(next);
	if (targetTopic == null) {
		targetTopic = defaultTopicId;
	}

	int[] partitions = this.topicPartitionsMap.get(targetTopic);
	if(null == partitions) {
		partitions = getPartitionsByTopic(targetTopic, producer);
		this.topicPartitionsMap.put(targetTopic, partitions);
	}

	ProducerRecord<byte[], byte[]> record;
	if (flinkKafkaPartitioner == null) {
		record = new ProducerRecord<>(targetTopic, serializedKey, serializedValue);
	} else {
		record = new ProducerRecord<>(
				targetTopic,
				flinkKafkaPartitioner.partition(next, serializedKey, serializedValue, targetTopic, partitions),
				serializedKey,
				serializedValue);
	}
	if (flushOnCheckpoint) {
		synchronized (pendingRecordsLock) {
			pendingRecords++;
		}
	}
	producer.send(record, callback);
}

问题

  1. kafka恢复状态直接从状态中去获取了之前保存的partition和offset，但是如果是扩容partition的场景就不会从新的Partition消费 issue:FLINK-8869

  2. flink内部维护了offset，为什么向kafka提交的时候还需要在checkpoint之后再提交而不是定时提交就算了？

    因为虽然从checkpoint点恢复的时候不需要从kafka broker获取消费点的位置了，但是如果是应用重启消费上次消费到的点的数据，这个offset就是flink向kafka提交的，放在checkpoint完成后去做的好处就是让应用即使不是从上个点恢复的，也能够从kafka消费正确的offset点。

  3. 如果在新的checkpoint没打之前任务失败了，重新从上次的offset点消费的话下游数据是不是重复了?

    是的，因为有一部分数据经过处理已经sink出去了，因此才需要0.11的一致性语义

以上代码：
kafka-connector0.10 来源于release1.3.2
kafka-connector0.11 来源于release1.4.0

谢谢支持