Zhuoxi2000 commented on code in PR #290:
URL:
https://github.com/apache/flink-connector-kafka/pull/290#discussion_r4056325153
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -295,7 +355,7 @@ void setConsumerClientRack(Properties consumerProps, String
rackId) {
}
long getConsumerPosition(TopicPartition tp, String msg) {
- return retryOnWakeup(() -> consumer.position(tp), msg);
+ return retryOnWakeup(() -> ensureConsumer().position(tp), msg);
Review Comment:
Done the helpers now take the consumer directly, and `ensureConsumer()` is
only used at the SplitReader entry points and in the test accessor.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java:
##########
@@ -94,17 +104,49 @@ 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() {
+ // 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);
+ maybeRegisterKafkaConsumerMetrics(
+ consumerProps, kafkaSourceReaderMetrics, currentConsumer);
+ kafkaSourceReaderMetrics.registerNumBytesIn(currentConsumer);
+ this.consumer = currentConsumer;
Review Comment:
Done the field is assigned before metric registration now, with a
regression test covering the failure case.
--
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]
