This is an automated email from the ASF dual-hosted git repository.
MartijnVisser 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 9c0ccf3b [FLINK-36434][Connectors/Kafka] Create the KafkaConsumer
lazily on the split fetcher thread
9c0ccf3b is described below
commit 9c0ccf3b1b52ccd98ccd9cd714330a522b19850b
Author: Edson <[email protected]>
AuthorDate: Sun Sep 20 02:01:40 2026 -0400
[FLINK-36434][Connectors/Kafka] Create the KafkaConsumer lazily on the
split fetcher thread
Create the consumer on the first consumer-touching call instead of in the
constructor, which by construction happens on the fetcher thread.
Generated-by: Claude Code (Opus 5)
---
.../source/reader/KafkaPartitionSplitReader.java | 118 ++++++++++--
...rtitionSplitReaderLazyConsumerCreationTest.java | 211 +++++++++++++++++++++
2 files changed, 308 insertions(+), 21 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 bf8b0a97..b87ce1ce 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
@@ -65,7 +65,7 @@ public class KafkaPartitionSplitReader
implements SplitReader<ConsumerRecord<byte[], byte[]>,
KafkaPartitionSplit> {
private static final Logger LOG =
LoggerFactory.getLogger(KafkaPartitionSplitReader.class);
- private final KafkaConsumer<byte[], byte[]> consumer;
+ private final Properties consumerProps;
private final Map<TopicPartition, Long> stoppingOffsets;
private final String groupId;
private final int subtaskId;
@@ -73,6 +73,16 @@ public class KafkaPartitionSplitReader
private final KafkaSourceReaderMetrics kafkaSourceReaderMetrics;
+ /**
+ * The Kafka consumer. Created lazily on the first thread that actually
uses it — the split
+ * fetcher thread — rather than on the thread constructing this reader
(the source-reader or
+ * checkpoint thread). {@link KafkaConsumer} is not thread-safe, so it
must live on the thread
+ * that uses it (FLINK-36434). Only ever written by {@link
#ensureConsumer()} on the fetcher
+ * thread; {@code volatile} solely for the cross-thread reads in {@link
#wakeUp()} and {@link
+ * #close()}.
+ */
+ @Nullable private volatile KafkaConsumer<byte[], byte[]> consumer;
+
// Tracking empty splits that has not been added to finished splits in
fetch()
private final Set<String> emptySplits = new HashSet<>();
@@ -90,23 +100,58 @@ public class KafkaPartitionSplitReader
String rackIdSupplier) {
this.subtaskId = context.getIndexOfSubtask();
this.kafkaSourceReaderMetrics = kafkaSourceReaderMetrics;
- // Parsed before creating the consumer so that an invalid value does
not leak a consumer.
+ // Parsed eagerly so that an invalid value fails at construction
rather than on the
+ // fetcher thread when the consumer is first used.
this.pollTimeout = parsePollTimeout(props);
Properties consumerProps = new Properties();
consumerProps.putAll(props);
consumerProps.setProperty(ConsumerConfig.CLIENT_ID_CONFIG,
createConsumerClientId(props));
setConsumerClientRack(consumerProps, rackIdSupplier);
- this.consumer = new KafkaConsumer<>(consumerProps);
+ this.consumerProps = consumerProps;
this.stoppingOffsets = new HashMap<>();
this.groupId =
consumerProps.getProperty(ConsumerConfig.GROUP_ID_CONFIG);
+ }
- // Metric registration
- maybeRegisterKafkaConsumerMetrics(props, kafkaSourceReaderMetrics,
consumer);
- this.kafkaSourceReaderMetrics.registerNumBytesIn(consumer);
+ /**
+ * Returns the consumer, creating it on the calling thread on first use.
+ *
+ * <p>The reader is constructed on the source-reader (or, when a fetcher
is re-created for an
+ * offset commit, the checkpoint) thread, while the consumer is used
almost exclusively on the
+ * split fetcher thread. Creating the consumer eagerly in the constructor
therefore puts it on
+ * the wrong thread. All consumer-touching {@link SplitReader} methods run
on the fetcher
+ * thread, so deferring creation to the first such call keeps construction
and use on the same
+ * thread. The only cross-thread entry point remains {@link #wakeUp()},
which is the one call
+ * {@link KafkaConsumer} documents as thread-safe.
+ */
+ private KafkaConsumer<byte[], byte[]> ensureConsumer() {
+ // Single-writer: only the fetcher thread calls this, so the creation
branch needs no
+ // guard; the field is volatile solely for the cross-thread reads in
wakeUp() and close().
+ KafkaConsumer<byte[], byte[]> currentConsumer = this.consumer;
+ if (currentConsumer == null) {
+ currentConsumer = createConsumer(consumerProps);
+ // Published before metric registration: if registration throws,
close() still sees
+ // the consumer and releases it instead of leaking it.
+ this.consumer = currentConsumer;
+ maybeRegisterKafkaConsumerMetrics(
+ consumerProps, kafkaSourceReaderMetrics, currentConsumer);
+ kafkaSourceReaderMetrics.registerNumBytesIn(currentConsumer);
+ }
+ return currentConsumer;
+ }
+
+ /**
+ * Creates the {@link KafkaConsumer}. Overridable by same-package tests to
observe creation.
+ *
+ * <p>Deliberately not annotated with {@code @VisibleForTesting}: the
freezing ArchUnit rule
+ * flags new uses of the internal annotation in connector production code.
+ */
+ KafkaConsumer<byte[], byte[]> createConsumer(Properties consumerProps) {
+ return new KafkaConsumer<>(consumerProps);
}
@Override
public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws
IOException {
+ final KafkaConsumer<byte[], byte[]> consumer = ensureConsumer();
ConsumerRecords<byte[], byte[]> consumerRecords;
try {
consumerRecords = consumer.poll(pollTimeout);
@@ -126,7 +171,8 @@ public class KafkaPartitionSplitReader
List<TopicPartition> finishedPartitions = new ArrayList<>();
for (TopicPartition tp : consumer.assignment()) {
long stoppingOffset = getStoppingOffset(tp);
- long consumerPosition = getConsumerPosition(tp, "retrieving
consumer position");
+ long consumerPosition =
+ getConsumerPosition(consumer, tp, "retrieving consumer
position");
// 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.
@@ -155,7 +201,7 @@ public class KafkaPartitionSplitReader
// Unassign the partitions that has finished.
if (!finishedPartitions.isEmpty()) {
finishedPartitions.forEach(kafkaSourceReaderMetrics::removeRecordsLagMetric);
- unassignPartitions(finishedPartitions);
+ unassignPartitions(consumer, finishedPartitions);
}
// Update numBytesIn
@@ -182,6 +228,7 @@ public class KafkaPartitionSplitReader
"The SplitChange type of %s is not supported.",
splitsChange.getClass()));
}
+ final KafkaConsumer<byte[], byte[]> consumer = ensureConsumer();
// Assignment.
List<TopicPartition> newPartitionAssignments = new ArrayList<>();
@@ -216,32 +263,51 @@ public class KafkaPartitionSplitReader
// Seek on the newly assigned partitions to their stating offsets.
seekToStartingOffsets(
+ consumer,
partitionsStartingFromEarliest,
partitionsStartingFromLatest,
partitionsStartingFromSpecifiedOffsets);
// Setup the stopping offsets.
- acquireAndSetStoppingOffsets(partitionsStoppingAtLatest,
partitionsStoppingAtCommitted);
+ acquireAndSetStoppingOffsets(
+ consumer, partitionsStoppingAtLatest,
partitionsStoppingAtCommitted);
// After acquiring the starting and stopping offsets, remove the empty
splits if necessary.
- removeEmptySplits();
+ removeEmptySplits(consumer);
- maybeLogSplitChangesHandlingResult(splitsChange);
+ maybeLogSplitChangesHandlingResult(consumer, splitsChange);
}
@Override
public void wakeUp() {
- consumer.wakeup();
+ // The only method called from a thread other than the fetcher thread,
relying on
+ // KafkaConsumer.wakeup() being the one documented thread-safe
consumer call.
+ //
+ // wakeUp() only ever targets a running fetch task (see
SplitReader#wakeUp javadoc), which
+ // the base SplitFetcher can only schedule after handleSplitsChanges()
— and therefore
+ // ensureConsumer() — has run at least once on the fetcher thread. So
the consumer is
+ // always non-null here in practice; a pre-creation wakeup has nothing
to interrupt and is
+ // safely dropped. Note this leans on an implementation invariant of
+ // SplitFetcher/AddSplitsTask rather than a documented contract beyond
wakeUp()'s own
+ // javadoc scoping.
+ KafkaConsumer<byte[], byte[]> currentConsumer = this.consumer;
+ if (currentConsumer != null) {
+ currentConsumer.wakeup();
+ }
}
@Override
public void close() throws Exception {
- consumer.close();
+ KafkaConsumer<byte[], byte[]> currentConsumer = this.consumer;
+ if (currentConsumer != null) {
+ currentConsumer.close();
+ }
}
@Override
public void pauseOrResumeSplits(
Collection<KafkaPartitionSplit> splitsToPause,
Collection<KafkaPartitionSplit> splitsToResume) {
+ final KafkaConsumer<byte[], byte[]> consumer = ensureConsumer();
// Filter against current assignment to avoid IllegalStateException
when a partition
// was concurrently unassigned by fetch() or removeEmptySplits().
Set<TopicPartition> assigned = consumer.assignment();
@@ -272,12 +338,12 @@ public class KafkaPartitionSplitReader
public void notifyCheckpointComplete(
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit,
OffsetCommitCallback offsetCommitCallback) {
- consumer.commitAsync(offsetsToCommit, offsetCommitCallback);
+ ensureConsumer().commitAsync(offsetsToCommit, offsetCommitCallback);
}
@VisibleForTesting
KafkaConsumer<byte[], byte[]> consumer() {
- return consumer;
+ return ensureConsumer();
}
@VisibleForTesting
@@ -316,7 +382,10 @@ public class KafkaPartitionSplitReader
}
}
- long getConsumerPosition(TopicPartition tp, String msg) {
+ // The helpers below take the consumer their entry point (fetch() /
handleSplitsChanges())
+ // already ensured, so consumer creation stays confined to the SplitReader
entry points.
+ private long getConsumerPosition(
+ KafkaConsumer<byte[], byte[]> consumer, TopicPartition tp, String
msg) {
return retryOnWakeup(() -> consumer.position(tp), msg);
}
@@ -364,6 +433,7 @@ public class KafkaPartitionSplitReader
}
private void seekToStartingOffsets(
+ KafkaConsumer<byte[], byte[]> consumer,
List<TopicPartition> partitionsStartingFromEarliest,
List<TopicPartition> partitionsStartingFromLatest,
Map<TopicPartition, Long> partitionsStartingFromSpecifiedOffsets) {
@@ -387,6 +457,7 @@ public class KafkaPartitionSplitReader
}
private void acquireAndSetStoppingOffsets(
+ KafkaConsumer<byte[], byte[]> consumer,
List<TopicPartition> partitionsStoppingAtLatest,
Set<TopicPartition> partitionsStoppingAtCommitted) {
Map<TopicPartition, Long> endOffset =
consumer.endOffsets(partitionsStoppingAtLatest);
@@ -408,12 +479,13 @@ public class KafkaPartitionSplitReader
}
}
- private void removeEmptySplits() {
+ private void removeEmptySplits(KafkaConsumer<byte[], byte[]> consumer) {
List<TopicPartition> emptyPartitions = new ArrayList<>();
// If none of the partitions have any records,
for (TopicPartition tp : consumer.assignment()) {
long startingOffset =
- getConsumerPosition(tp, "getting starting offset to check
if split is empty");
+ getConsumerPosition(
+ consumer, tp, "getting starting offset to check if
split is empty");
if (startingOffset >= getStoppingOffset(tp)) {
emptyPartitions.add(tp);
}
@@ -428,11 +500,12 @@ public class KafkaPartitionSplitReader
.map(KafkaPartitionSplit::toSplitId)
.collect(Collectors.toSet()));
// Un-assign partitions from Kafka consumer
- unassignPartitions(emptyPartitions);
+ unassignPartitions(consumer, emptyPartitions);
}
}
private void maybeLogSplitChangesHandlingResult(
+ KafkaConsumer<byte[], byte[]> consumer,
SplitsChange<KafkaPartitionSplit> splitsChange) {
if (LOG.isDebugEnabled()) {
StringJoiner splitsInfo = new StringJoiner(",");
@@ -443,7 +516,8 @@ public class KafkaPartitionSplitReader
}
long startingOffset =
- getConsumerPosition(split.getTopicPartition(),
"logging starting position");
+ getConsumerPosition(
+ consumer, split.getTopicPartition(), "logging
starting position");
long stoppingOffset =
getStoppingOffset(split.getTopicPartition());
splitsInfo.add(
@@ -455,7 +529,9 @@ public class KafkaPartitionSplitReader
}
}
- private void unassignPartitions(Collection<TopicPartition>
partitionsToUnassign) {
+ private void unassignPartitions(
+ KafkaConsumer<byte[], byte[]> consumer,
+ Collection<TopicPartition> partitionsToUnassign) {
Collection<TopicPartition> newAssignment = new
HashSet<>(consumer.assignment());
newAssignment.removeAll(partitionsToUnassign);
consumer.assign(newAssignment);
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderLazyConsumerCreationTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderLazyConsumerCreationTest.java
new file mode 100644
index 00000000..93d6649d
--- /dev/null
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderLazyConsumerCreationTest.java
@@ -0,0 +1,211 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.kafka.source.reader;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.kafka.source.KafkaSourceOptions;
+import
org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.junit.jupiter.api.Test;
+
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Unit tests for FLINK-36434: the {@link KafkaConsumer} inside {@link
KafkaPartitionSplitReader}
+ * must be created lazily on the thread that uses it (the split fetcher
thread), not on the thread
+ * that constructs the reader (the source-reader or checkpoint thread),
because the consumer is not
+ * thread-safe.
+ *
+ * <p>These tests need no running Kafka cluster: constructing a {@link
KafkaConsumer} performs no
+ * network I/O, and {@link KafkaPartitionSplitReader#fetch()} tolerates the
{@code
+ * IllegalStateException}/{@code WakeupException} that polling without an
assignment raises.
+ */
+class KafkaPartitionSplitReaderLazyConsumerCreationTest {
+
+ /** Records the thread the consumer was created on and every {@code
wakeup()} call. */
+ private static class CreationRecordingReader extends
KafkaPartitionSplitReader {
+ private final AtomicReference<Thread> creationThread = new
AtomicReference<>();
+ private final AtomicInteger wakeups = new AtomicInteger();
+
+ private final AtomicInteger closes = new AtomicInteger();
+
+ CreationRecordingReader() {
+ this(
+ new KafkaSourceReaderMetrics(
+
UnregisteredMetricsGroup.createSourceReaderMetricGroup()));
+ }
+
+ CreationRecordingReader(KafkaSourceReaderMetrics metrics) {
+ super(
+ testProperties(),
+ new TestingReaderContext(
+ new Configuration(),
+
UnregisteredMetricsGroup.createSourceReaderMetricGroup()),
+ metrics);
+ }
+
+ @Override
+ KafkaConsumer<byte[], byte[]> createConsumer(Properties consumerProps)
{
+ creationThread.set(Thread.currentThread());
+ return new KafkaConsumer<byte[], byte[]>(consumerProps) {
+ @Override
+ public void wakeup() {
+ wakeups.incrementAndGet();
+ super.wakeup();
+ }
+
+ @Override
+ public void close() {
+ closes.incrementAndGet();
+ super.close();
+ }
+ };
+ }
+ }
+
+ /** Metrics whose consumer registration fails, standing in for a broken
metric group. */
+ private static class ThrowingRegistrationMetrics extends
KafkaSourceReaderMetrics {
+ ThrowingRegistrationMetrics() {
+ super(UnregisteredMetricsGroup.createSourceReaderMetricGroup());
+ }
+
+ @Override
+ public void registerNumBytesIn(KafkaConsumer<?, ?> consumer) {
+ throw new IllegalStateException("metric registration failed");
+ }
+ }
+
+ private static Properties testProperties() {
+ Properties props = new Properties();
+ // Never contacted: the consumer performs no network I/O until the
first poll-style call.
+ props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:1");
+ props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
+ // Keep close() from attempting an offset auto-commit against the
unreachable broker.
+ props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+ props.setProperty(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+ ByteArrayDeserializer.class.getName());
+ props.setProperty(
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+ ByteArrayDeserializer.class.getName());
+ props.setProperty(KafkaSourceOptions.CLIENT_ID_PREFIX.key(),
"test-client");
+ return props;
+ }
+
+ @Test
+ void testConsumerIsCreatedLazilyOnTheUsingThread() throws Exception {
+ CreationRecordingReader reader = new CreationRecordingReader();
+
+ // Constructing the reader (source-reader / checkpoint thread) must
not create a consumer.
+ assertThat(reader.creationThread.get()).isNull();
+
+ // First use on a different thread (standing in for the split fetcher
thread).
+ Thread useThread =
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ reader.fetch();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .thenApply(ignored -> reader.creationThread.get())
+ .get();
+
+ assertThat(useThread).isNotNull();
+ assertThat(useThread).isNotSameAs(Thread.currentThread());
+ reader.close();
+ }
+
+ @Test
+ void testWakeUpBeforeConsumerCreationIsSafelyDropped() throws Exception {
+ CreationRecordingReader reader = new CreationRecordingReader();
+
+ // wakeUp() before the consumer exists: no NPE, no consumer created.
The base
+ // SplitFetcher only wakes a running fetch task, which cannot exist
before
+ // handleSplitsChanges() has created the consumer, so a pre-creation
wakeup has
+ // nothing to interrupt and is dropped rather than deferred.
+ assertThatCode(reader::wakeUp).doesNotThrowAnyException();
+ assertThat(reader.creationThread.get()).isNull();
+ assertThat(reader.wakeups.get()).isZero();
+
+ // First use creates the consumer; the dropped wakeup is NOT replayed
against it.
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ reader.fetch();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .get();
+
+ assertThat(reader.creationThread.get()).isNotNull();
+ assertThat(reader.wakeups.get()).isZero();
+ reader.close();
+ }
+
+ @Test
+ void testWakeUpAfterCreationDelegatesDirectly() throws Exception {
+ CreationRecordingReader reader = new CreationRecordingReader();
+ // Trigger creation on this thread via the test accessor.
+ reader.consumer();
+ assertThat(reader.wakeups.get()).isZero();
+
+ reader.wakeUp();
+ assertThat(reader.wakeups.get()).isEqualTo(1);
+ reader.close();
+ }
+
+ @Test
+ void testCloseWithoutUseNeverCreatesConsumer() {
+ CreationRecordingReader reader = new CreationRecordingReader();
+ assertThatCode(reader::close).doesNotThrowAnyException();
+ assertThat(reader.creationThread.get()).isNull();
+ }
+
+ @Test
+ void testConsumerIsReleasedWhenMetricRegistrationFails() throws Exception {
+ CreationRecordingReader reader =
+ new CreationRecordingReader(new ThrowingRegistrationMetrics());
+
+ // The consumer is published before metric registration, so the
failure surfaces to the
+ // caller but the created consumer is still owned by the reader...
+ assertThatThrownBy(reader::consumer)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("metric registration failed");
+ assertThat(reader.creationThread.get()).isNotNull();
+
+ // ...and close() releases it instead of leaking it.
+ reader.close();
+ assertThat(reader.closes.get()).isEqualTo(1);
+ }
+}
