Bug: Message acknowledgment succeeds without receiving ACK_RESPONSE

[danielbackes]

Description

The consumer currently sends the ACK command to the broker but does not wait for or validate the corresponding ACK_RESPONSE.

As a result, the client may consider a message successfully acknowledged even when the broker did not process the acknowledgment.

Reproduction scenario

1. Receive a message from Pulsar.
2. Stop the Pulsar broker gracefully before acknowledging the message.
3. Call the consumer acknowledgment method.
4. The client writes the ACK command but does not verify that the broker returned an ACK_RESPONSE.
5. The acknowledgment method may complete successfully even though the message was not acknowledged.

In my test, waiting for a response after sending the command exposed broker responses such as CLOSE_CONSUMER.

Actual behavior

The client only writes the acknowledgment command to the socket. A successful socket write confirms that the bytes were accepted by the local socket, but it does not confirm that the Pulsar broker received or processed the acknowledgment.

This can cause the application to log or treat a message as acknowledged while Pulsar still considers it unacknowledged, resulting in message redelivery.

Expected behavior

After sending an ACK command, the client should:

1. Include a request ID in the command.
2. Wait for the corresponding ACK_RESPONSE.
3. Queue any MESSAGE commands received while waiting, so they can be processed later.
4. Handle unrelated commands, including responding to PING and ignoring PONG.
5. Handle CLOSE_CONSUMER appropriately.
6. Match the ACK_RESPONSE using its request ID.
7. Throw an exception if:
   • the broker rejects the acknowledgment;
   • the connection is closed;
   • no matching response is received before the client’s configured timeout.

The acknowledgment method should only return successfully after the matching ACK_RESPONSE has been received and validated.

[danielbackes]

I reproduced another consequence of ack() returning without waiting for confirmation from the broker.

In my test, the broker’s keepalive timeout was 10 seconds, and two messages had already been delivered to the consumer. The first handler took 30 seconds. During that time, the broker sent a PING and closed the connection because the client did not respond with a PONG.

The client then wrote an ACK for the first message and continued processing the second message. Neither ACK raised an error, but a successful socket write does not confirm that the broker received or processed it. After the buffered messages were consumed, the client read the pending PING and attempted to send a PONG. That write finally raised write data of close socket.

Enabling reconnection does not solve this problem; it can obscure it. The client recreates the connection and may receive the same messages again because their ACKs were not confirmed, causing their handlers to run twice. The earlier ack() calls still appeared successful to the application.

The error can therefore appear after all delivered messages have been processed, or be masked by reconnection. ack() needs to wait for its matching ACK_RESPONSE while the client continues handling other incoming frames, including MESSAGE and PING.