MartijnVisser commented on code in PR #293:
URL:
https://github.com/apache/flink-connector-kafka/pull/293#discussion_r4054494816
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -285,6 +294,16 @@ Duration getPollTimeout() {
return pollTimeout;
}
+ @VisibleForTesting
+ Map<TopicPartition, Long> lastFetchedOffsets() {
Review Comment:
AGENTS.md rules out adding lines to the violation stores. Package-private
fields without these two accessors keep it unchanged, like
`getConsumerPosition` below.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java:
##########
@@ -266,6 +277,456 @@ void testOffsetCommitOnCheckpointComplete() throws
Exception {
}
}
+ /** Writes the records that a {@link #offsetConvergenceScenarios} scenario
needs. */
+ @FunctionalInterface
+ private interface RecordProducer {
+ void produce(String topic) throws Throwable;
+ }
+
+ private static Stream<Arguments> offsetConvergenceScenarios() {
Review Comment:
16 tests in 34s becomes 26 in 173s here. Do all seven earn that?
`MultiRecordTransaction`, `MultipleTransactions` and `InterleavedTransactions`
exercise the same skip as `Transactional`.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java:
##########
@@ -159,9 +159,10 @@ public void notifyCheckpointComplete(long checkpointId)
throws Exception {
"Successfully committed offsets for
checkpoint {}",
checkpointId);
kafkaSourceReaderMetrics.recordSucceededCommit();
- // If the finished topic partition has been
committed, we remove it
- // from the offsets of the finished splits map.
- committedPartitions.forEach(
+ // offsets committed to Kafka can differ from
what was requested,
Review Comment:
The comment you replaced explained the `removeIf` below, not the metric
loop. Worth keeping that one too.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java:
##########
@@ -695,6 +1156,72 @@ private long getCommittedOffsetMetric(TopicPartition tp,
MetricListener listener
// ---------------------
+ private static KafkaConsumer<String, String> createReadCommittedProbe() {
+ final Properties props = new Properties();
+ props.putAll(KafkaSourceTestEnv.standardProps);
+ props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "read-probe-" +
UUID.randomUUID());
+ props.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,
"read_committed");
+ props.setProperty(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ props.setProperty(
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ return new KafkaConsumer<>(props);
+ }
+
+ private static int getReadCommittedRecordsCount(TopicPartition tp) {
+ try (KafkaConsumer<String, String> probe = createReadCommittedProbe())
{
+ final List<TopicPartition> partitions =
Collections.singletonList(tp);
+ probe.assign(partitions);
+ probe.seekToBeginning(partitions);
+ final long lastStableOffset = probe.endOffsets(partitions).get(tp);
+ int count = 0;
+ final long deadline = System.currentTimeMillis() + 30_000L;
+ while (probe.position(tp) < lastStableOffset &&
System.currentTimeMillis() < deadline) {
+ List<ConsumerRecord<String, String>> records =
+ probe.poll(Duration.ofMillis(500)).records(tp);
+ count += records.size();
+ }
+ return count;
+ }
+ }
+
+ /**
+ * Waits until the last stable offset of {@code tp} reaches {@code
expectedOffset}, and returns
+ * it. In most cases, this will return the expected value on the first
check, but the
+ * transaction coordinator propagates it to partition leaders
asynchronously, so LSO can briefly
+ * lag behind a committed transaction. To avoid introducing a test race
condition, this method
+ * checks again after a brief wait.
+ */
+ private static long awaitLastStableOffset(TopicPartition tp, long
expectedOffset)
+ throws Exception {
+ try (KafkaConsumer<String, String> probe = createReadCommittedProbe())
{
+ final List<TopicPartition> partitions =
Collections.singletonList(tp);
+ long lastStableOffset = probe.endOffsets(partitions).get(tp);
+ final long deadline = System.currentTimeMillis() + 3_000L;
Review Comment:
Three seconds is tight for marker propagation on a loaded runner, and this
asserts rather than retries. Thirty seconds costs nothing here.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -125,6 +131,7 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>>
fetch() throws IOExce
for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
long consumerPosition = getConsumerPosition(tp, "retrieving
consumer position");
+ lastKnownPositions.put(tp, consumerPosition);
Review Comment:
`KafkaPartitionSplitRecords.nextSplit()` walks
`consumerRecords.partitions()`, so a partition with no records in a poll never
reaches the emitter and its state never moves. That is the idle case here.
##########
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:
The guard is needed. Without it
`testCommittedOffsetForBoundedSplitDoesNotGoBackwards` goes red: once the split
finishes the offset comes from `offsetsOfFinishedSplits`, unreconciled, so it
drops back.
--
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]
