You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "jnh5y (via GitHub)" <gi...@apache.org> on 2026/05/21 19:00:10 UTC
[PR] [FLINK-39728] Fix pauseOrResumeSplits race with unassigned partitions [flink-connector-kafka]
jnh5y opened a new pull request, #257:
URL: https://github.com/apache/flink-connector-kafka/pull/257
Filter against consumer.assignment() before calling pause()/resume() to prevent IllegalStateException when a partition is concurrently unassigned by fetch() or removeEmptySplits().
Generated-by: Claude Opus 4.6 (Anthropic)
--
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-39728] Fix pauseOrResumeSplits race with unassigned partitions [flink-connector-kafka]
Posted by "jnh5y (via GitHub)" <gi...@apache.org>.
jnh5y commented on code in PR #257:
URL: https://github.com/apache/flink-connector-kafka/pull/257#discussion_r3452373264
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -240,13 +240,18 @@ public void close() throws Exception {
public void pauseOrResumeSplits(
Collection<KafkaPartitionSplit> splitsToPause,
Collection<KafkaPartitionSplit> splitsToResume) {
+ // Filter against current assignment to avoid IllegalStateException when a partition
+ // was concurrently unassigned by fetch() or removeEmptySplits().
+ Set<TopicPartition> assigned = consumer.assignment();
Review Comment:
I am not a committer; I don't see a button to merge 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-39728] Fix pauseOrResumeSplits race with unassigned partitions [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise merged PR #257:
URL: https://github.com/apache/flink-connector-kafka/pull/257
--
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-39728] Fix pauseOrResumeSplits race with unassigned partitions [flink-connector-kafka]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #257:
URL: https://github.com/apache/flink-connector-kafka/pull/257#issuecomment-4832729120
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-39728] Fix pauseOrResumeSplits race with unassigned partitions [flink-connector-kafka]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #257:
URL: https://github.com/apache/flink-connector-kafka/pull/257#issuecomment-4511850123
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-39728] Fix pauseOrResumeSplits race with unassigned partitions [flink-connector-kafka]
Posted by "jnh5y (via GitHub)" <gi...@apache.org>.
jnh5y commented on code in PR #257:
URL: https://github.com/apache/flink-connector-kafka/pull/257#discussion_r3318343529
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -240,13 +240,18 @@ public void close() throws Exception {
public void pauseOrResumeSplits(
Collection<KafkaPartitionSplit> splitsToPause,
Collection<KafkaPartitionSplit> splitsToResume) {
+ // Filter against current assignment to avoid IllegalStateException when a partition
+ // was concurrently unassigned by fetch() or removeEmptySplits().
+ Set<TopicPartition> assigned = consumer.assignment();
Review Comment:
Added!
--
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-39728] Fix pauseOrResumeSplits race with unassigned partitions [flink-connector-kafka]
Posted by "Efrat19 (via GitHub)" <gi...@apache.org>.
Efrat19 commented on code in PR #257:
URL: https://github.com/apache/flink-connector-kafka/pull/257#discussion_r3448689390
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -240,13 +240,18 @@ public void close() throws Exception {
public void pauseOrResumeSplits(
Collection<KafkaPartitionSplit> splitsToPause,
Collection<KafkaPartitionSplit> splitsToResume) {
+ // Filter against current assignment to avoid IllegalStateException when a partition
+ // was concurrently unassigned by fetch() or removeEmptySplits().
+ Set<TopicPartition> assigned = consumer.assignment();
Review Comment:
Nope it looks good, do you require anything else to merge 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-39728] Fix pauseOrResumeSplits race with unassigned partitions [flink-connector-kafka]
Posted by "Efrat19 (via GitHub)" <gi...@apache.org>.
Efrat19 commented on code in PR #257:
URL: https://github.com/apache/flink-connector-kafka/pull/257#discussion_r3329766432
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -240,13 +240,18 @@ public void close() throws Exception {
public void pauseOrResumeSplits(
Collection<KafkaPartitionSplit> splitsToPause,
Collection<KafkaPartitionSplit> splitsToResume) {
+ // Filter against current assignment to avoid IllegalStateException when a partition
+ // was concurrently unassigned by fetch() or removeEmptySplits().
+ Set<TopicPartition> assigned = consumer.assignment();
Review Comment:
Thanks
I'd give up on the filter to avoid the 2X iterations
`.forEach(tp -> assigned.contains(tp) ? .. : ..` might be simpler wdyt
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -240,13 +240,18 @@ public void close() throws Exception {
public void pauseOrResumeSplits(
Collection<KafkaPartitionSplit> splitsToPause,
Collection<KafkaPartitionSplit> splitsToResume) {
+ // Filter against current assignment to avoid IllegalStateException when a partition
+ // was concurrently unassigned by fetch() or removeEmptySplits().
+ Set<TopicPartition> assigned = consumer.assignment();
Review Comment:
Thanks
I'd give up on the filter to avoid the 2X iterations
`.forEach(tp -> assigned.contains(tp) ? .. : ..)` might be simpler wdyt
--
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-39728] Fix pauseOrResumeSplits race with unassigned partitions [flink-connector-kafka]
Posted by "jnh5y (via GitHub)" <gi...@apache.org>.
jnh5y commented on code in PR #257:
URL: https://github.com/apache/flink-connector-kafka/pull/257#discussion_r3334758263
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -240,13 +240,18 @@ public void close() throws Exception {
public void pauseOrResumeSplits(
Collection<KafkaPartitionSplit> splitsToPause,
Collection<KafkaPartitionSplit> splitsToResume) {
+ // Filter against current assignment to avoid IllegalStateException when a partition
+ // was concurrently unassigned by fetch() or removeEmptySplits().
+ Set<TopicPartition> assigned = consumer.assignment();
Review Comment:
Ok, how does this look? https://github.com/apache/flink-connector-kafka/compare/5a53da774d83ce6ddccdf04ebbdbf015c58e9d40..12701fa48b54693fda5d5bdb7034aadcfba321a1
I think there are a lot of ways this could look; happy to go with something else if you prefer.
--
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-39728] Fix pauseOrResumeSplits race with unassigned partitions [flink-connector-kafka]
Posted by "Efrat19 (via GitHub)" <gi...@apache.org>.
Efrat19 commented on code in PR #257:
URL: https://github.com/apache/flink-connector-kafka/pull/257#discussion_r3294114149
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -240,13 +240,18 @@ public void close() throws Exception {
public void pauseOrResumeSplits(
Collection<KafkaPartitionSplit> splitsToPause,
Collection<KafkaPartitionSplit> splitsToResume) {
+ // Filter against current assignment to avoid IllegalStateException when a partition
+ // was concurrently unassigned by fetch() or removeEmptySplits().
+ Set<TopicPartition> assigned = consumer.assignment();
Review Comment:
Nit: Wdyt about leaving a warn here to unmask any edge case where pauseOrResumeSplits is called for unassigned partition from an unfinished split?
--
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