From 44bf0a8fdcb481c731bfcd6d593408958225c047 Mon Sep 17 00:00:00 2001 From: Daniel Backes Date: Tue, 8 Sep 2026 10:33:45 -0400 Subject: [PATCH 1/3] fix: wait for ACK_RESPONSE when acknowledging messages Refs #49 --- src/PartitionConsumer.php | 42 +++++++++++++++++++++++++++++++++++---- 1 file changed, 38 insertions(+), 4 deletions(-) diff --git a/src/PartitionConsumer.php b/src/PartitionConsumer.php index 96cae6e..d4fc0f2 100644 --- a/src/PartitionConsumer.php +++ b/src/PartitionConsumer.php @@ -12,10 +12,12 @@ use Pulsar\Exception\IOException; +use Pulsar\Exception\RuntimeException; use Pulsar\IO\AbstractIO; use Pulsar\Proto\BaseCommand\Type; use Pulsar\Proto\CommandAck; use Pulsar\Proto\CommandAck\AckType; +use Pulsar\Proto\CommandAckResponse; use Pulsar\Proto\CommandCloseConsumer; use Pulsar\Proto\CommandFlow; use Pulsar\Proto\CommandRedeliverUnacknowledgedMessages; @@ -155,19 +157,51 @@ public function flow() /** * @param Message $message - * @return void + * @return CommandAckResponse * @throws IOException + * @throws RuntimeException */ - public function ack(Message $message) + public function ack(Message $message): CommandAckResponse { - // send CommandAck + $requestId = Helper::getRequestID(); + $command = new CommandAck(); $command->setConsumerId($this->id); $command->setAckType(AckType::Individual()); $command->addMessageId($message->getMessageIdData()); $command->setTxnidLeastBits(null); $command->setTxnidMostBits(null); - $this->connection->writeCommand(Type::ACK(), $command); + $command->setRequestId($requestId); + + $response = $this->connection + ->writeCommand(Type::ACK(), $command) + ->wait() + ; + + $baseCommand = $response->getBaseCommand(); + + $commandType = $baseCommand->getType(); + + if (Type::CLOSE_CONSUMER_VALUE === $commandType->value()) { + throw new RuntimeException( + 'The consumer was closed before the message acknowledgment was confirmed.' + ); + } + + if (Type::ACK_RESPONSE_VALUE !== $commandType->value()) { + throw new RuntimeException(sprintf( + 'Unexpected Pulsar command type "%d" while waiting for the ACK response.', + $commandType->value() + )); + } + + $ackResponse = $baseCommand->getAckResponse(); + + if ($ackResponse->getRequestId() !== $requestId) { + throw new RuntimeException('ACK response request ID does not match.'); + } + + return $ackResponse; } From e22e958e2e8db9510b158d7d7ab334fdd98cce29 Mon Sep 17 00:00:00 2001 From: Daniel Backes Date: Wed, 16 Sep 2026 14:55:11 -0400 Subject: [PATCH 2/3] fix: wait for ackResponse in ack method and enqueue decoded messages Update the ack method to wait for the ackResponse before proceeding. Any messages decoded while waiting are safely added to the queue to prevent data loss. Refs #49 --- README-CN.md | 12 ++ README.md | 12 ++ src/Consumer.php | 223 ++++++++++++++++++++++++++++++-------- src/ConsumerOptions.php | 27 +++++ src/PartitionConsumer.php | 38 +------ 5 files changed, 234 insertions(+), 78 deletions(-) diff --git a/README-CN.md b/README-CN.md index 612b7ad..9cecc66 100644 --- a/README-CN.md +++ b/README-CN.md @@ -260,6 +260,17 @@ $options->setReconnectPolicy(true,3); $options->setReconnectPolicy(true,3,100); ``` +> Ack 超时时间 + +* `ack()` 会阻塞等待 broker 返回的 `ACK_RESPONSE`，通过 request ID 进行匹配 +* 如果在超时时间内没有收到匹配的响应，`ack()` 会抛出 `RuntimeException` +* 如果在等待响应期间发生重连，重连后重新发送的 ack 同样受此超时时间限制 +* 默认值为 30 秒 + +```php +$options->setAckTimeout(30); +``` + > 不循环接收消息，且平滑退出 ```php @@ -464,6 +475,7 @@ $reader->close(); * setDeadLetterPolicy() * setSubscriptionInitialPosition() * setReconnectPolicy() + * setAckTimeout() * setSchema() * ReaderOptions * setTopic() diff --git a/README.md b/README.md index 8c0e2bf..33ebb21 100644 --- a/README.md +++ b/README.md @@ -272,6 +272,17 @@ $options->setReconnectPolicy(true,3); $options->setReconnectPolicy(true,3,100); ``` +> Ack timeout + +* `ack()` blocks waiting for the broker's `ACK_RESPONSE`, matched to the request by request ID +* If no matching response arrives within this timeout, `ack()` throws a `RuntimeException` +* Also applied to each resend after a reconnect occurs while waiting for the response +* Default is 30 seconds + +```php +$options->setAckTimeout(30); +``` + > Not loop Receive And Smooth exit ```php @@ -478,6 +489,7 @@ $reader->close(); * setDeadLetterPolicy() * setSubscriptionInitialPosition() * setReconnectPolicy() + * setAckTimeout() * setSchema() * ReaderOptions * setTopic() diff --git a/src/Consumer.php b/src/Consumer.php index 435d185..2f89587 100644 --- a/src/Consumer.php +++ b/src/Consumer.php @@ -13,7 +13,11 @@ use Pulsar\Exception\MessageNotFound; use Pulsar\Exception\OptionsException; use Pulsar\Exception\RuntimeException; +use Pulsar\Proto\BaseCommand\Type; +use Pulsar\Proto\CommandAckResponse; use Pulsar\Proto\CommandMessage; +use Pulsar\Util\Buffer; +use Pulsar\Util\Helper; use Pulsar\Util\Packer; use SplPriorityQueue; use SplQueue; @@ -166,29 +170,11 @@ public function receive(bool $loop = true): Message return $this->messageQueue->dequeue(); } - try { - $response = $this->eventloop->wait($this->getWaitSeconds()); - } catch (IOException $e) { - $response = null; - - $policy = $this->options->getReconnectPolicy(); - // not enable reconnect - if (!$policy['status']) { - throw $e; - } - - if ($this->reconnect($policy)) { - return $this->receive($loop); - } - } - + $response = $this->pollFrame($this->getWaitSeconds()); // nack $this->executeInternalNack(); - - // ping - $this->ping(); - + if (is_null($response)) { if (!$loop) { throw new MessageNotFound(); @@ -209,28 +195,11 @@ public function receive(bool $loop = true): Message return $this->receive($loop); } - - $consumer = $this->getPartitionConsumer($commandMessage->getConsumerId()); - - /** - * @var $messages array - */ - $messages = Packer::decode($commandMessage, $response->getBuffer(), $consumer->getTopic()); - - foreach ($messages as $message) { - - // Save Options to Message Object - $message->setOptions($this->options); - - $this->messageQueue->enqueue($message); - } - - $consumer->decrement(sizeof($messages)); + $this->enqueueCommandMessage($commandMessage, $response->getBuffer()); return $this->messageQueue->dequeue(); } - /** * @return array * @throws IOException @@ -248,18 +217,83 @@ public function batchReceive(bool $loop = true): array } /** + * Sends the CommandAck and blocks for its CommandAckResponse, correlated by request ID. + * + * While waiting, any MESSAGE frame that arrives first (the broker may deliver one before + * the ACK_RESPONSE, since the connection is asynchronous) is queued rather than discarded. + * If the connection is already dead -- or drops while waiting -- sendAckRequest() reconnects + * and resends the CommandAck with a fresh request ID on the new connection before (re)waiting. + * * @param Message $message - * @return void - * @throws \Exception + * @return CommandAckResponse|null + * @throws IOException + * @throws RuntimeException */ - public function ack(Message $message) + public function ack(Message $message): ?CommandAckResponse { if (!$message->canAck()) { - return; + return null; } - // send CommandAck - $this->getPartitionConsumer($message->getConsumerID())->ack($message); + $requestId = $this->sendAckRequest($message); + + $deadline = microtime(true) + $this->options->getAckTimeout(); + + do { + $remaining = $deadline - microtime(true); + if ($remaining <= 0) { + throw new RuntimeException('Timed out waiting for ACK response.'); + } + + // Select::wait() forwards this to stream_select()'s tv_sec, which requires an int + $response = $this->pollFrame((int) ceil($remaining), function () use ($message, &$requestId) { + // the connection was rebuilt; the outstanding ACK request died with it + $requestId = $this->sendAckRequest($message); + }); + + if (null === $response) { + continue; + } + + $baseCommand = $response->getBaseCommand(); + + $commandType = $baseCommand->getType(); + + if (Type::CLOSE_CONSUMER_VALUE === $commandType->value()) { + // only abort if it's the consumer this message belongs to; the connection + // may be shared with other partition consumers that closed independently + if ($baseCommand->getCloseConsumer()->getConsumerId() === $message->getConsumerID()) { + throw new RuntimeException( + 'The consumer was closed before the message acknowledgment was confirmed.' + ); + } + continue; + } + + if (Type::MESSAGE_VALUE === $commandType->value()) { + $this->enqueueCommandMessage($baseCommand->getMessage(), $response->getBuffer()); + continue; + } + + if (Type::ACK_RESPONSE_VALUE === $commandType->value()) { + $ackResponse = $baseCommand->getAckResponse(); + + if ($ackResponse->getRequestId() !== $requestId) { + throw new RuntimeException('ACK response request ID does not match.'); + } + + if ($ackResponse->hasError()) { + $msg = $ackResponse->hasMessage() ? $ackResponse->getMessage() : $ackResponse->getError()->name(); + throw new RuntimeException( + sprintf('The broker rejected the acknowledgment: %s', $msg), + $ackResponse->getError()->value() + ); + } + + return $ackResponse; + } + + } while (true); } @@ -358,4 +392,105 @@ protected function getPartitionConsumer(int $consumerID): PartitionConsumer return $this->consumers[ $consumerID ]; } + /** + * Waits for a single frame from the event loop, applying the reconnect policy on + * connection loss and running the ping bookkeeping. + * + * On a dropped connection, once it's rebuilt, $onReconnect (if given) is invoked to + * resend whatever the caller had outstanding on the old connection -- a CommandAck, + * for instance, has no way to be resumed on a new one. Either way this returns null, + * same as a plain timeout, so the caller's own retry loop drives the next poll. + * + * @param int|float|null $timeoutSeconds + * @param callable|null $onReconnect called with no arguments right after a successful reconnect + * @return Response|null + * @throws IOException + */ + private function pollFrame($timeoutSeconds = null, ?callable $onReconnect = null): ?Response + { + try { + $response = $this->eventloop->wait($timeoutSeconds); + + $this->ping(); + + return $response; + } catch (IOException $e) { + $policy = $this->options->getReconnectPolicy(); + + // not enable reconnect + if (!$policy['status']) { + throw $e; + } + + $this->reconnect($policy); + + if ($onReconnect) { + $onReconnect(); + } + + return null; + } + } + + /** + * Sends a CommandAck for $message with a fresh request ID, applying the reconnect policy + * (and retrying the send on the rebuilt connection) if the connection is already dead. + * + * @param Message $message + * @return int the request ID the ack was sent with + * @throws IOException + */ + private function sendAckRequest(Message $message): int + { + $requestId = Helper::getRequestID(); + + try { + $this->getPartitionConsumer($message->getConsumerID())->ack($message, $requestId); + } catch (IOException $e) { + $policy = $this->options->getReconnectPolicy(); + + // not enable reconnect + if (!$policy['status']) { + throw $e; + } + + $this->reconnect($policy); + + return $this->sendAckRequest($message); + } + + return $requestId; + } + + /** + * Decodes a MESSAGE frame's payload into Message objects, queues them locally, and + * decrements the partition's available flow-control permits accordingly. + * + * Shared by receive() and ack()'s wait loop, since a MESSAGE frame can arrive while + * ack() is waiting on the same connection for an unrelated ACK_RESPONSE -- it must be + * queued here rather than discarded, or the message would be silently lost. + * + * @param CommandMessage $commandMessage + * @param Buffer $buffer + * @return void + */ + private function enqueueCommandMessage(CommandMessage $commandMessage, Buffer $buffer) + { + $consumer = $this->getPartitionConsumer($commandMessage->getConsumerId()); + + /** + * @var array $messages + */ + $messages = Packer::decode($commandMessage, $buffer, $consumer->getTopic()); + + foreach ($messages as $message) { + // Save Options to Message Object + $message->setOptions($this->options); + + $this->messageQueue->enqueue($message); + } + + $consumer->decrement(sizeof($messages)); + } + } diff --git a/src/ConsumerOptions.php b/src/ConsumerOptions.php index dee11f0..652ceca 100644 --- a/src/ConsumerOptions.php +++ b/src/ConsumerOptions.php @@ -70,6 +70,13 @@ final class ConsumerOptions extends Options */ const ENABLE_RECONNECT = 'reconnect'; + /** + * Max seconds to wait for an ACK_RESPONSE before giving up, default 30 seconds + * + * @var int + */ + const ACK_TIMEOUT = 'ack_timeout'; + /** * @param array $topics * @return void @@ -259,6 +266,26 @@ public function getNackRedeliveryDelay() } + /** + * @param int $seconds + * @return void + */ + public function setAckTimeout(int $seconds) + { + $this->data[ self::ACK_TIMEOUT ] = $seconds; + } + + + /** + * @return int|mixed + */ + public function getAckTimeout() + { + $timeout = $this->data[ self::ACK_TIMEOUT ] ?? 30; + return $timeout <= 0 ? 30 : $timeout; + } + + /** * @return mixed|string * @throws \Exception diff --git a/src/PartitionConsumer.php b/src/PartitionConsumer.php index d4fc0f2..bd43596 100644 --- a/src/PartitionConsumer.php +++ b/src/PartitionConsumer.php @@ -157,14 +157,13 @@ public function flow() /** * @param Message $message + * @param int $requestId * @return CommandAckResponse * @throws IOException - * @throws RuntimeException */ - public function ack(Message $message): CommandAckResponse + public function ack(Message $message, int $requestId) { - $requestId = Helper::getRequestID(); - + // send CommandAck $command = new CommandAck(); $command->setConsumerId($this->id); $command->setAckType(AckType::Individual()); @@ -172,36 +171,7 @@ public function ack(Message $message): CommandAckResponse $command->setTxnidLeastBits(null); $command->setTxnidMostBits(null); $command->setRequestId($requestId); - - $response = $this->connection - ->writeCommand(Type::ACK(), $command) - ->wait() - ; - - $baseCommand = $response->getBaseCommand(); - - $commandType = $baseCommand->getType(); - - if (Type::CLOSE_CONSUMER_VALUE === $commandType->value()) { - throw new RuntimeException( - 'The consumer was closed before the message acknowledgment was confirmed.' - ); - } - - if (Type::ACK_RESPONSE_VALUE !== $commandType->value()) { - throw new RuntimeException(sprintf( - 'Unexpected Pulsar command type "%d" while waiting for the ACK response.', - $commandType->value() - )); - } - - $ackResponse = $baseCommand->getAckResponse(); - - if ($ackResponse->getRequestId() !== $requestId) { - throw new RuntimeException('ACK response request ID does not match.'); - } - - return $ackResponse; + $this->connection->writeCommand(Type::ACK(), $command); } From 0c1440d6dfbbb2c47374828038338c451e544e48 Mon Sep 17 00:00:00 2001 From: Daniel Backes Date: Wed, 16 Sep 2026 20:03:40 -0400 Subject: [PATCH 3/3] Remove reconnect-on-disconnect from Consumer::ack() ack() previously reused the pollFrame()/sendAckRequest() machinery to transparently reconnect and resend the CommandAck if the connection dropped mid-wait. That's the wrong recovery model for this specific call: a dropped connection during ack() can mean the broker already decided this consumer is gone and redelivered the message to another consumer on the same subscription (Shared/Key_Shared). Silently reconnecting and resending can't undo that redelivery, and it hides from the caller the fact that their handler may now be running twice. ack() now throws IOException immediately on a dropped connection, regardless of ConsumerOptions::getReconnectPolicy(), so the caller gets an honest, timely signal instead of the library quietly retrying underneath it. receive() is unaffected and keeps its own inline reconnect handling for the ordinary receive-loop case. Removes the now-unused pollFrame()/sendAckRequest() helpers. Refs #49 --- src/Consumer.php | 109 +++++++++++++---------------------------------- 1 file changed, 30 insertions(+), 79 deletions(-) diff --git a/src/Consumer.php b/src/Consumer.php index 2f89587..9b14831 100644 --- a/src/Consumer.php +++ b/src/Consumer.php @@ -170,10 +170,27 @@ public function receive(bool $loop = true): Message return $this->messageQueue->dequeue(); } - $response = $this->pollFrame($this->getWaitSeconds()); + try { + $response = $this->eventloop->wait($this->getWaitSeconds()); + } catch (IOException $e) { + $response = null; + + $policy = $this->options->getReconnectPolicy(); + // not enable reconnect + if (!$policy['status']) { + throw $e; + } + + if ($this->reconnect($policy)) { + return $this->receive($loop); + } + } // nack $this->executeInternalNack(); + + // ping + $this->ping(); if (is_null($response)) { if (!$loop) { @@ -221,8 +238,15 @@ public function batchReceive(bool $loop = true): array * * While waiting, any MESSAGE frame that arrives first (the broker may deliver one before * the ACK_RESPONSE, since the connection is asynchronous) is queued rather than discarded. - * If the connection is already dead -- or drops while waiting -- sendAckRequest() reconnects - * and resends the CommandAck with a fresh request ID on the new connection before (re)waiting. + * + * Unlike receive(), this does not reconnect on a dropped connection: a lost connection + * mid-ack throws IOException immediately, regardless of ConsumerOptions::getReconnectPolicy(). + * + * This is deliberate, not an oversight. A dropped connection during ack() means the + * broker may already have decided this consumer is gone and redelivered the message to + * another consumer on the same subscription (Shared/Key_Shared). Throwing immediately + * gives the caller an honest, timely signal instead of a client library quietly retrying + * underneath it. * * @param Message $message * @return CommandAckResponse|null @@ -235,7 +259,8 @@ public function ack(Message $message): ?CommandAckResponse return null; } - $requestId = $this->sendAckRequest($message); + $requestId = Helper::getRequestID(); + $this->getPartitionConsumer($message->getConsumerID())->ack($message, $requestId); $deadline = microtime(true) + $this->options->getAckTimeout(); @@ -245,11 +270,7 @@ public function ack(Message $message): ?CommandAckResponse throw new RuntimeException('Timed out waiting for ACK response.'); } - // Select::wait() forwards this to stream_select()'s tv_sec, which requires an int - $response = $this->pollFrame((int) ceil($remaining), function () use ($message, &$requestId) { - // the connection was rebuilt; the outstanding ACK request died with it - $requestId = $this->sendAckRequest($message); - }); + $response = $this->eventloop->wait((int) ceil($remaining)); if (null === $response) { continue; @@ -392,76 +413,6 @@ protected function getPartitionConsumer(int $consumerID): PartitionConsumer return $this->consumers[ $consumerID ]; } - /** - * Waits for a single frame from the event loop, applying the reconnect policy on - * connection loss and running the ping bookkeeping. - * - * On a dropped connection, once it's rebuilt, $onReconnect (if given) is invoked to - * resend whatever the caller had outstanding on the old connection -- a CommandAck, - * for instance, has no way to be resumed on a new one. Either way this returns null, - * same as a plain timeout, so the caller's own retry loop drives the next poll. - * - * @param int|float|null $timeoutSeconds - * @param callable|null $onReconnect called with no arguments right after a successful reconnect - * @return Response|null - * @throws IOException - */ - private function pollFrame($timeoutSeconds = null, ?callable $onReconnect = null): ?Response - { - try { - $response = $this->eventloop->wait($timeoutSeconds); - - $this->ping(); - - return $response; - } catch (IOException $e) { - $policy = $this->options->getReconnectPolicy(); - - // not enable reconnect - if (!$policy['status']) { - throw $e; - } - - $this->reconnect($policy); - - if ($onReconnect) { - $onReconnect(); - } - - return null; - } - } - - /** - * Sends a CommandAck for $message with a fresh request ID, applying the reconnect policy - * (and retrying the send on the rebuilt connection) if the connection is already dead. - * - * @param Message $message - * @return int the request ID the ack was sent with - * @throws IOException - */ - private function sendAckRequest(Message $message): int - { - $requestId = Helper::getRequestID(); - - try { - $this->getPartitionConsumer($message->getConsumerID())->ack($message, $requestId); - } catch (IOException $e) { - $policy = $this->options->getReconnectPolicy(); - - // not enable reconnect - if (!$policy['status']) { - throw $e; - } - - $this->reconnect($policy); - - return $this->sendAckRequest($message); - } - - return $requestId; - } - /** * Decodes a MESSAGE frame's payload into Message objects, queues them locally, and * decrements the partition's available flow-control permits accordingly.