(flink-connector-kafka) branch main updated: [FLINK-39728] Fix pauseOrResumeSplits race with unassigned partitions

[arvid]

This is an automated email from the ASF dual-hosted git repository.

AHeise pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git


The following commit(s) were added to refs/heads/main by this push:
     new eb8ff700 [FLINK-39728] Fix pauseOrResumeSplits race with unassigned 
partitions
eb8ff700 is described below

commit eb8ff700b2390d5bc3a8c67fcdb3f3ad93824e24
Author: Jim Hughes <jhug...@confluent.io>
AuthorDate: Thu May 21 14:57:37 2026 -0400

    [FLINK-39728] Fix pauseOrResumeSplits race with unassigned partitions
    
    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)
---
 .../source/reader/KafkaPartitionSplitReader.java   | 31 ++++++++++++++++------
 .../reader/KafkaPartitionSplitReaderTest.java      | 23 ++++++++++++++++
 2 files changed, 46 insertions(+), 8 deletions(-)

diff --git 
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
 
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
index 74d4bc1f..256a9cf1 100644
--- 
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
+++ 
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
@@ -240,14 +240,29 @@ public class KafkaPartitionSplitReader
     public void pauseOrResumeSplits(
             Collection<KafkaPartitionSplit> splitsToPause,
             Collection<KafkaPartitionSplit> splitsToResume) {
-        consumer.resume(
-                splitsToResume.stream()
-                        .map(KafkaPartitionSplit::getTopicPartition)
-                        .collect(Collectors.toList()));
-        consumer.pause(
-                splitsToPause.stream()
-                        .map(KafkaPartitionSplit::getTopicPartition)
-                        .collect(Collectors.toList()));
+        // Filter against current assignment to avoid IllegalStateException 
when a partition
+        // was concurrently unassigned by fetch() or removeEmptySplits().
+        Set<TopicPartition> assigned = consumer.assignment();
+        List<TopicPartition> toResume = new ArrayList<>();
+        for (KafkaPartitionSplit split : splitsToResume) {
+            TopicPartition tp = split.getTopicPartition();
+            if (assigned.contains(tp)) {
+                toResume.add(tp);
+            } else {
+                LOG.warn("Skipping resume for unassigned partition {}", tp);
+            }
+        }
+        List<TopicPartition> toPause = new ArrayList<>();
+        for (KafkaPartitionSplit split : splitsToPause) {
+            TopicPartition tp = split.getTopicPartition();
+            if (assigned.contains(tp)) {
+                toPause.add(tp);
+            } else {
+                LOG.warn("Skipping pause for unassigned partition {}", tp);
+            }
+        }
+        consumer.resume(toResume);
+        consumer.pause(toPause);
     }
 
     // ---------------
diff --git 
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java
 
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java
index af000438..adf240df 100644
--- 
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java
+++ 
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java
@@ -354,6 +354,29 @@ public class KafkaPartitionSplitReaderTest {
         
assertThat(properties.containsKey(ConsumerConfig.CLIENT_RACK_CONFIG)).isFalse();
     }
 
+    @Test
+    void testPauseOrResumeSplitsWithUnassignedPartition() {
+        KafkaPartitionSplitReader reader = createReader();
+        // Create a split for a partition that is NOT assigned to the consumer.
+        // Without the fix, this would throw IllegalStateException:
+        // "No current assignment for partition".
+        TopicPartition unassignedPartition = new TopicPartition(TOPIC1, 0);
+        KafkaPartitionSplit unassignedSplit =
+                new KafkaPartitionSplit(unassignedPartition, 
KafkaPartitionSplit.EARLIEST_OFFSET);
+
+        // Verify the partition is indeed not assigned
+        
assertThat(reader.consumer().assignment()).doesNotContain(unassignedPartition);
+
+        // This should be a no-op, not throw an exception
+        reader.pauseOrResumeSplits(
+                Collections.singletonList(unassignedSplit), 
Collections.emptyList());
+        reader.pauseOrResumeSplits(
+                Collections.emptyList(), 
Collections.singletonList(unassignedSplit));
+        reader.pauseOrResumeSplits(
+                Collections.singletonList(unassignedSplit),
+                Collections.singletonList(unassignedSplit));
+    }
+
     // ------------------
 
     private void assignSplitsAndFetchUntilFinish(KafkaPartitionSplitReader 
reader, int readerId)