Efrat19 commented on code in PR #290:
URL:
https://github.com/apache/flink-connector-kafka/pull/290#discussion_r3751430369
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -66,13 +66,28 @@ public class KafkaPartitionSplitReader
private static final Logger LOG =
LoggerFactory.getLogger(KafkaPartitionSplitReader.class);
private static final long POLL_TIMEOUT = 10000L;
- private final KafkaConsumer<byte[], byte[]> consumer;
+ private final Properties consumerProps;
private final Map<TopicPartition, Long> stoppingOffsets;
private final String groupId;
private final int subtaskId;
private final KafkaSourceReaderMetrics kafkaSourceReaderMetrics;
+ /** Guards lazy creation of {@link #consumer} and the {@link
#pendingWakeup} flag. */
+ private final Object consumerLock = new Object();
+
+ /**
+ * 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). Access outside {@link #ensureConsumer()} /
{@link #wakeUp()} is
+ * only safe after a call to {@link #ensureConsumer()} on the same thread.
+ */
+ @Nullable private volatile KafkaConsumer<byte[], byte[]> consumer;
Review Comment:
`@GuardedBy("lock")`
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -94,17 +109,56 @@ public KafkaPartitionSplitReader(
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);
+ }
+
+ /**
+ * 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() {
+ KafkaConsumer<byte[], byte[]> currentConsumer = this.consumer;
+ if (currentConsumer != null) {
+ return currentConsumer;
+ }
+ synchronized (consumerLock) {
+ currentConsumer = this.consumer;
+ if (currentConsumer == null) {
+ currentConsumer = createConsumer(consumerProps);
+ maybeRegisterKafkaConsumerMetrics(
+ consumerProps, kafkaSourceReaderMetrics,
currentConsumer);
+ kafkaSourceReaderMetrics.registerNumBytesIn(currentConsumer);
+ if (pendingWakeup) {
+ // A wakeUp() arrived before the consumer existed. Apply
it now so that the
+ // first blocking call still observes it, matching the
behavior of a wakeup
+ // against an eagerly-created consumer.
+ currentConsumer.wakeup();
+ pendingWakeup = false;
+ }
+ this.consumer = currentConsumer;
+ }
+ return currentConsumer;
+ }
+ }
- // Metric registration
- maybeRegisterKafkaConsumerMetrics(props, kafkaSourceReaderMetrics,
consumer);
- this.kafkaSourceReaderMetrics.registerNumBytesIn(consumer);
+ /** Creates the {@link KafkaConsumer}. Overridable for tests to observe
the creation. */
+ @VisibleForTesting
+ protected KafkaConsumer<byte[], byte[]> createConsumer(Properties
consumerProps) {
Review Comment:
It feels a bit weird to have the `ensureConsumer` wrapper private and
`createConsumer` inner call `protected`, wdyt?
If this is just for the tests you could probably call consumer.poll with a
long timeout and expect it to throw a wakeUpException
Or at least make package-private
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -66,13 +66,28 @@ public class KafkaPartitionSplitReader
private static final Logger LOG =
LoggerFactory.getLogger(KafkaPartitionSplitReader.class);
private static final long POLL_TIMEOUT = 10000L;
- private final KafkaConsumer<byte[], byte[]> consumer;
+ private final Properties consumerProps;
private final Map<TopicPartition, Long> stoppingOffsets;
private final String groupId;
private final int subtaskId;
private final KafkaSourceReaderMetrics kafkaSourceReaderMetrics;
+ /** Guards lazy creation of {@link #consumer} and the {@link
#pendingWakeup} flag. */
+ private final Object consumerLock = new Object();
+
+ /**
+ * 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). Access outside {@link #ensureConsumer()} /
{@link #wakeUp()} is
+ * only safe after a call to {@link #ensureConsumer()} on the same thread.
+ */
+ @Nullable private volatile KafkaConsumer<byte[], byte[]> consumer;
+
+ /** Set when {@link #wakeUp()} is called before the consumer exists;
guarded by the lock. */
+ private boolean pendingWakeup;
Review Comment:
`@GuardedBy("lock")`
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderConsumerThreadTest.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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;
+
+/**
+ * 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 KafkaPartitionSplitReaderConsumerThreadTest {
Review Comment:
Nit:
```suggestion
class KafkaPartitionSplitReaderLazyConsumerCreationTest {
```
or even move to `KafkaPartitionSplitReaderTest`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
