Savonitar commented on code in PR #293:
URL:
https://github.com/apache/flink-connector-kafka/pull/293#discussion_r3882403570
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -270,7 +279,7 @@ public void pauseOrResumeSplits(
public void notifyCheckpointComplete(
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit,
OffsetCommitCallback offsetCommitCallback) {
- consumer.commitAsync(offsetsToCommit, offsetCommitCallback);
+ consumer.commitAsync(reconcileOffsetsToCommit(offsetsToCommit),
offsetCommitCallback);
Review Comment:
reconcileOffsetsToCommit creates a new map, so the offsets that we send to
Kafka here can differ from the original committedPartitions map from
[KafkaSourceReader](https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java#L126)
(that callback
[ignores](https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java#L148)
its first argument and updates offsets from original map). That means Kafka
will report 2 while Flink's committed-offset gauge will report 1. Which is a
contradiction.
Could we update the callback to record metrics from its offsets argument?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -298,6 +307,58 @@ long getConsumerPosition(TopicPartition tp, String msg) {
return retryOnWakeup(() -> consumer.position(tp), msg);
}
+ private void trackLastFetchedRecordOffsets(ConsumerRecords<byte[], byte[]>
consumerRecords) {
+ for (TopicPartition tp : consumerRecords.partitions()) {
+ List<ConsumerRecord<byte[], byte[]>> partitionRecords =
consumerRecords.records(tp);
+ if (!partitionRecords.isEmpty()) {
+ lastFetchedOffsets.put(
+ tp, partitionRecords.get(partitionRecords.size() -
1).offset());
+ }
+ }
+ }
+
+ /**
+ * Advances the offsets to commit over the entries that the Kafka consumer
read but never
+ * delivered, such as transaction control markers and records of aborted
transactions.
+ *
+ * <p>{@link KafkaRecordEmitter} derives the offset to commit from the
records it receives, so
+ * the offset stops at the first entry that Kafka does not deliver, for as
long as the partition
+ * is idle. The consumer's own position accounts for those entries, so it
is the offset that
+ * external tooling expects to see.
+ *
+ * <p>This relies on {@link #lastKnownPositions}, populated as a side
effect of the regular
+ * {@link #fetch()} poll loop, rather than querying the consumer for the
position again here.
+ * {@link #notifyCheckpointComplete} runs on this same split fetcher
thread, but at a point
+ * outside that poll loop, so this allows us to avoid a separate blocking
call to the consumer.
+ */
+ private Map<TopicPartition, OffsetAndMetadata> reconcileOffsetsToCommit(
+ Map<TopicPartition, OffsetAndMetadata> offsetsToCommit) {
+ Map<TopicPartition, OffsetAndMetadata> reconciled = new
HashMap<>(offsetsToCommit);
+ Set<TopicPartition> assignment = consumer.assignment();
+ offsetsToCommit.forEach(
+ (tp, offsetAndMetadata) -> {
+ if (!assignment.contains(tp) ||
stoppingOffsets.containsKey(tp)) {
+ return;
+ }
+ Long lastFetchedOffset = lastFetchedOffsets.get(tp);
+ if (lastFetchedOffset == null
Review Comment:
Can lastFetchedOffset be legimately null when the consumer position has
advanced? e.g. read_committed consumer position moves past entries without
delivering anything.
example:
partition has: 0-5 offsets as aborted records, 6 offset abort marker, 7
offset LSO.
consumer position: 7
lastFetchedOffsets will be null (nothing was delivered)
offset that will be commited: 0
As a result: The reader is fully caught up (position 7 equals the LSO)
meanwhile every checkpoint commits 0. Monitoring shows a lag of 7, forever,
until some producer finally writes a committed record.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -76,6 +76,12 @@ public class KafkaPartitionSplitReader
// Tracking empty splits that has not been added to finished splits in
fetch()
private final Set<String> emptySplits = new HashSet<>();
+ // Offset of the last record that fetch() handed to the source reader, per
partition
+ private final Map<TopicPartition, Long> lastFetchedOffsets = new
HashMap<>();
+
+ // Consumer position observed at the end of the most recent fetch(), per
partition.
+ private final Map<TopicPartition, Long> lastKnownPositions = new
HashMap<>();
Review Comment:
Do (when?) we cleanup these entries? Or we never cleanup them?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -298,6 +307,58 @@ long getConsumerPosition(TopicPartition tp, String msg) {
return retryOnWakeup(() -> consumer.position(tp), msg);
}
+ private void trackLastFetchedRecordOffsets(ConsumerRecords<byte[], byte[]>
consumerRecords) {
+ for (TopicPartition tp : consumerRecords.partitions()) {
+ List<ConsumerRecord<byte[], byte[]>> partitionRecords =
consumerRecords.records(tp);
+ if (!partitionRecords.isEmpty()) {
+ lastFetchedOffsets.put(
+ tp, partitionRecords.get(partitionRecords.size() -
1).offset());
+ }
+ }
+ }
+
+ /**
+ * Advances the offsets to commit over the entries that the Kafka consumer
read but never
+ * delivered, such as transaction control markers and records of aborted
transactions.
+ *
+ * <p>{@link KafkaRecordEmitter} derives the offset to commit from the
records it receives, so
+ * the offset stops at the first entry that Kafka does not deliver, for as
long as the partition
+ * is idle. The consumer's own position accounts for those entries, so it
is the offset that
+ * external tooling expects to see.
+ *
+ * <p>This relies on {@link #lastKnownPositions}, populated as a side
effect of the regular
+ * {@link #fetch()} poll loop, rather than querying the consumer for the
position again here.
+ * {@link #notifyCheckpointComplete} runs on this same split fetcher
thread, but at a point
+ * outside that poll loop, so this allows us to avoid a separate blocking
call to the consumer.
+ */
+ private Map<TopicPartition, OffsetAndMetadata> reconcileOffsetsToCommit(
+ Map<TopicPartition, OffsetAndMetadata> offsetsToCommit) {
+ Map<TopicPartition, OffsetAndMetadata> reconciled = new
HashMap<>(offsetsToCommit);
+ Set<TopicPartition> assignment = consumer.assignment();
+ offsetsToCommit.forEach(
+ (tp, offsetAndMetadata) -> {
+ if (!assignment.contains(tp) ||
stoppingOffsets.containsKey(tp)) {
Review Comment:
Is it safe to skip reconcilation for splits with stopped offsets? Could you
please clarify why?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
