You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Yu Wang (Jira)" <ji...@apache.org> on 2024/08/06 07:49:00 UTC
[jira] [Assigned] (KAFKA-17139) MirrorSourceTask will stop mirroring when get BufferUnderflowException
[ https://issues.apache.org/jira/browse/KAFKA-17139?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Yu Wang reassigned KAFKA-17139:
-------------------------------
Assignee: Yu Wang
> MirrorSourceTask will stop mirroring when get BufferUnderflowException
> ----------------------------------------------------------------------
>
> Key: KAFKA-17139
> URL: https://issues.apache.org/jira/browse/KAFKA-17139
> Project: Kafka
> Issue Type: Bug
> Components: connect, mirrormaker
> Affects Versions: 3.0.0, 3.5.2, 3.6.2, 3.7.1
> Reporter: Yu Wang
> Assignee: Yu Wang
> Priority: Major
> Attachments: image-2024-07-15-15-35-12-489.png
>
>
> Recently we found the data mirroring of one of our partition stopped after got the following exception
> {code:java}
> [2024-07-05 13:36:07,058] WARN Failure during poll. (org.apache.kafka.connect.mirror.MirrorSourceTask)org.apache.kafka.common.protocol.types.SchemaException: Buffer underflow while parsing response for request with header RequestHeader(apiKey=FETCH, apiVersion=11, clientId=consumer-null-8, correlationId=-855959214) at org.apache.kafka.clients.NetworkClient.parseResponse(NetworkClient.java:722) at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:865) at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:560) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:265) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:236) at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1297) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1238) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211) at org.apache.kafka.connect.mirror.MirrorSourceTask.poll(MirrorSourceTask.java:141) at org.apache.kafka.connect.runtime.WorkerSourceTask.poll(WorkerSourceTask.java:291) at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:248) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:186) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:241) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:750)Caused by: java.nio.BufferUnderflowException at java.nio.Buffer.nextGetIndex(Buffer.java:510) at java.nio.HeapByteBuffer.getLong(HeapByteBuffer.java:427) at org.apache.kafka.common.protocol.ByteBufferAccessor.readLong(ByteBufferAccessor.java:48) at org.apache.kafka.common.message.FetchResponseData$AbortedTransaction.read(FetchResponseData.java:1928) at org.apache.kafka.common.message.FetchResponseData$AbortedTransaction.<init>(FetchResponseData.java:1904) at org.apache.kafka.common.message.FetchResponseData$PartitionData.read(FetchResponseData.java:881) at org.apache.kafka.common.message.FetchResponseData$PartitionData.<init>(FetchResponseData.java:805) at org.apache.kafka.common.message.FetchResponseData$FetchableTopicResponse.read(FetchResponseData.java:524) at org.apache.kafka.common.message.FetchResponseData$FetchableTopicResponse.<init>(FetchResponseData.java:464) at org.apache.kafka.common.message.FetchResponseData.read(FetchResponseData.java:199) at org.apache.kafka.common.message.FetchResponseData.<init>(FetchResponseData.java:136) at org.apache.kafka.common.requests.FetchResponse.parse(FetchResponse.java:119) at org.apache.kafka.common.requests.AbstractResponse.parseResponse(AbstractResponse.java:117) at org.apache.kafka.common.requests.AbstractResponse.parseResponse(AbstractResponse.java:109) at org.apache.kafka.clients.NetworkClient.parseResponse(NetworkClient.java:720) ... 17 more {code}
> The exception only thrown once, then the consumer stopped to fetrch from the node, the request rate to one of the Kafka broker dropped to 0 !image-2024-07-15-15-35-12-489.png|width=601,height=233!
>
> After going through the code of KafkaConsumer, every time KafkaConsumer tries to generate the fetch request to Kafka brokers, it will check if the target broker exists in {*}nodesWithPendingFetchRequests{*}. If it exists, then skip the target kafka broker in this round.
> [https://github.com/apache/kafka/blob/3.7.1/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java#L433]
> The broker id can be removed only when the response completed.
> [https://github.com/apache/kafka/blob/3.7.1/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java#L600]
> But in this case, the exception was thrown at *handleCompletedReceives,* which means the node id will never be removed from the *nodesWithPendingFetchRequests.*
> [https://github.com/apache/kafka/blob/3.7.1/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java#L594]
>
> {code:java}
> handleCompletedSends(responses, updatedNow);
> handleCompletedReceives(responses, updatedNow);
> handleDisconnections(responses, updatedNow);
> handleConnections();
> handleInitiateApiVersionRequests(updatedNow);
> handleTimedOutConnections(responses, updatedNow);
> handleTimedOutRequests(responses, updatedNow);
> completeResponses(responses); {code}
> So every time, when KafkaConnect source task try to call the *poll* method of the MirrorSourceTask, the KafkaConsumer will skip fetch from the node id that left in the *nodesWithPendingFetchRequests.*
> This will make the MirrorMaker tasks stop the data mirroring with only 1 WARN log.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)