feat: add originalTraceId header in error topic exception filter

Byndyusoft · Jan 22, 2024 · 30a1cee · 30a1cee
1 parent 6971db3
commit 30a1cee
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 0 deletions.
diff --git a/src/consumer/exceptions/kafkaConsumerErrorTopicExceptionFilter.ts b/src/consumer/exceptions/kafkaConsumerErrorTopicExceptionFilter.ts
@@ -151,6 +151,7 @@ export class KafkaConsumerErrorTopicExceptionFilter
             originalPartition: String(payload.rawPayload.partition),
             originalOffset: payload.rawPayload.message.offset,
             originalTimestamp: payload.rawPayload.message.timestamp,
+            originalTraceId: context.traceId,
             error: JSON.stringify(serializeError(cause)),
           },
         },

diff --git a/src/consumer/interfaces/kafkaConsumerContextInterface.ts b/src/consumer/interfaces/kafkaConsumerContextInterface.ts
@@ -29,4 +29,5 @@ export interface IKafkaConsumerContext {
   readonly kafkaConsumerMessageHandlerLogger: KafkaConsumerMessageHandlerLogger;
 
   readonly isFinalAttempt: boolean;
+  readonly traceId: string;
 }
diff --git a/src/consumer/kafkaConsumerMessageHandler.ts b/src/consumer/kafkaConsumerMessageHandler.ts
@@ -167,6 +167,7 @@ export class KafkaConsumerMessageHandler {
         kafkaConsumerMessageHandlerLogger:
           this.kafkaConsumerMessageHandlerLogger,
         isFinalAttempt,
+        traceId: rootSpan.context().toTraceId(),
       };
 
       const resultOrStream = await messageHandler(payload, context);