Details
-
Bug
-
Status: Resolved
-
Major
-
Resolution: Fixed
-
0.9.0.1
-
None
-
None
-
None
Description
I foud that KafkaProducer may send duplicated message sometimes, which is happend when:
In Sender thread:
NetworkClient::poll()
-> this.selector.poll() //send the message, such as "abc", and send it to broker successfully
-> handleTimedOutRequests(responses,updatedNow); //Judge whether the message "abc" which is sent above is expired or timeout, and the judge is based on the parameter this.requestTimeoutMs and updatedNow;
-> response.request().callback().onComplete()
-> completeBatch(batch,Errors.NETWORK_EXCEPTION,-1L,correlationId,now); //If themessage was judged as expired, then it will be reenqueued and send repeatly next loop;
-> this.accumulator.reenqueue(batch,now);
The problem comes out: If the message "abc" is sent successfully to broker, but it may be judged to expired, so the message will be sent repeately next loop, which make the message duplicated.
I can reproduce this scenario normally.
In my opinion, I think Send::handleTimedOutRequests() is not much useful, because the response of sending request from broker is succesfully and has no error, which means brokers persist it successfully. And this function will induce to the duplicated message problems.