[jira] [Created] (KAFKA-17139) MirrorSourceTask will stop mirroring when get BufferUnderflowException

["Yu Wang (Jira)" <ji...@apache.org>]

Yu Wang created KAFKA-17139:
-------------------------------

             Summary: 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.7.1, 3.6.2, 3.5.2, 3.0.0
            Reporter: Yu Wang
         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.RheosHaMirrorSourceTask)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.RheosHaMirrorSourceTask.poll(RheosHaMirrorSourceTask.java:130)        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)