[
https://issues.apache.org/jira/browse/FLINK-36434?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18101536#comment-18101536
]
Edson Wang commented on FLINK-36434:
------------------------------------
Thanks for supporting! Here’s the mini design I’m planning to fold into the PR
description.
for now I think the point is the Kafka consumer is created too early:
{{source / checkpoint thread
-> createSplitFetcher()
-> splitReaderFactory.get()
-> new KafkaPartitionSplitReader()
-> new KafkaConsumer()
fetcher thread
-> fetch()
-> handleSplitsChanges()
-> commit offsets
-> close()}}
So the {{KafkaConsumer}} can be created on one thread but mostly used on the
fetcher thread.
what i proposed Kafka-side fix like
{{source / checkpoint thread
-> new KafkaPartitionSplitReader()
stores consumer props only
fetcher thread
-> first fetch / split change / commit
-> ensureConsumer()
-> new KafkaConsumer()}}
which keeps the change contained in {{{}flink-connector-kafka{}}}: make the
consumer lazy inside {{{}KafkaPartitionSplitReader{}}}, create it on first
fetcher-thread use, and keep {{wakeUp()}} safe before creation by recording a
pending wakeup.
I agree that deferring {{splitReaderFactory.get()}} in {{flink-connector-base}}
would be the broader fix for all connectors, but I’d treat that as orthogonal
and keep this PR focused on Kafka unless you prefer otherwise.
> Revise threading model of (KafkaPartition)SplitReader
> -----------------------------------------------------
>
> Key: FLINK-36434
> URL: https://issues.apache.org/jira/browse/FLINK-36434
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: kafka-3.2.0
> Reporter: Arvid Heise
> Assignee: Edson Wang
> Priority: Major
>
> The KafkaPartitionSplitReader is created in the source thread, where it
> initializes the consumer. However, it later access the consumer almost
> exclusively through the fetcher thread. Since the consumer is not
> thread-safe, this thread model looks broken.
> However, I'd challenge that the overall SplitReader implementation is already
> suboptimal as the same issue is probably happening in other connectors. I'd
> probably first create the fetch task and within the fetch task create the
> split reader.
> If left as-is, we can't upgrade Kafka client anymore because we receive
> sporadic
> {code:java}
> Caused by: org.apache.kafka.common.requests.CorrelationIdMismatchException:
> Correlation id for response (1179651) does not match request (0), request
> header: RequestHeader(apiKey=API_VERSIONS, apiVersion=3,
> clientId=kafka-source-external-context-6092797646400842179-3,
> correlationId=0, headerVersion=2)
> at
> org.apache.kafka.common.requests.AbstractResponse.parseResponse(AbstractResponse.java:106)
> at
> org.apache.kafka.clients.NetworkClient.parseResponse(NetworkClient.java:740)
> at
> org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:913)
> at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:580)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:280)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:251)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:242)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.position(KafkaConsumer.java:1728)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.position(KafkaConsumer.java:1686)
> at
> org.apache.flink.connector.kafka.source.reader.KafkaPartitionSplitReader.lambda$removeEmptySplits$5(KafkaPartitionSplitReader.java:375)
> at
> org.apache.flink.connector.kafka.source.reader.KafkaPartitionSplitReader.retryOnWakeup(KafkaPartitionSplitReader.java:481)
> at
> org.apache.flink.connector.kafka.source.reader.KafkaPartitionSplitReader.removeEmptySplits(KafkaPartitionSplitReader.java:374)
> at
> org.apache.flink.connector.kafka.source.reader.KafkaPartitionSplitReader.handleSplitsChanges(KafkaPartitionSplitReader.java:224)
> at
> org.apache.flink.connector.base.source.reader.fetcher.AddSplitsTask.run(AddSplitsTask.java:51)
> at
> org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:165)
> ... 6 more
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
