MartijnVisser commented on code in PR #290:
URL:
https://github.com/apache/flink-connector-kafka/pull/290#discussion_r4054244363
##########
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:
Assigning the field straight after `createConsumer` would be a bit safer I
think: if metric registration throws, `close()` sees null and never closes the
consumer.
##########
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:
These helpers only run after `fetch()` or `handleSplitsChanges()` already
ensured the consumer, and `fetch()` reaches this one per assigned partition per
poll rather than once. Passing the consumer down keeps creation to the four
entry points.
--
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]
