You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "dongwoo6kim (via GitHub)" <gi...@apache.org> on 2024/04/29 03:15:05 UTC
[PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
dongwoo6kim opened a new pull request, #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100
## Problem
When using the flink kafka connector in batch scenarios, consuming transactional messages can cause indefinite hanging.
This issue can be easily reproduced with following steps.
1. Produce transactional messages and commit them.
2. Configure `scan.bounded.mode` to `latest-offset` and run consumer using flink kafka connector
## Cause
The previous stopping condition in the `KafkaPartitionSplitReader` compared the offset of the last record with the `stoppingOffset`. This approach works for streaming use cases and batch processing of non-transactional messages. However, in scenarios involving transactional messages, this is insufficient.
[Control messages](https://kafka.apache.org/documentation/#controlbatch), which are not visible to clients, can occupy the entire range between the last record's offset and the stoppingOffset which leads to indefinite blocking.
## Workaround
I've modified the stopping condition to use `consumer.position(tp)`, which effectively skips any control messages present in the current poll, pointing directly to the next record's offset.
To handle edge cases, particularly when `properties.max.poll.records` is set to `1`, I've adjusted the fetch method to always check all assigned partitions, even if no records are returned in a poll.
#### Edge case example
Consider partition `0`, where offsets `13` and `14` are valid records and `15` is a control record. If `stoppingOffset` is set to 15 for partition `0`and `properties.max.poll.records` is configured to `1`, checking only partitions that return records would miss offset 15. By consistently reviewing all assigned partitions, the consumer’s position jumps control record in the subsequent poll, allowing the system to escape.
## Discussion
To address the metric issue in [FLINK-33484](https://issues.apache.org/jira/browse/FLINK-33484), I think we need to make wrapper class of `ConsumerRecord` for example `ConsumerRecordWithOffsetJump`.
```java
public ConsumerRecordWithOffsetJump(ConsumerRecord<K, V> record, long offsetJump) {
this.record = record;
this.offsetJump = offsetJump;
}
```
And we may need new `KafkaPartitionSplitReader` that implements `SplitReader<ConsumerRecordWithOffsetJump<byte[], byte[]>, KafkaPartitionSplit>`.
So when record is emitted it should set current offset not just `record.offset()+1` but`record.offset() + record.jumpValue` in [here](https://github.com/apache/flink-connector-kafka/blob/369e7be46a70fd50d68746498aed82105741e7d6/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaRecordEmitter.java#L54).
`jumpValue` is typically 1, except for the last record of each poll where it's calculated as `consumer.position() - lastRecord.offset()`.
If this sounds good to everyone, I'm happy to work on this.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1762233642
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java:
##########
@@ -369,6 +370,38 @@ public void testConsumingTopicWithEmptyPartitions() throws Throwable {
WatermarkStrategy.noWatermarks(),
"testConsumingTopicWithEmptyPartitions"));
}
+
+ @Test
+ @Timeout(value = 10)
+ public void testConsumingTransactionalMessage() throws Throwable {
+ String transactionalTopic = "transactionalTopic-" + UUID.randomUUID();
+ KafkaSourceTestEnv.createTestTopic(
+ transactionalTopic, KafkaSourceTestEnv.NUM_PARTITIONS, 1);
+ List<ProducerRecord<String, Integer>> records =
+ KafkaSourceTestEnv.getRecordsForTopic(transactionalTopic);
+ // Prepare records for executeAndVerify method
+ records.removeIf(record -> record.partition() > record.value());
Review Comment:
I've made two changes.
1. Replaced `records.removeIf(record -> record.partition() > record.value())` with `KafkaSourceTestEnv.setupEarliestOffsets(transactionalTopic)`, as it serves a similar purpose.
2. Added a comment explaining how the data looks after the setup.
> I'm assuming this will preserve one record per partition?
After data modification, each partition p will contain records from p to NUM_RECORDS_PER_PARTITION (which is 10). For example, partition 1 has records 1 to 10, and partition 5 has records 5 to 10.
> what would happen if we retain all records originally generated
If we retain all records we need to make new assertion logic for the generated records.
The main purpose of this data modification setup is to reuse `executeAndVerify` method.
When you look at [here](https://github.com/apache/flink-connector-kafka/blob/7929b16dcfe648da30b6cc9755f63de2ed3d5319/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java#L505) `executeAndVerify` method expects the input data to be modified like this way.
I intended to reuse existing test util functions and follow the test code convention but if you think it is causing unnecessary confusion I can change the test code to have custom assertion logic.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#issuecomment-2164206983
Thanks for confirming @morazow,
Please feel free to provide any additional advice before merging this fix.
It would be also helpful if you could elaborate more on the issue you mentioned and consider adding relevant test code for it.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#issuecomment-2143497305
Hello @morazow I've added ITCase for this case.
This case fails for current main branch due to timeout and works okay with fixed code.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "LinMingQiang (via GitHub)" <gi...@apache.org>.
LinMingQiang commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1600879571
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -122,27 +122,21 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
KafkaPartitionSplitRecords recordsBySplits =
new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
List<TopicPartition> finishedPartitions = new ArrayList<>();
- for (TopicPartition tp : consumerRecords.partitions()) {
+ for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
- consumerRecords.records(tp);
-
- if (recordsFromPartition.size() > 0) {
- final ConsumerRecord<byte[], byte[]> lastRecord =
- recordsFromPartition.get(recordsFromPartition.size() - 1);
-
- // After processing a record with offset of "stoppingOffset - 1", the split reader
- // should not continue fetching because the record with stoppingOffset may not
- // exist. Keep polling will just block forever.
- if (lastRecord.offset() >= stoppingOffset - 1) {
- recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
- finishSplitAtRecord(
- tp,
- stoppingOffset,
- lastRecord.offset(),
- finishedPartitions,
- recordsBySplits);
- }
+ long consumerPosition = consumer.position(tp);
+ // Stop fetching when the consumer's position reaches the stoppingOffset.
+ // Control messages may follow the last record; therefore, using the last record's
+ // offset as a stopping condition could result in indefinite blocking.
+ if (consumerPosition >= stoppingOffset) {
+ LOG.debug(
+ "Position of {}: {}, has reached stopping offset: {}",
+ tp,
+ consumerPosition,
+ stoppingOffset);
+ recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
+ finishSplitAtRecord(
+ tp, stoppingOffset, consumerPosition, finishedPartitions, recordsBySplits);
}
// Track this partition's record lag if it never appears before
kafkaSourceReaderMetrics.maybeAddRecordsLagMetric(consumer, tp);
Review Comment:
we do not need to track tp when consumerRecords is empty.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#issuecomment-2355822850
Awesome work, congrats on your first merged pull request!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise merged PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1762219777
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java:
##########
@@ -369,6 +370,38 @@ public void testConsumingTopicWithEmptyPartitions() throws Throwable {
WatermarkStrategy.noWatermarks(),
"testConsumingTopicWithEmptyPartitions"));
}
+
+ @Test
+ @Timeout(value = 10)
Review Comment:
Got it, thanks for the explanation. I've deleted all custom timeout logic in test codes
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#issuecomment-2081822026
Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#issuecomment-2352165424
Seems like the deleted arch unit rule was still needed. What was your intent when you deleted it?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1759157019
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java:
##########
@@ -270,6 +270,42 @@ public static <K, V> void produceToKafka(
}
}
+ public static <K, V> void produceTransactionalMessageToKafka(
Review Comment:
I agree, I've applied your idea. Thanks for the input
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "LinMingQiang (via GitHub)" <gi...@apache.org>.
LinMingQiang commented on PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#issuecomment-2111454911
its work , i had try.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1762233642
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java:
##########
@@ -369,6 +370,38 @@ public void testConsumingTopicWithEmptyPartitions() throws Throwable {
WatermarkStrategy.noWatermarks(),
"testConsumingTopicWithEmptyPartitions"));
}
+
+ @Test
+ @Timeout(value = 10)
+ public void testConsumingTransactionalMessage() throws Throwable {
+ String transactionalTopic = "transactionalTopic-" + UUID.randomUUID();
+ KafkaSourceTestEnv.createTestTopic(
+ transactionalTopic, KafkaSourceTestEnv.NUM_PARTITIONS, 1);
+ List<ProducerRecord<String, Integer>> records =
+ KafkaSourceTestEnv.getRecordsForTopic(transactionalTopic);
+ // Prepare records for executeAndVerify method
+ records.removeIf(record -> record.partition() > record.value());
Review Comment:
I've made two changes.
1. Replaced `records.removeIf(record -> record.partition() > record.value())` with `KafkaSourceTestEnv.setupEarliestOffsets(transactionalTopic)`, as it serves a similar purpose.
2. Added a comment explaining how the data looks after the setup.
> I'm assuming this will preserve one record per partition?
After data modification, each partition p will contain records from p to NUM_RECORDS_PER_PARTITION (which is 10). For example, partition 1 has records 1 to 10, and partition 5 has records 5 to 10.
> what would happen if we retain all records originally generated
The main purpose of this data modification setup is to reuse `executeAndVerify` method.
When you look at [here](https://github.com/apache/flink-connector-kafka/blob/7929b16dcfe648da30b6cc9755f63de2ed3d5319/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java#L505) `executeAndVerify` method expects the input data to be modified like this way.
I intended to reuse existing test util functions and follow the test code convention but if you think it is causing unnecessary confusion I can change the test code to have custom assertion logic.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1762809895
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java:
##########
@@ -369,6 +370,38 @@ public void testConsumingTopicWithEmptyPartitions() throws Throwable {
WatermarkStrategy.noWatermarks(),
"testConsumingTopicWithEmptyPartitions"));
}
+
+ @Test
+ @Timeout(value = 10)
+ public void testConsumingTransactionalMessage() throws Throwable {
+ String transactionalTopic = "transactionalTopic-" + UUID.randomUUID();
+ KafkaSourceTestEnv.createTestTopic(
+ transactionalTopic, KafkaSourceTestEnv.NUM_PARTITIONS, 1);
+ List<ProducerRecord<String, Integer>> records =
+ KafkaSourceTestEnv.getRecordsForTopic(transactionalTopic);
+ // Prepare records for executeAndVerify method
+ records.removeIf(record -> record.partition() > record.value());
Review Comment:
If it's needed for verification, then all good. We should probably revise the setup then at a later point.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1760654197
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -122,32 +122,32 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
KafkaPartitionSplitRecords recordsBySplits =
new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
List<TopicPartition> finishedPartitions = new ArrayList<>();
- for (TopicPartition tp : consumerRecords.partitions()) {
+ for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
- consumerRecords.records(tp);
-
- if (recordsFromPartition.size() > 0) {
- final ConsumerRecord<byte[], byte[]> lastRecord =
- recordsFromPartition.get(recordsFromPartition.size() - 1);
-
- // After processing a record with offset of "stoppingOffset - 1", the split reader
- // should not continue fetching because the record with stoppingOffset may not
- // exist. Keep polling will just block forever.
- if (lastRecord.offset() >= stoppingOffset - 1) {
- recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
- finishSplitAtRecord(
- tp,
- stoppingOffset,
- lastRecord.offset(),
- finishedPartitions,
- recordsBySplits);
- }
+ long consumerPosition = consumer.position(tp);
+ // Stop fetching when the consumer's position reaches the stoppingOffset.
+ // Control messages may follow the last record; therefore, using the last record's
+ // offset as a stopping condition could result in indefinite blocking.
+ if (consumerPosition >= stoppingOffset) {
Review Comment:
Thank you very much for the detailed explanation.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1758735450
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -122,32 +122,32 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
KafkaPartitionSplitRecords recordsBySplits =
new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
List<TopicPartition> finishedPartitions = new ArrayList<>();
- for (TopicPartition tp : consumerRecords.partitions()) {
+ for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
- consumerRecords.records(tp);
-
- if (recordsFromPartition.size() > 0) {
- final ConsumerRecord<byte[], byte[]> lastRecord =
- recordsFromPartition.get(recordsFromPartition.size() - 1);
-
- // After processing a record with offset of "stoppingOffset - 1", the split reader
- // should not continue fetching because the record with stoppingOffset may not
- // exist. Keep polling will just block forever.
- if (lastRecord.offset() >= stoppingOffset - 1) {
- recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
- finishSplitAtRecord(
- tp,
- stoppingOffset,
- lastRecord.offset(),
- finishedPartitions,
- recordsBySplits);
- }
+ long consumerPosition = consumer.position(tp);
Review Comment:
This change solves the issue of not having any data in a specific partition. If I use the `OffsetsInitializer#latest()`, I may get starting offset=end offset and the old code would have never compared them. So good fix.
I checked that in this case, `position` needs to issue another call to the broker (to get the partition state) but I don't really see a way around that and this happens at most once per partition.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -122,32 +122,32 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
KafkaPartitionSplitRecords recordsBySplits =
new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
List<TopicPartition> finishedPartitions = new ArrayList<>();
- for (TopicPartition tp : consumerRecords.partitions()) {
+ for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
- consumerRecords.records(tp);
-
- if (recordsFromPartition.size() > 0) {
- final ConsumerRecord<byte[], byte[]> lastRecord =
- recordsFromPartition.get(recordsFromPartition.size() - 1);
-
- // After processing a record with offset of "stoppingOffset - 1", the split reader
- // should not continue fetching because the record with stoppingOffset may not
- // exist. Keep polling will just block forever.
- if (lastRecord.offset() >= stoppingOffset - 1) {
- recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
- finishSplitAtRecord(
- tp,
- stoppingOffset,
- lastRecord.offset(),
- finishedPartitions,
- recordsBySplits);
- }
+ long consumerPosition = consumer.position(tp);
+ // Stop fetching when the consumer's position reaches the stoppingOffset.
+ // Control messages may follow the last record; therefore, using the last record's
+ // offset as a stopping condition could result in indefinite blocking.
Review Comment:
This is not really clear to me. Are you discussing `if (consumerPosition > stoppingOffset) {`?
Is the stop offset inclusive or exclusive? It looks like exclusive both in the old and the new version. But that's a bit unintuitive if I would like to read all data currently present in the Kafka topic and this approach discards it.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java:
##########
@@ -369,6 +370,38 @@ public void testConsumingTopicWithEmptyPartitions() throws Throwable {
WatermarkStrategy.noWatermarks(),
"testConsumingTopicWithEmptyPartitions"));
}
+
+ @Test
+ @Timeout(value = 10)
Review Comment:
Why are you using Timeout here? We usually discourage the usage as any infra hiccup will result in test failure. Instead we rely on a global timeout where small infra hiccups are absorbed.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java:
##########
@@ -369,6 +370,38 @@ public void testConsumingTopicWithEmptyPartitions() throws Throwable {
WatermarkStrategy.noWatermarks(),
"testConsumingTopicWithEmptyPartitions"));
}
+
+ @Test
+ @Timeout(value = 10)
+ public void testConsumingTransactionalMessage() throws Throwable {
+ String transactionalTopic = "transactionalTopic-" + UUID.randomUUID();
+ KafkaSourceTestEnv.createTestTopic(
+ transactionalTopic, KafkaSourceTestEnv.NUM_PARTITIONS, 1);
+ List<ProducerRecord<String, Integer>> records =
+ KafkaSourceTestEnv.getRecordsForTopic(transactionalTopic);
+ // Prepare records for executeAndVerify method
+ records.removeIf(record -> record.partition() > record.value());
+
+ KafkaSourceTestEnv.produceTransactionalMessageToKafka(records);
+
+ KafkaSource<PartitionAndValue> source =
+ KafkaSource.<PartitionAndValue>builder()
+ .setBootstrapServers(KafkaSourceTestEnv.brokerConnectionStrings)
+ .setTopics(transactionalTopic)
+ .setGroupId("topic-with-transactional-message-test")
+ .setDeserializer(new TestingKafkaRecordDeserializationSchema(false))
+ .setStartingOffsets(OffsetsInitializer.earliest())
+ .setBounded(OffsetsInitializer.latest())
+ .build();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ executeAndVerify(
+ env,
+ env.fromSource(
+ source,
+ WatermarkStrategy.noWatermarks(),
+ "testConsumingTransactionalMessage"));
Review Comment:
Is this failing with the old code?
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java:
##########
@@ -188,6 +190,66 @@ public void testKafkaSourceSink() throws Exception {
deleteTestTopic(topic);
}
+ @Test
+ public void testKafkaSourceEmptyResultOnDeletedOffsets() throws Exception {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "bounded_" + format + "_" + UUID.randomUUID();
+ createTestTopic(topic, 1, 1);
+ // ---------- Produce an event time stream into Kafka -------------------
+ String groupId = getStandardProps().getProperty("group.id");
+ String bootstraps = getBootstrapServers();
+
+ final String createTable =
+ String.format(
+ "CREATE TABLE kafka (\n"
+ + " `user_id` INT,\n"
+ + " `item_id` INT,\n"
+ + " `behavior` STRING\n"
+ + ") WITH (\n"
+ + " 'connector' = '%s',\n"
+ + " 'topic' = '%s',\n"
+ + " 'properties.bootstrap.servers' = '%s',\n"
+ + " 'properties.group.id' = '%s',\n"
+ + " 'scan.startup.mode' = 'specific-offsets',\n"
+ + " 'scan.bounded.mode' = 'specific-offsets',\n"
+ + " 'scan.startup.specific-offsets' = 'partition:0,offset:1',\n"
+ + " 'scan.bounded.specific-offsets' = 'partition:0,offset:3',\n"
+ + " %s\n"
+ + ")\n",
+ KafkaDynamicTableFactory.IDENTIFIER,
+ topic,
+ bootstraps,
+ groupId,
+ formatOptions());
+ tEnv.executeSql(createTable);
+ List<Row> values =
+ Arrays.asList(
+ Row.of(1, 1102, "behavior 1"),
+ Row.of(2, 1103, "behavior 2"),
+ Row.of(3, 1104, "behavior 3"));
+ tEnv.fromValues(values).insertInto("kafka").execute().await();
+ // ---------- Delete events from Kafka -------------------
+ Map<Integer, Long> partitionOffsetsToDelete = new HashMap<>();
+ partitionOffsetsToDelete.put(0, 3L);
+ deleteRecords(topic, partitionOffsetsToDelete);
+ // ---------- Consume stream from Kafka -------------------
+ List<Row> results = new ArrayList<>();
+ assertTimeoutPreemptively(
Review Comment:
Again please remove all custom timeout logic.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -122,32 +122,32 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
KafkaPartitionSplitRecords recordsBySplits =
new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
List<TopicPartition> finishedPartitions = new ArrayList<>();
- for (TopicPartition tp : consumerRecords.partitions()) {
+ for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
- consumerRecords.records(tp);
-
- if (recordsFromPartition.size() > 0) {
- final ConsumerRecord<byte[], byte[]> lastRecord =
- recordsFromPartition.get(recordsFromPartition.size() - 1);
-
- // After processing a record with offset of "stoppingOffset - 1", the split reader
- // should not continue fetching because the record with stoppingOffset may not
- // exist. Keep polling will just block forever.
- if (lastRecord.offset() >= stoppingOffset - 1) {
- recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
- finishSplitAtRecord(
- tp,
- stoppingOffset,
- lastRecord.offset(),
- finishedPartitions,
- recordsBySplits);
- }
+ long consumerPosition = consumer.position(tp);
+ // Stop fetching when the consumer's position reaches the stoppingOffset.
+ // Control messages may follow the last record; therefore, using the last record's
+ // offset as a stopping condition could result in indefinite blocking.
+ if (consumerPosition >= stoppingOffset) {
Review Comment:
Afaik this is equivalent to the old `if (lastRecord.offset() >= stoppingOffset - 1) {`. This code gave the offset of the last method and your new version is using the offset of the next record.
Is this what you wanted? I'm not sure if this is fixing anything then except for the first fix above.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java:
##########
@@ -369,6 +370,38 @@ public void testConsumingTopicWithEmptyPartitions() throws Throwable {
WatermarkStrategy.noWatermarks(),
"testConsumingTopicWithEmptyPartitions"));
}
+
+ @Test
+ @Timeout(value = 10)
+ public void testConsumingTransactionalMessage() throws Throwable {
+ String transactionalTopic = "transactionalTopic-" + UUID.randomUUID();
+ KafkaSourceTestEnv.createTestTopic(
+ transactionalTopic, KafkaSourceTestEnv.NUM_PARTITIONS, 1);
+ List<ProducerRecord<String, Integer>> records =
+ KafkaSourceTestEnv.getRecordsForTopic(transactionalTopic);
+ // Prepare records for executeAndVerify method
+ records.removeIf(record -> record.partition() > record.value());
Review Comment:
Why is this necessary?
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java:
##########
@@ -188,6 +190,66 @@ public void testKafkaSourceSink() throws Exception {
deleteTestTopic(topic);
}
+ @Test
+ public void testKafkaSourceEmptyResultOnDeletedOffsets() throws Exception {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "bounded_" + format + "_" + UUID.randomUUID();
+ createTestTopic(topic, 1, 1);
+ // ---------- Produce an event time stream into Kafka -------------------
+ String groupId = getStandardProps().getProperty("group.id");
+ String bootstraps = getBootstrapServers();
+
+ final String createTable =
+ String.format(
+ "CREATE TABLE kafka (\n"
+ + " `user_id` INT,\n"
+ + " `item_id` INT,\n"
+ + " `behavior` STRING\n"
+ + ") WITH (\n"
+ + " 'connector' = '%s',\n"
+ + " 'topic' = '%s',\n"
+ + " 'properties.bootstrap.servers' = '%s',\n"
+ + " 'properties.group.id' = '%s',\n"
+ + " 'scan.startup.mode' = 'specific-offsets',\n"
+ + " 'scan.bounded.mode' = 'specific-offsets',\n"
+ + " 'scan.startup.specific-offsets' = 'partition:0,offset:1',\n"
+ + " 'scan.bounded.specific-offsets' = 'partition:0,offset:3',\n"
+ + " %s\n"
+ + ")\n",
+ KafkaDynamicTableFactory.IDENTIFIER,
+ topic,
+ bootstraps,
+ groupId,
+ formatOptions());
+ tEnv.executeSql(createTable);
+ List<Row> values =
+ Arrays.asList(
+ Row.of(1, 1102, "behavior 1"),
+ Row.of(2, 1103, "behavior 2"),
+ Row.of(3, 1104, "behavior 3"));
+ tEnv.fromValues(values).insertInto("kafka").execute().await();
+ // ---------- Delete events from Kafka -------------------
+ Map<Integer, Long> partitionOffsetsToDelete = new HashMap<>();
+ partitionOffsetsToDelete.put(0, 3L);
+ deleteRecords(topic, partitionOffsetsToDelete);
Review Comment:
Is it the deletion marker that would cause a hang?
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java:
##########
@@ -270,6 +270,42 @@ public static <K, V> void produceToKafka(
}
}
+ public static <K, V> void produceTransactionalMessageToKafka(
Review Comment:
Instead of code replication, how about patching `produceToKafka` to create transactions if transaction.id is set?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1759154224
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java:
##########
@@ -369,6 +370,38 @@ public void testConsumingTopicWithEmptyPartitions() throws Throwable {
WatermarkStrategy.noWatermarks(),
"testConsumingTopicWithEmptyPartitions"));
}
+
+ @Test
+ @Timeout(value = 10)
+ public void testConsumingTransactionalMessage() throws Throwable {
+ String transactionalTopic = "transactionalTopic-" + UUID.randomUUID();
+ KafkaSourceTestEnv.createTestTopic(
+ transactionalTopic, KafkaSourceTestEnv.NUM_PARTITIONS, 1);
+ List<ProducerRecord<String, Integer>> records =
+ KafkaSourceTestEnv.getRecordsForTopic(transactionalTopic);
+ // Prepare records for executeAndVerify method
+ records.removeIf(record -> record.partition() > record.value());
+
+ KafkaSourceTestEnv.produceTransactionalMessageToKafka(records);
+
+ KafkaSource<PartitionAndValue> source =
+ KafkaSource.<PartitionAndValue>builder()
+ .setBootstrapServers(KafkaSourceTestEnv.brokerConnectionStrings)
+ .setTopics(transactionalTopic)
+ .setGroupId("topic-with-transactional-message-test")
+ .setDeserializer(new TestingKafkaRecordDeserializationSchema(false))
+ .setStartingOffsets(OffsetsInitializer.earliest())
+ .setBounded(OffsetsInitializer.latest())
+ .build();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ executeAndVerify(
+ env,
+ env.fromSource(
+ source,
+ WatermarkStrategy.noWatermarks(),
+ "testConsumingTransactionalMessage"));
Review Comment:
Yes this test fails with the current main branch due to timeout. (when `@Timeout` is used)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1759150772
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -122,32 +122,32 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
KafkaPartitionSplitRecords recordsBySplits =
new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
List<TopicPartition> finishedPartitions = new ArrayList<>();
- for (TopicPartition tp : consumerRecords.partitions()) {
+ for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
- consumerRecords.records(tp);
-
- if (recordsFromPartition.size() > 0) {
- final ConsumerRecord<byte[], byte[]> lastRecord =
- recordsFromPartition.get(recordsFromPartition.size() - 1);
-
- // After processing a record with offset of "stoppingOffset - 1", the split reader
- // should not continue fetching because the record with stoppingOffset may not
- // exist. Keep polling will just block forever.
- if (lastRecord.offset() >= stoppingOffset - 1) {
- recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
- finishSplitAtRecord(
- tp,
- stoppingOffset,
- lastRecord.offset(),
- finishedPartitions,
- recordsBySplits);
- }
+ long consumerPosition = consumer.position(tp);
+ // Stop fetching when the consumer's position reaches the stoppingOffset.
+ // Control messages may follow the last record; therefore, using the last record's
+ // offset as a stopping condition could result in indefinite blocking.
+ if (consumerPosition >= stoppingOffset) {
Review Comment:
As you mentioned, `consumerPosition >= stoppingOffset` is generally equivalent to `lastRecord.offset() >= stoppingOffset - 1`, but there are some edge cases where the new condition works better.
### Edge case
When reading `transactional batch data` using flink kafka connector.
#### Producer setup
Let’s say we are producing transactional messages, and the `lastRecord's offset` is `100`.
After producing the lastRecord, we commit the transaction, which writes a `marker message` at offset `101`.
Now, the `nextRecord's offset` becomes `102` (one more than the last message including marker message).
#### Consumer setup
When the flink consumer is reading with `scan.bounded.mode` set to `latest-offset`, the `stoppingOffset` is set to the `nextRecord's offset`, which in this case is `102`.
#### Comparison
In the old condition `(lastRecord.offset() >= stoppingOffset - 1)`, it would compare `100 >= 102 - 1`, which would never be met, leading to indefinite blocking.
However, with the new condition `(consumerPosition >= stoppingOffset)`, it compares `102 >= 102`, which is immediately satisfied.
This happens because consumer.position() skips the marker message.
Using lastRecord.offset() in the stopping condition is insufficient in transactional case (when a marker message follows the last record).
| Condition | As-Is (Old Condition) | To-Be (New Condition) |
|-----------------|---------------------------------------------------|---------------------------------------------------|
| Expression | `lastRecord.offset() >= stoppingOffset - 1` | `consumerPosition >= stoppingOffset` |
| Example Comparison | Compares `100 >= 102 - 1`, which never meets | Compares `102 >= 102`, which is immediately satisfied |
| Reason | The condition fails because it checks the last data record's offset, not accounting for the marker message | The condition succeeds because `consumer.position()` skips the marker message and correctly evaluates the stopping condition |
| Issue | Leads to indefinite polling | Stops polling immediately
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1759154224
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java:
##########
@@ -369,6 +370,38 @@ public void testConsumingTopicWithEmptyPartitions() throws Throwable {
WatermarkStrategy.noWatermarks(),
"testConsumingTopicWithEmptyPartitions"));
}
+
+ @Test
+ @Timeout(value = 10)
+ public void testConsumingTransactionalMessage() throws Throwable {
+ String transactionalTopic = "transactionalTopic-" + UUID.randomUUID();
+ KafkaSourceTestEnv.createTestTopic(
+ transactionalTopic, KafkaSourceTestEnv.NUM_PARTITIONS, 1);
+ List<ProducerRecord<String, Integer>> records =
+ KafkaSourceTestEnv.getRecordsForTopic(transactionalTopic);
+ // Prepare records for executeAndVerify method
+ records.removeIf(record -> record.partition() > record.value());
+
+ KafkaSourceTestEnv.produceTransactionalMessageToKafka(records);
+
+ KafkaSource<PartitionAndValue> source =
+ KafkaSource.<PartitionAndValue>builder()
+ .setBootstrapServers(KafkaSourceTestEnv.brokerConnectionStrings)
+ .setTopics(transactionalTopic)
+ .setGroupId("topic-with-transactional-message-test")
+ .setDeserializer(new TestingKafkaRecordDeserializationSchema(false))
+ .setStartingOffsets(OffsetsInitializer.earliest())
+ .setBounded(OffsetsInitializer.latest())
+ .build();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ executeAndVerify(
+ env,
+ env.fromSource(
+ source,
+ WatermarkStrategy.noWatermarks(),
+ "testConsumingTransactionalMessage"));
Review Comment:
Yes this test fails with the current main branch due to timeout.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1760607528
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java:
##########
@@ -369,6 +370,38 @@ public void testConsumingTopicWithEmptyPartitions() throws Throwable {
WatermarkStrategy.noWatermarks(),
"testConsumingTopicWithEmptyPartitions"));
}
+
+ @Test
+ @Timeout(value = 10)
+ public void testConsumingTransactionalMessage() throws Throwable {
+ String transactionalTopic = "transactionalTopic-" + UUID.randomUUID();
+ KafkaSourceTestEnv.createTestTopic(
+ transactionalTopic, KafkaSourceTestEnv.NUM_PARTITIONS, 1);
+ List<ProducerRecord<String, Integer>> records =
+ KafkaSourceTestEnv.getRecordsForTopic(transactionalTopic);
+ // Prepare records for executeAndVerify method
+ records.removeIf(record -> record.partition() > record.value());
Review Comment:
Could you change the comment to reflect how the data looks afterwards? I'm assuming this will preserve one record per partition?
I'm also curios why we need a specific data layout for this test to work? Or rephrased: what would happen if we retain all records originally generated? Wouldn't the test still assert similar things?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1760650597
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -122,32 +122,32 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
KafkaPartitionSplitRecords recordsBySplits =
new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
List<TopicPartition> finishedPartitions = new ArrayList<>();
- for (TopicPartition tp : consumerRecords.partitions()) {
+ for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
- consumerRecords.records(tp);
-
- if (recordsFromPartition.size() > 0) {
- final ConsumerRecord<byte[], byte[]> lastRecord =
- recordsFromPartition.get(recordsFromPartition.size() - 1);
-
- // After processing a record with offset of "stoppingOffset - 1", the split reader
- // should not continue fetching because the record with stoppingOffset may not
- // exist. Keep polling will just block forever.
- if (lastRecord.offset() >= stoppingOffset - 1) {
- recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
- finishSplitAtRecord(
- tp,
- stoppingOffset,
- lastRecord.offset(),
- finishedPartitions,
- recordsBySplits);
- }
+ long consumerPosition = consumer.position(tp);
Review Comment:
(This was just remark for other viewers of this PR).
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#issuecomment-2354458757
> Seems like the deleted arch unit rule was still needed. What was your intent when you deleted it?
It was automatically deleted after running `mvn clean verify` locally. I manually rolled back the archunit changes.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1759158825
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -122,32 +122,32 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
KafkaPartitionSplitRecords recordsBySplits =
new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
List<TopicPartition> finishedPartitions = new ArrayList<>();
- for (TopicPartition tp : consumerRecords.partitions()) {
+ for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
- consumerRecords.records(tp);
-
- if (recordsFromPartition.size() > 0) {
- final ConsumerRecord<byte[], byte[]> lastRecord =
- recordsFromPartition.get(recordsFromPartition.size() - 1);
-
- // After processing a record with offset of "stoppingOffset - 1", the split reader
- // should not continue fetching because the record with stoppingOffset may not
- // exist. Keep polling will just block forever.
- if (lastRecord.offset() >= stoppingOffset - 1) {
- recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
- finishSplitAtRecord(
- tp,
- stoppingOffset,
- lastRecord.offset(),
- finishedPartitions,
- recordsBySplits);
- }
+ long consumerPosition = consumer.position(tp);
+ // Stop fetching when the consumer's position reaches the stoppingOffset.
+ // Control messages may follow the last record; therefore, using the last record's
+ // offset as a stopping condition could result in indefinite blocking.
Review Comment:
It is inclusive `(consumerPosition >= stoppingOffset)`. I can't understand about discarding issue, could you elaborate more?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1759150772
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -122,32 +122,32 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
KafkaPartitionSplitRecords recordsBySplits =
new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
List<TopicPartition> finishedPartitions = new ArrayList<>();
- for (TopicPartition tp : consumerRecords.partitions()) {
+ for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
- consumerRecords.records(tp);
-
- if (recordsFromPartition.size() > 0) {
- final ConsumerRecord<byte[], byte[]> lastRecord =
- recordsFromPartition.get(recordsFromPartition.size() - 1);
-
- // After processing a record with offset of "stoppingOffset - 1", the split reader
- // should not continue fetching because the record with stoppingOffset may not
- // exist. Keep polling will just block forever.
- if (lastRecord.offset() >= stoppingOffset - 1) {
- recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
- finishSplitAtRecord(
- tp,
- stoppingOffset,
- lastRecord.offset(),
- finishedPartitions,
- recordsBySplits);
- }
+ long consumerPosition = consumer.position(tp);
+ // Stop fetching when the consumer's position reaches the stoppingOffset.
+ // Control messages may follow the last record; therefore, using the last record's
+ // offset as a stopping condition could result in indefinite blocking.
+ if (consumerPosition >= stoppingOffset) {
Review Comment:
As you mentioned, `consumerPosition >= stoppingOffset` is generally equivalent to `lastRecord.offset() >= stoppingOffset - 1`, but there are some edge cases where the new condition works better.
### Edge case
#### Producer setup
Let’s say we are producing transactional messages, and the `lastRecord's offset` is `100`.
After producing the lastRecord, we commit the transaction, which writes a `marker message` at offset `101`.
Now, the `nextRecord's offset` becomes `102` (one more than the last message including marker message).
#### Consumer setup
When the flink consumer is reading with `scan.bounded.mode` set to `latest-offset`, the `stoppingOffset` is set to the `nextRecord's offset`, which in this case is `102`.
#### Comparison
In the old condition `(lastRecord.offset() >= stoppingOffset - 1)`, it would compare `100 >= 102 - 1`, which would never be met, leading to indefinite blocking.
However, with the new condition `(consumerPosition >= stoppingOffset)`, it compares `102 >= 102`, which is immediately satisfied.
This happens because consumer.position() skips the marker message.
Using lastRecord.offset() in the stopping condition is insufficient in transactional case (when a marker message follows the last record).
| Condition | As-Is (Old Condition) | To-Be (New Condition) |
|-----------------|---------------------------------------------------|---------------------------------------------------|
| Expression | `lastRecord.offset() >= stoppingOffset - 1` | `consumerPosition >= stoppingOffset` |
| Example Comparison | Compares `100 >= 102 - 1`, which never meets | Compares `102 >= 102`, which is immediately satisfied |
| Reason | The condition fails because it checks the last data record's offset, not accounting for the marker message | The condition succeeds because `consumer.position()` skips the marker message and correctly evaluates the stopping condition |
| Issue | Leads to indefinite polling | Stops polling immediately
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#issuecomment-2355026088
Changes all LGTM. I'm running CI and merge when it's green.
Thank you very much for this much needed fix!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1760653304
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -122,32 +122,32 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
KafkaPartitionSplitRecords recordsBySplits =
new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
List<TopicPartition> finishedPartitions = new ArrayList<>();
- for (TopicPartition tp : consumerRecords.partitions()) {
+ for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
- consumerRecords.records(tp);
-
- if (recordsFromPartition.size() > 0) {
- final ConsumerRecord<byte[], byte[]> lastRecord =
- recordsFromPartition.get(recordsFromPartition.size() - 1);
-
- // After processing a record with offset of "stoppingOffset - 1", the split reader
- // should not continue fetching because the record with stoppingOffset may not
- // exist. Keep polling will just block forever.
- if (lastRecord.offset() >= stoppingOffset - 1) {
- recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
- finishSplitAtRecord(
- tp,
- stoppingOffset,
- lastRecord.offset(),
- finishedPartitions,
- recordsBySplits);
- }
+ long consumerPosition = consumer.position(tp);
+ // Stop fetching when the consumer's position reaches the stoppingOffset.
+ // Control messages may follow the last record; therefore, using the last record's
+ // offset as a stopping condition could result in indefinite blocking.
Review Comment:
I think I understand now:
* If I use stopping offset `endOffset` and it picks let's say 100, it would abort reading at 100 and sets the stopping offset of the partition to 100.
* The record 100 has been read and is part of the batch. It will not be filtered out.
* So it indeed appears as the last record of the batch.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1760649683
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java:
##########
@@ -369,6 +370,38 @@ public void testConsumingTopicWithEmptyPartitions() throws Throwable {
WatermarkStrategy.noWatermarks(),
"testConsumingTopicWithEmptyPartitions"));
}
+
+ @Test
+ @Timeout(value = 10)
Review Comment:
The global timeout is applied during CI only. So after X minutes, the CI build will fail. If you test it locally, you have to abort manually (or you use the scripts to run the build as CI does).
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1762220365
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java:
##########
@@ -188,6 +190,66 @@ public void testKafkaSourceSink() throws Exception {
deleteTestTopic(topic);
}
+ @Test
+ public void testKafkaSourceEmptyResultOnDeletedOffsets() throws Exception {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "bounded_" + format + "_" + UUID.randomUUID();
+ createTestTopic(topic, 1, 1);
+ // ---------- Produce an event time stream into Kafka -------------------
+ String groupId = getStandardProps().getProperty("group.id");
+ String bootstraps = getBootstrapServers();
+
+ final String createTable =
+ String.format(
+ "CREATE TABLE kafka (\n"
+ + " `user_id` INT,\n"
+ + " `item_id` INT,\n"
+ + " `behavior` STRING\n"
+ + ") WITH (\n"
+ + " 'connector' = '%s',\n"
+ + " 'topic' = '%s',\n"
+ + " 'properties.bootstrap.servers' = '%s',\n"
+ + " 'properties.group.id' = '%s',\n"
+ + " 'scan.startup.mode' = 'specific-offsets',\n"
+ + " 'scan.bounded.mode' = 'specific-offsets',\n"
+ + " 'scan.startup.specific-offsets' = 'partition:0,offset:1',\n"
+ + " 'scan.bounded.specific-offsets' = 'partition:0,offset:3',\n"
+ + " %s\n"
+ + ")\n",
+ KafkaDynamicTableFactory.IDENTIFIER,
+ topic,
+ bootstraps,
+ groupId,
+ formatOptions());
+ tEnv.executeSql(createTable);
+ List<Row> values =
+ Arrays.asList(
+ Row.of(1, 1102, "behavior 1"),
+ Row.of(2, 1103, "behavior 2"),
+ Row.of(3, 1104, "behavior 3"));
+ tEnv.fromValues(values).insertInto("kafka").execute().await();
+ // ---------- Delete events from Kafka -------------------
+ Map<Integer, Long> partitionOffsetsToDelete = new HashMap<>();
+ partitionOffsetsToDelete.put(0, 3L);
+ deleteRecords(topic, partitionOffsetsToDelete);
+ // ---------- Consume stream from Kafka -------------------
+ List<Row> results = new ArrayList<>();
+ assertTimeoutPreemptively(
Review Comment:
I've deleted timeout logic thanks for the feedback
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1601015487
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -122,27 +122,21 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
KafkaPartitionSplitRecords recordsBySplits =
new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
List<TopicPartition> finishedPartitions = new ArrayList<>();
- for (TopicPartition tp : consumerRecords.partitions()) {
+ for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
- consumerRecords.records(tp);
-
- if (recordsFromPartition.size() > 0) {
- final ConsumerRecord<byte[], byte[]> lastRecord =
- recordsFromPartition.get(recordsFromPartition.size() - 1);
-
- // After processing a record with offset of "stoppingOffset - 1", the split reader
- // should not continue fetching because the record with stoppingOffset may not
- // exist. Keep polling will just block forever.
- if (lastRecord.offset() >= stoppingOffset - 1) {
- recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
- finishSplitAtRecord(
- tp,
- stoppingOffset,
- lastRecord.offset(),
- finishedPartitions,
- recordsBySplits);
- }
+ long consumerPosition = consumer.position(tp);
+ // Stop fetching when the consumer's position reaches the stoppingOffset.
+ // Control messages may follow the last record; therefore, using the last record's
+ // offset as a stopping condition could result in indefinite blocking.
+ if (consumerPosition >= stoppingOffset) {
+ LOG.debug(
+ "Position of {}: {}, has reached stopping offset: {}",
+ tp,
+ consumerPosition,
+ stoppingOffset);
+ recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
+ finishSplitAtRecord(
+ tp, stoppingOffset, consumerPosition, finishedPartitions, recordsBySplits);
}
// Track this partition's record lag if it never appears before
kafkaSourceReaderMetrics.maybeAddRecordsLagMetric(consumer, tp);
Review Comment:
@LinMingQiang, thanks for the review.
I've changed to track tp, only when there is record for that tp.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "morazow (via GitHub)" <gi...@apache.org>.
morazow commented on PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#issuecomment-2213814301
Thanks @dongwoo6kim, Looks good!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "morazow (via GitHub)" <gi...@apache.org>.
morazow commented on PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#issuecomment-2159296755
Thanks @dongwoo6kim ,
Tests looks good from my side 👍
(_Recently I faced similar issue which maybe related, when running batch mode with setting `startingOffsets`. The change should solve that issue. But we may create issue for it_)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1759153861
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java:
##########
@@ -369,6 +370,38 @@ public void testConsumingTopicWithEmptyPartitions() throws Throwable {
WatermarkStrategy.noWatermarks(),
"testConsumingTopicWithEmptyPartitions"));
}
+
+ @Test
+ @Timeout(value = 10)
Review Comment:
I used the Timeout to prevent infinite blocking and fail the test if it enters an infinite loop.
Since I'm not familiar with the project's global timeout — could you elaborate more about it?
Where is it configured?
If global timeout is set then it should be enough but when I tested this method on the current main branch without the Timeout, it didn't fail but just hung.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java:
##########
@@ -369,6 +370,38 @@ public void testConsumingTopicWithEmptyPartitions() throws Throwable {
WatermarkStrategy.noWatermarks(),
"testConsumingTopicWithEmptyPartitions"));
}
+
+ @Test
+ @Timeout(value = 10)
+ public void testConsumingTransactionalMessage() throws Throwable {
+ String transactionalTopic = "transactionalTopic-" + UUID.randomUUID();
+ KafkaSourceTestEnv.createTestTopic(
+ transactionalTopic, KafkaSourceTestEnv.NUM_PARTITIONS, 1);
+ List<ProducerRecord<String, Integer>> records =
+ KafkaSourceTestEnv.getRecordsForTopic(transactionalTopic);
+ // Prepare records for executeAndVerify method
+ records.removeIf(record -> record.partition() > record.value());
Review Comment:
This is preprocessing step before using [executeAndVerify](https://github.com/apache/flink-connector-kafka/blob/7929b16dcfe648da30b6cc9755f63de2ed3d5319/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java#L482) method to verify test result.
This method expects records from `partition P` should be an integer sequence from `P` to `NUM_RECORDS_PER_PARTITION`. So I deleted records where the value is less than the partition number.
Similar approach to [here](https://github.com/apache/flink-connector-kafka/blob/7929b16dcfe648da30b6cc9755f63de2ed3d5319/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java#L348)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1759155830
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java:
##########
@@ -188,6 +190,66 @@ public void testKafkaSourceSink() throws Exception {
deleteTestTopic(topic);
}
+ @Test
+ public void testKafkaSourceEmptyResultOnDeletedOffsets() throws Exception {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "bounded_" + format + "_" + UUID.randomUUID();
+ createTestTopic(topic, 1, 1);
+ // ---------- Produce an event time stream into Kafka -------------------
+ String groupId = getStandardProps().getProperty("group.id");
+ String bootstraps = getBootstrapServers();
+
+ final String createTable =
+ String.format(
+ "CREATE TABLE kafka (\n"
+ + " `user_id` INT,\n"
+ + " `item_id` INT,\n"
+ + " `behavior` STRING\n"
+ + ") WITH (\n"
+ + " 'connector' = '%s',\n"
+ + " 'topic' = '%s',\n"
+ + " 'properties.bootstrap.servers' = '%s',\n"
+ + " 'properties.group.id' = '%s',\n"
+ + " 'scan.startup.mode' = 'specific-offsets',\n"
+ + " 'scan.bounded.mode' = 'specific-offsets',\n"
+ + " 'scan.startup.specific-offsets' = 'partition:0,offset:1',\n"
+ + " 'scan.bounded.specific-offsets' = 'partition:0,offset:3',\n"
+ + " %s\n"
+ + ")\n",
+ KafkaDynamicTableFactory.IDENTIFIER,
+ topic,
+ bootstraps,
+ groupId,
+ formatOptions());
+ tEnv.executeSql(createTable);
+ List<Row> values =
+ Arrays.asList(
+ Row.of(1, 1102, "behavior 1"),
+ Row.of(2, 1103, "behavior 2"),
+ Row.of(3, 1104, "behavior 3"));
+ tEnv.fromValues(values).insertInto("kafka").execute().await();
+ // ---------- Delete events from Kafka -------------------
+ Map<Integer, Long> partitionOffsetsToDelete = new HashMap<>();
+ partitionOffsetsToDelete.put(0, 3L);
+ deleteRecords(topic, partitionOffsetsToDelete);
Review Comment:
Yes by deleting message in this step we can cause hanging.
This test code is for issue [here](https://issues.apache.org/jira/browse/FLINK-35565)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1759158825
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -122,32 +122,32 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
KafkaPartitionSplitRecords recordsBySplits =
new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
List<TopicPartition> finishedPartitions = new ArrayList<>();
- for (TopicPartition tp : consumerRecords.partitions()) {
+ for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
- consumerRecords.records(tp);
-
- if (recordsFromPartition.size() > 0) {
- final ConsumerRecord<byte[], byte[]> lastRecord =
- recordsFromPartition.get(recordsFromPartition.size() - 1);
-
- // After processing a record with offset of "stoppingOffset - 1", the split reader
- // should not continue fetching because the record with stoppingOffset may not
- // exist. Keep polling will just block forever.
- if (lastRecord.offset() >= stoppingOffset - 1) {
- recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
- finishSplitAtRecord(
- tp,
- stoppingOffset,
- lastRecord.offset(),
- finishedPartitions,
- recordsBySplits);
- }
+ long consumerPosition = consumer.position(tp);
+ // Stop fetching when the consumer's position reaches the stoppingOffset.
+ // Control messages may follow the last record; therefore, using the last record's
+ // offset as a stopping condition could result in indefinite blocking.
Review Comment:
It is inclusive `(consumerPosition >= stoppingOffset)`. I can't understand about discarding issue, could you elaborate more?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#issuecomment-2349366111
@AHeise thanks for the feedback!
I've addressed your comments and applied the suggested changes. When you have a moment, please take a look. Thanks!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#issuecomment-2354460790
@AHeise Thanks for the feedback. I've left some comments and made updates. Please have a look.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "LinMingQiang (via GitHub)" <gi...@apache.org>.
LinMingQiang commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1600878564
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -122,27 +122,21 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
KafkaPartitionSplitRecords recordsBySplits =
new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
List<TopicPartition> finishedPartitions = new ArrayList<>();
- for (TopicPartition tp : consumerRecords.partitions()) {
+ for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
- consumerRecords.records(tp);
-
- if (recordsFromPartition.size() > 0) {
- final ConsumerRecord<byte[], byte[]> lastRecord =
- recordsFromPartition.get(recordsFromPartition.size() - 1);
-
- // After processing a record with offset of "stoppingOffset - 1", the split reader
- // should not continue fetching because the record with stoppingOffset may not
- // exist. Keep polling will just block forever.
- if (lastRecord.offset() >= stoppingOffset - 1) {
- recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
- finishSplitAtRecord(
- tp,
- stoppingOffset,
- lastRecord.offset(),
- finishedPartitions,
- recordsBySplits);
- }
+ long consumerPosition = consumer.position(tp);
+ // Stop fetching when the consumer's position reaches the stoppingOffset.
+ // Control messages may follow the last record; therefore, using the last record's
+ // offset as a stopping condition could result in indefinite blocking.
+ if (consumerPosition >= stoppingOffset) {
+ LOG.debug(
+ "Position of {}: {}, has reached stopping offset: {}",
+ tp,
+ consumerPosition,
+ stoppingOffset);
+ recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
+ finishSplitAtRecord(
+ tp, stoppingOffset, consumerPosition, finishedPartitions, recordsBySplits);
}
// Track this partition's record lag if it never appears before
kafkaSourceReaderMetrics.maybeAddRecordsLagMetric(consumer, tp);
Review Comment:
consumerRecords.partitions().forEach(trackTp -> {
// Track this partition's record lag if it never appears before
kafkaSourceReaderMetrics.maybeAddRecordsLagMetric(consumer, trackTp);
});
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34470][Connectors/Kafka] Fix indefinite blocking by adjusting stopping condition in split reader [flink-connector-kafka]
Posted by "dongwoo6kim (via GitHub)" <gi...@apache.org>.
dongwoo6kim commented on code in PR #100:
URL: https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1601015487
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -122,27 +122,21 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
KafkaPartitionSplitRecords recordsBySplits =
new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
List<TopicPartition> finishedPartitions = new ArrayList<>();
- for (TopicPartition tp : consumerRecords.partitions()) {
+ for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
- consumerRecords.records(tp);
-
- if (recordsFromPartition.size() > 0) {
- final ConsumerRecord<byte[], byte[]> lastRecord =
- recordsFromPartition.get(recordsFromPartition.size() - 1);
-
- // After processing a record with offset of "stoppingOffset - 1", the split reader
- // should not continue fetching because the record with stoppingOffset may not
- // exist. Keep polling will just block forever.
- if (lastRecord.offset() >= stoppingOffset - 1) {
- recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
- finishSplitAtRecord(
- tp,
- stoppingOffset,
- lastRecord.offset(),
- finishedPartitions,
- recordsBySplits);
- }
+ long consumerPosition = consumer.position(tp);
+ // Stop fetching when the consumer's position reaches the stoppingOffset.
+ // Control messages may follow the last record; therefore, using the last record's
+ // offset as a stopping condition could result in indefinite blocking.
+ if (consumerPosition >= stoppingOffset) {
+ LOG.debug(
+ "Position of {}: {}, has reached stopping offset: {}",
+ tp,
+ consumerPosition,
+ stoppingOffset);
+ recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
+ finishSplitAtRecord(
+ tp, stoppingOffset, consumerPosition, finishedPartitions, recordsBySplits);
}
// Track this partition's record lag if it never appears before
kafkaSourceReaderMetrics.maybeAddRecordsLagMetric(consumer, tp);
Review Comment:
@LinMingQiang, thanks for the review.
I've changed to track tp, only when records for that tp is not empty.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org