Add span creation and error handling in AsyncQueueJobMessageAspect an…

…… (#574) * Add span creation and error handling in AsyncQueueJobMessageAspect and KafkaProducerAspect * Fix span nullability issue in AsyncQueueJobMessageAspect --------- Co-authored-by: Deeka Wong <[email protected]>
friendsofhyperf · Feb 26, 2024 · 8bf5d16 · 8bf5d16
1 parent cb2221c
commit 8bf5d16
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 6 deletions.
diff --git a/src/Tracing/Aspect/AsyncQueueJobMessageAspect.php b/src/Tracing/Aspect/AsyncQueueJobMessageAspect.php
@@ -62,11 +62,16 @@ public function process(ProceedingJoinPoint $proceedingJoinPoint)
 
     public function handlePush(ProceedingJoinPoint $proceedingJoinPoint)
     {
+        $span = $this->startSpan(
+            'async_queue.job.publish',
+            $proceedingJoinPoint->arguments['keys']['job']::class
+        );
+
+        if (! $span) {
+            return $proceedingJoinPoint->process();
+        }
+
         try {
-            $span = $this->startSpan(
-                'async_queue.job.publish',
-                $proceedingJoinPoint->arguments['keys']['job']::class
-            );
             $data = [];
 
             /** @var \Hyperf\AsyncQueue\Driver\Driver $driver */
@@ -77,7 +82,7 @@ public function handlePush(ProceedingJoinPoint $proceedingJoinPoint)
             };
 
             if (count($data)) {
-                $span?->setData($data);
+                $span->setData($data);
             }
 
             $carrier = $this->packer->pack($span);
@@ -86,7 +91,7 @@ public function handlePush(ProceedingJoinPoint $proceedingJoinPoint)
             return $proceedingJoinPoint->process();
         } catch (Throwable) {
         } finally {
-            $span?->finish();
+            $span->finish();
         }
     }
 

diff --git a/src/Tracing/Aspect/KafkaProducerAspect.php b/src/Tracing/Aspect/KafkaProducerAspect.php
@@ -52,6 +52,11 @@ protected function sendAsync(ProceedingJoinPoint $proceedingJoinPoint)
             'kafka.send',
             sprintf('%s::%s', $proceedingJoinPoint->className, $proceedingJoinPoint->methodName)
         );
+
+        if (! $span) {
+            return $proceedingJoinPoint->process();
+        }
+
         $carrier = $this->packer->pack($span);
         $headers = $proceedingJoinPoint->arguments['keys']['headers'] ?? [];
         $headers[] = (new RecordHeader())
@@ -70,6 +75,11 @@ protected function sendBatchAsync(ProceedingJoinPoint $proceedingJoinPoint)
             'kafka.send_batch',
             sprintf('%s::%s', $proceedingJoinPoint->className, $proceedingJoinPoint->methodName)
         );
+
+        if (! $span) {
+            return $proceedingJoinPoint->process();
+        }
+
         $carrier = $this->packer->pack($span);
 
         foreach ($messages as $message) {