chore: rollback catching Throwable in callback proxies

idealo · Jan 25, 2025 · 1070e00 · 1070e00
1 parent 57d26fe
commit 1070e00
Show file tree

Hide file tree

Showing 9 changed files with 53 additions and 93 deletions.
diff --git a/src/RdKafka/FFI/ConsumeCallbackProxy.php b/src/RdKafka/FFI/ConsumeCallbackProxy.php
@@ -11,13 +11,9 @@ class ConsumeCallbackProxy extends CallbackProxy
 {
     public function __invoke(CData $nativeMessage, ?CData $opaque = null): void
     {
-        try {
-            ($this->callback)(
-                new Message($nativeMessage),
-                OpaqueMap::get($opaque)
-            );
-        } catch (\Throwable $exception) {
-            error_log($exception->getMessage(), E_ERROR);
-        }
+        ($this->callback)(
+            new Message($nativeMessage),
+            OpaqueMap::get($opaque)
+        );
     }
 }
diff --git a/src/RdKafka/FFI/DrMsgCallbackProxy.php b/src/RdKafka/FFI/DrMsgCallbackProxy.php
@@ -12,14 +12,10 @@ class DrMsgCallbackProxy extends CallbackProxy
 {
     public function __invoke(CData $producer, CData $nativeMessage, ?CData $opaque = null): void
     {
-        try {
-            ($this->callback)(
-                RdKafka::resolveFromCData($producer),
-                new Message($nativeMessage),
-                OpaqueMap::get($opaque)
-            );
-        } catch (\Throwable $exception) {
-            error_log($exception->getMessage(), E_ERROR);
-        }
+        ($this->callback)(
+            RdKafka::resolveFromCData($producer),
+            new Message($nativeMessage),
+            OpaqueMap::get($opaque)
+        );
     }
 }
diff --git a/src/RdKafka/FFI/ErrorCallbackProxy.php b/src/RdKafka/FFI/ErrorCallbackProxy.php
@@ -11,15 +11,11 @@ class ErrorCallbackProxy extends CallbackProxy
 {
     public function __invoke(CData $consumerOrProducer, int $err, string $reason, ?CData $opaque = null): void
     {
-        try {
-            ($this->callback)(
-                RdKafka::resolveFromCData($consumerOrProducer),
-                $err,
-                $reason,
-                OpaqueMap::get($opaque)
-            );
-        } catch (\Throwable $exception) {
-            error_log($exception->getMessage(), E_ERROR);
-        }
+        ($this->callback)(
+            RdKafka::resolveFromCData($consumerOrProducer),
+            $err,
+            $reason,
+            OpaqueMap::get($opaque)
+        );
     }
 }
diff --git a/src/RdKafka/FFI/LogCallbackProxy.php b/src/RdKafka/FFI/LogCallbackProxy.php
@@ -11,15 +11,11 @@ class LogCallbackProxy extends CallbackProxy
 {
     public function __invoke(CData $rdkafka, int $level, string $facility, string $message): void
     {
-        try {
-            ($this->callback)(
-                RdKafka::resolveFromCData($rdkafka),
-                $level,
-                $facility,
-                $message
-            );
-        } catch (\Throwable $exception) {
-            error_log($exception->getMessage(), E_ERROR);
-        }
+        ($this->callback)(
+            RdKafka::resolveFromCData($rdkafka),
+            $level,
+            $facility,
+            $message
+        );
     }
 }
diff --git a/src/RdKafka/FFI/NativePartitionerCallbackProxy.php b/src/RdKafka/FFI/NativePartitionerCallbackProxy.php
@@ -25,20 +25,14 @@ public function __invoke(
         ?CData $topic_opaque = null,
         ?CData $msg_opaque = null
     ): int {
-        try {
-            return (int) Library::{$this->partitionerMethod}(
-                $topic,
-                $keydata,
-                $keylen,
-                $partition_cnt,
-                OpaqueMap::get($topic_opaque),
-                OpaqueMap::get($msg_opaque)
-            );
-        } catch (\Throwable $exception) {
-            error_log($exception->getMessage(), E_ERROR);
-        }
-
-        return RD_KAFKA_PARTITION_UA;
+        return (int) Library::{$this->partitionerMethod}(
+            $topic,
+            $keydata,
+            $keylen,
+            $partition_cnt,
+            OpaqueMap::get($topic_opaque),
+            OpaqueMap::get($msg_opaque)
+        );
     }
 
     public static function create(string $partitionerMethod): Closure

