Added event Hyperf\Kafka\Event\FailToAck and Hyperf\Kafka\Event\FailToRequeue. (#7585)

Author: SkibidiProduction

src/ConsumerManager.php

@@ -23,7 +23,9 @@
 use Hyperf\Kafka\Event\AfterConsumerConfigCreated;
 use Hyperf\Kafka\Event\BeforeConsume;
 use Hyperf\Kafka\Event\BeforeLongLangConsumerCreated;
+use Hyperf\Kafka\Event\FailToAck;
 use Hyperf\Kafka\Event\FailToConsume;
+use Hyperf\Kafka\Event\FailToRequeue;
 use Hyperf\Kafka\Exception\InvalidConsumeResultException;
 use Hyperf\Process\AbstractProcess;
 use Hyperf\Process\ProcessManager;
@@ -131,11 +133,26 @@ function (ConsumeMessage $message) use ($consumer, $consumerConfig) {
                                 }
 
                                 if ($result === Result::ACK) {
-                                    $message->getConsumer()->ack($message);
+                                    try {
+                                        $message->getConsumer()->ack($message);
+                                    } catch (Throwable $exception) {
+                                        $this->dispatcher?->dispatch(new FailToAck($consumer, $message, $exception));
+                                        throw $exception;
+                                    }
                                 }
 
                                 if ($result === Result::REQUEUE) {
-                                    $this->producer->send($message->getTopic(), $message->getValue(), $message->getKey(), $message->getHeaders());
+                                    try {
+                                        $this->producer->send(
+                                            $message->getTopic(),
+                                            $message->getValue(),
+                                            $message->getKey(),
+                                            $message->getHeaders()
+                                        );
+                                    } catch (Throwable $exception) {
+                                        $this->dispatcher?->dispatch(new FailToRequeue($consumer, $message, $exception));
+                                        throw $exception;
+                                    }
                                 }
                             }
 

src/Event/FailToAck.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  group@hyperf.io
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+
+namespace Hyperf\Kafka\Event;
+
+use Hyperf\Kafka\AbstractConsumer;
+use Throwable;
+
+class FailToAck extends Consume
+{
+    public function __construct(AbstractConsumer $consumer, $data, protected Throwable $throwable)
+    {
+        parent::__construct($consumer, $data);
+    }
+
+    public function getThrowable(): Throwable
+    {
+        return $this->throwable;
+    }
+}

src/Event/FailToRequeue.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  group@hyperf.io
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+
+namespace Hyperf\Kafka\Event;
+
+use Hyperf\Kafka\AbstractConsumer;
+use Throwable;
+
+class FailToRequeue extends Consume
+{
+    public function __construct(AbstractConsumer $consumer, $data, protected Throwable $throwable)
+    {
+        parent::__construct($consumer, $data);
+    }
+
+    public function getThrowable(): Throwable
+    {
+        return $this->throwable;
+    }
+}