[
https://issues.apache.org/jira/browse/FLINK-36434?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18100988#comment-18100988
]
Edson Wang commented on FLINK-36434:
------------------------------------
I’d like to work on this.
My rough approach would be to create the {{KafkaPartitionSplitReader}} /
{{KafkaConsumer}} lazily on the fetcher thread, for example through a supplier
invoked on the first {{{}fetch(){}}}. That way, the consumer is created on the
same thread that uses it, instead of on the source thread.
Before I start, do you think this can stay within
{{{}flink-connector-kafka{}}}, or would it need changes in
{{flink-connector-base}} / {{{}SplitFetcherManager{}}}? I’m happy to write up a
short design note first if that’s preferred.
For context, I run Flink + Kafka pipelines at billions-of-events/day scale in
production, so I can also help validate the change under realistic
consumer-lifecycle churn.
Could you assign this to me?
> 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
> 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)