diff --git a/src/RdKafka/FFI/OffsetCommitCallbackProxy.php b/src/RdKafka/FFI/OffsetCommitCallbackProxy.php
@@ -12,15 +12,11 @@ class OffsetCommitCallbackProxy extends CallbackProxy
 {
     public function __invoke(CData $consumer, int $err, CData $nativeTopicPartitionList, ?CData $opaque = null): void
     {
-        try {
-            ($this->callback)(
-                RdKafka::resolveFromCData($consumer),
-                $err,
-                TopicPartitionList::fromCData($nativeTopicPartitionList)->asArray(),
-                OpaqueMap::get($opaque)
-            );
-        } catch (\Throwable $exception) {
-            error_log($exception->getMessage(), E_ERROR);
-        }
+        ($this->callback)(
+            RdKafka::resolveFromCData($consumer),
+            $err,
+            TopicPartitionList::fromCData($nativeTopicPartitionList)->asArray(),
+            OpaqueMap::get($opaque)
+        );
     }
 }
diff --git a/src/RdKafka/FFI/PartitionerCallbackProxy.php b/src/RdKafka/FFI/PartitionerCallbackProxy.php
@@ -17,17 +17,11 @@ public function __invoke(
         ?CData $topic_opaque = null,
         ?CData $msg_opaque = null
     ): int {
-        try {
-            return (int) ($this->callback)(
-                $keydata === null ? null : FFI::string($keydata, $keylen),
-                $partition_cnt,
-                OpaqueMap::get($topic_opaque),
-                OpaqueMap::get($msg_opaque)
-            );
-        } catch (\Throwable $exception) {
-            error_log($exception->getMessage(), E_ERROR);
-        }
-
-        return RD_KAFKA_PARTITION_UA;
+        return (int) ($this->callback)(
+            $keydata === null ? null : FFI::string($keydata, $keylen),
+            $partition_cnt,
+            OpaqueMap::get($topic_opaque),
+            OpaqueMap::get($msg_opaque)
+        );
     }
 }
diff --git a/src/RdKafka/FFI/RebalanceCallbackProxy.php b/src/RdKafka/FFI/RebalanceCallbackProxy.php
@@ -12,15 +12,11 @@ class RebalanceCallbackProxy extends CallbackProxy
 {
     public function __invoke(CData $consumer, int $err, CData $nativeTopicPartitionList, ?CData $opaque = null): void
     {
-        try {
-            ($this->callback)(
-                RdKafka::resolveFromCData($consumer),
-                $err,
-                TopicPartitionList::fromCData($nativeTopicPartitionList)->asArray(),
-                OpaqueMap::get($opaque)
-            );
-        } catch (\Throwable $exception) {
-            error_log($exception->getMessage(), E_ERROR);
-        }
+        ($this->callback)(
+            RdKafka::resolveFromCData($consumer),
+            $err,
+            TopicPartitionList::fromCData($nativeTopicPartitionList)->asArray(),
+            OpaqueMap::get($opaque)
+        );
     }
 }
diff --git a/src/RdKafka/FFI/StatsCallbackProxy.php b/src/RdKafka/FFI/StatsCallbackProxy.php
@@ -12,16 +12,12 @@ class StatsCallbackProxy extends CallbackProxy
 {
     public function __invoke(CData $consumerOrProducer, CData $json, int $json_len, ?CData $opaque = null): int
     {
-        try {
-            ($this->callback)(
-                RdKafka::resolveFromCData($consumerOrProducer),
-                FFI::string($json, $json_len),
-                $json_len,
-                OpaqueMap::get($opaque)
-            );
-        } catch (\Throwable $exception) {
-            error_log($exception->getMessage(), E_ERROR);
-        }
+        ($this->callback)(
+            RdKafka::resolveFromCData($consumerOrProducer),
+            FFI::string($json, $json_len),
+            $json_len,
+            OpaqueMap::get($opaque)
+        );
 
         return 0;
     }