{"version":3,"file":"instrumentation.js","sourceRoot":"","sources":["../../src/instrumentation.ts"],"names":[],"mappings":";AAAA;;;;;;;;;;;;;;GAcG;;;AAEH,4CAU4B;AAC5B,8EAM6C;AAa7C,kBAAkB;AAClB,uCAA0D;AAC1D,6CAAmD;AACnD,oEAKwC;AAExC,MAAa,sBAAuB,SAAQ,qCAAiD;IAC3F,YAAY,SAAuC,EAAE;QACnD,KAAK,CAAC,sBAAY,EAAE,yBAAe,EAAE,MAAM,CAAC,CAAC;IAC/C,CAAC;IAES,IAAI;QACZ,MAAM,OAAO,GAAG,CAAC,aAA6B,EAAE,EAAE;;YAChD,IAAI,IAAA,2BAAS,EAAC,MAAA,aAAa,aAAb,aAAa,uBAAb,aAAa,CAAE,KAAK,0CAAE,SAAS,CAAC,QAAQ,CAAC,EAAE;gBACvD,IAAI,CAAC,OAAO,CAAC,aAAa,CAAC,KAAK,CAAC,SAAS,EAAE,UAAU,CAAC,CAAC;aACzD;YACD,IAAI,IAAA,2BAAS,EAAC,MAAA,aAAa,aAAb,aAAa,uBAAb,aAAa,CAAE,KAAK,0CAAE,SAAS,CAAC,QAAQ,CAAC,EAAE;gBACvD,IAAI,CAAC,OAAO,CAAC,aAAa,CAAC,KAAK,CAAC,SAAS,EAAE,UAAU,CAAC,CAAC;aACzD;QACH,CAAC,CAAC;QAEF,MAAM,MAAM,GAAG,IAAI,qDAAmC,CACpD,SAAS,EACT,CAAC,YAAY,CAAC,EACd,CAAC,aAA6B,EAAE,EAAE;;YAChC,OAAO,CAAC,aAAa,CAAC,CAAC;YACvB,IAAI,CAAC,KAAK,CACR,MAAA,aAAa,aAAb,aAAa,uBAAb,aAAa,CAAE,KAAK,0CAAE,SAAS,EAC/B,UAAU,EACV,IAAI,CAAC,iBAAiB,EAAE,CACzB,CAAC;YACF,IAAI,CAAC,KAAK,CACR,MAAA,aAAa,aAAb,aAAa,uBAAb,aAAa,CAAE,KAAK,0CAAE,SAAS,EAC/B,UAAU,EACV,IAAI,CAAC,iBAAiB,EAAE,CACzB,CAAC;YAEF,OAAO,aAAa,CAAC;QACvB,CAAC,EACD,OAAO,CACR,CAAC;QACF,OAAO,MAAM,CAAC;IAChB,CAAC;IAEO,iBAAiB;QACvB,MAAM,eAAe,GAAG,IAAI,CAAC;QAC7B,OAAO,CAAC,QAAmC,EAAE,EAAE;YAC7C,OAAO,SAAS,QAAQ,CAEtB,GAAG,IAA2C;gBAE9C,MAAM,WAAW,GAAa,QAAQ,CAAC,KAAK,CAAC,IAAI,EAAE,IAAI,CAAC,CAAC;gBAEzD,IAAI,IAAA,2BAAS,EAAC,WAAW,CAAC,GAAG,CAAC,EAAE;oBAC9B,eAAe,CAAC,OAAO,CAAC,WAAW,EAAE,KAAK,CAAC,CAAC;iBAC7C;gBAED,eAAe,CAAC,KAAK,CACnB,WAAW,EACX,KAAK,EACL,eAAe,CAAC,oBAAoB,EAAE,CACvC,CAAC;gBAEF,OAAO,WAAW,CAAC;YACrB,CAAC,CAAC;QACJ,CAAC,CAAC;IACJ,CAAC;IAEO,iBAAiB;QACvB,MAAM,eAAe,GAAG,IAAI,CAAC;QAC7B,OAAO,CAAC,QAAmC,EAAE,EAAE;YAC7C,OAAO,SAAS,QAAQ,CAEtB,GAAG,IAA2C;gBAE9C,MAAM,WAAW,GAAa,QAAQ,CAAC,KAAK,CAAC,IAAI,EAAE,IAAI,CAAC,CAAC;gBAEzD,IAAI,IAAA,2BAAS,EAAC,WAAW,CAAC,SAAS,CAAC,EAAE;oBACpC,eAAe,CAAC,OAAO,CAAC,WAAW,EAAE,WAAW,CAAC,CAAC;iBACnD;gBACD,eAAe,CAAC,KAAK,CACnB,WAAW,EACX,WAAW,EACX,eAAe,CAAC,0BAA0B,EAAE,CAC7C,CAAC;gBAEF,IAAI,IAAA,2BAAS,EAAC,WAAW,CAAC,IAAI,CAAC,EAAE;oBAC/B,eAAe,CAAC,OAAO,CAAC,WAAW,EAAE,MAAM,CAAC,CAAC;iBAC9C;gBACD,eAAe,CAAC,KAAK,CACnB,WAAW,EACX,MAAM,EACN,eAAe,CAAC,qBAAqB,EAAE,CACxC,CAAC;gBAEF,OAAO,WAAW,CAAC;YACrB,CAAC,CAAC;QACJ,CAAC,CAAC;IACJ,CAAC;IAEO,oBAAoB;QAC1B,MAAM,eAAe,GAAG,IAAI,CAAC;QAC7B,OAAO,CAAC,QAAyB,EAAE,EAAE;YACnC,OAAO,SAAS,GAAG,CAEjB,GAAG,IAAiC;gBAEpC,MAAM,MAAM,GAAG,IAAI,CAAC,CAAC,CAAC,CAAC;gBACvB,IAAI,MAAM,aAAN,MAAM,uBAAN,MAAM,CAAE,WAAW,EAAE;oBACvB,IAAI,IAAA,2BAAS,EAAC,MAAM,CAAC,WAAW,CAAC,EAAE;wBACjC,eAAe,CAAC,OAAO,CAAC,MAAM,EAAE,aAAa,CAAC,CAAC;qBAChD;oBACD,eAAe,CAAC,KAAK,CACnB,MAAM,EACN,aAAa,EACb,eAAe,CAAC,4BAA4B,EAAE,CAC/C,CAAC;iBACH;gBACD,IAAI,MAAM,aAAN,MAAM,uBAAN,MAAM,CAAE,SAAS,EAAE;oBACrB,IAAI,IAAA,2BAAS,EAAC,MAAM,CAAC,SAAS,CAAC,EAAE;wBAC/B,eAAe,CAAC,OAAO,CAAC,MAAM,EAAE,WAAW,CAAC,CAAC;qBAC9C;oBACD,eAAe,CAAC,KAAK,CACnB,MAAM,EACN,WAAW,EACX,eAAe,CAAC,0BAA0B,EAAE,CAC7C,CAAC;iBACH;gBACD,OAAO,QAAQ,CAAC,IAAI,CAAC,IAAI,EAAE,MAAM,CAAC,CAAC;YACrC,CAAC,CAAC;QACJ,CAAC,CAAC;IACJ,CAAC;IAEO,4BAA4B;QAClC,MAAM,eAAe,GAAG,IAAI,CAAC;QAC7B,OAAO,CAAC,QAA0C,EAAE,EAAE;YACpD,OAAO,SAAS,WAAW,CAEzB,GAAG,IAAoC;gBAEvC,MAAM,OAAO,GAAG,IAAI,CAAC,CAAC,CAAC,CAAC;gBACxB,MAAM,iBAAiB,GAAY,iBAAW,CAAC,OAAO,CACpD,kBAAY,EACZ,OAAO,CAAC,OAAO,CAAC,OAAO,EACvB,gCAAmB,CACpB,CAAC;gBACF,MAAM,IAAI,GAAG,eAAe,CAAC,kBAAkB,CAC7C,OAAO,CAAC,KAAK,EACb,OAAO,CAAC,OAAO,EACf,uDAAgC,EAChC,iBAAiB,CAClB,CAAC;gBAEF,MAAM,kBAAkB,GAAG,aAAO,CAAC,IAAI,CACrC,WAAK,CAAC,OAAO,CAAC,iBAAiB,EAAE,IAAI,CAAC,EACtC,GAAG,EAAE;oBACH,OAAO,QAAS,CAAC,KAAK,CAAC,IAAI,EAAE,IAAI,CAAC,CAAC;gBACrC,CAAC,CACF,CAAC;gBACF,OAAO,eAAe,CAAC,kBAAkB,CAAC,CAAC,IAAI,CAAC,EAAE,kBAAkB,CAAC,CAAC;YACxE,CAAC,CAAC;QACJ,CAAC,CAAC;IACJ,CAAC;IAEO,0BAA0B;QAChC,OAAO,CAAC,QAAwC,EAAE,EAAE;YAClD,MAAM,eAAe,GAAG,IAAI,CAAC;YAC7B,OAAO,SAAS,SAAS,CAEvB,GAAG,IAAkC;gBAErC,MAAM,OAAO,GAAG,IAAI,CAAC,CAAC,CAAC,CAAC;gBACxB,gKAAgK;gBAChK,MAAM,aAAa,GAAG,eAAe,CAAC,kBAAkB,CACtD,OAAO,CAAC,KAAK,CAAC,KAAK,EACnB,SAAS,EACT,uDAAgC,EAChC,kBAAY,CACb,CAAC;gBACF,OAAO,aAAO,CAAC,IAAI,CACjB,WAAK,CAAC,OAAO,CAAC,aAAO,CAAC,MAAM,EAAE,EAAE,aAAa,CAAC,EAC9C,GAAG,EAAE;oBACH,MAAM,KAAK,GAAG,OAAO,CAAC,KAAK,CAAC,QAAQ,CAAC,GAAG,CACtC,CAAC,OAAqB,EAAE,EAAE;;wBACxB,MAAM,iBAAiB,GAAY,iBAAW,CAAC,OAAO,CACpD,kBAAY,EACZ,OAAO,CAAC,OAAO,EACf,gCAAmB,CACpB,CAAC;wBACF,MAAM,WAAW,GAAG,MAAA,WAAK;6BACtB,OAAO,CAAC,iBAAiB,CAAC,0CACzB,WAAW,EAAE,CAAC;wBAClB,IAAI,YAA8B,CAAC;wBACnC,IAAI,WAAW,EAAE;4BACf,YAAY,GAAG;gCACb,OAAO,EAAE,WAAW;6BACrB,CAAC;yBACH;wBACD,OAAO,eAAe,CAAC,kBAAkB,CACvC,OAAO,CAAC,KAAK,CAAC,KAAK,EACnB,OAAO,EACP,uDAAgC,EAChC,SAAS,EACT,YAAY,CACb,CAAC;oBACJ,CAAC,CACF,CAAC;oBACF,MAAM,mBAAmB,GAAkB,QAAS,CAAC,KAAK,CACxD,IAAI,EACJ,IAAI,CACL,CAAC;oBACF,KAAK,CAAC,OAAO,CAAC,aAAa,CAAC,CAAC;oBAC7B,OAAO,eAAe,CAAC,kBAAkB,CACvC,KAAK,EACL,mBAAmB,CACpB,CAAC;gBACJ,CAAC,CACF,CAAC;YACJ,CAAC,CAAC;QACJ,CAAC,CAAC;IACJ,CAAC;IAEO,0BAA0B;QAChC,MAAM,eAAe,GAAG,IAAI,CAAC;QAC7B,OAAO,CAAC,QAA+B,EAAE,EAAE;YACzC,OAAO,SAAS,SAAS,CAEvB,GAAG,IAAuC;gBAE1C,MAAM,KAAK,GAAG,IAAI,CAAC,CAAC,CAAC,CAAC;gBACtB,MAAM,QAAQ,GAAG,KAAK,CAAC,aAAa,IAAI,EAAE,CAAC;gBAC3C,MAAM,KAAK,GAAW,QAAQ;qBAC3B,GAAG,CAAC,YAAY,CAAC,EAAE,CAClB,YAAY,CAAC,QAAQ,CAAC,GAAG,CAAC,OAAO,CAAC,EAAE,CAClC,eAAe,CAAC,kBAAkB,CAAC,YAAY,CAAC,KAAK,EAAE,OAAO,CAAC,CAChE,CACF;qBACA,MAAM,CAAC,CAAC,GAAG,EAAE,GAAG,EAAE,EAAE,CAAC,GAAG,CAAC,MAAM,CAAC,GAAG,CAAC,EAAE,EAAE,CAAC,CAAC;gBAE7C,MAAM,cAAc,GAA8B,QAAQ,CAAC,KAAK,CAC9D,IAAI,EACJ,IAAI,CACL,CAAC;gBACF,OAAO,eAAe,CAAC,kBAAkB,CAAC,KAAK,EAAE,cAAc,CAAC,CAAC;YACnE,CAAC,CAAC;QACJ,CAAC,CAAC;IACJ,CAAC;IAEO,qBAAqB;QAC3B,MAAM,eAAe,GAAG,IAAI,CAAC;QAC7B,OAAO,CAAC,QAA0B,EAAE,EAAE;YACpC,OAAO,SAAS,IAAI,CAElB,GAAG,IAAkC;gBAErC,MAAM,MAAM,GAAG,IAAI,CAAC,CAAC,CAAC,CAAC;gBACvB,MAAM,KAAK,GAAW,MAAM,CAAC,QAAQ,CAAC,GAAG,CAAC,OAAO,CAAC,EAAE;oBAClD,OAAO,eAAe,CAAC,kBAAkB,CAAC,MAAM,CAAC,KAAK,EAAE,OAAO,CAAC,CAAC;gBACnE,CAAC,CAAC,CAAC;gBAEH,MAAM,cAAc,GAA8B,QAAQ,CAAC,KAAK,CAC9D,IAAI,EACJ,IAAI,CACL,CAAC;gBACF,OAAO,eAAe,CAAC,kBAAkB,CAAC,KAAK,EAAE,cAAc,CAAC,CAAC;YACnE,CAAC,CAAC;QACJ,CAAC,CAAC;IACJ,CAAC;IAEO,kBAAkB,CACxB,KAAa,EACb,WAAuB;QAEvB,OAAO,OAAO,CAAC,OAAO,CAAC,WAAW,CAAC;aAChC,KAAK,CAAC,MAAM,CAAC,EAAE;YACd,IAAI,YAAoB,CAAC;YACzB,IAAI,OAAO,MAAM,KAAK,QAAQ;gBAAE,YAAY,GAAG,MAAM,CAAC;iBACjD,IACH,OAAO,MAAM,KAAK,QAAQ;gBAC1B,MAAM,CAAC,SAAS,CAAC,cAAc,CAAC,IAAI,CAAC,MAAM,EAAE,SAAS,CAAC;gBAEvD,YAAY,GAAG,MAAM,CAAC,OAAO,CAAC;YAEhC,KAAK,CAAC,OAAO,CAAC,IAAI,CAAC,EAAE,CACnB,IAAI,CAAC,SAAS,CAAC;gBACb,IAAI,EAAE,oBAAc,CAAC,KAAK;gBAC1B,OAAO,EAAE,YAAY;aACtB,CAAC,CACH,CAAC;YAEF,MAAM,MAAM,CAAC;QACf,CAAC,CAAC;aACD,OAAO,CAAC,GAAG,EAAE;YACZ,KAAK,CAAC,OAAO,CAAC,IAAI,CAAC,EAAE,CAAC,IAAI,CAAC,GAAG,EAAE,CAAC,CAAC;QACpC,CAAC,CAAC,CAAC;IACP,CAAC;IAEO,kBAAkB,CACxB,KAAa,EACb,OAAiC,EACjC,SAAiB,EACjB,OAA4B,EAC5B,IAAW;QAEX,MAAM,IAAI,GAAG,IAAI,CAAC,MAAM,CAAC,SAAS,CAChC,KAAK,EACL;YACE,IAAI,EAAE,cAAQ,CAAC,QAAQ;YACvB,UAAU,EAAE;gBACV,CAAC,gDAAyB,CAAC,EAAE,OAAO;gBACpC,CAAC,qDAA8B,CAAC,EAAE,KAAK;gBACvC,CAAC,mDAA4B,CAAC,EAAE,SAAS;aAC1C;YACD,KAAK,EAAE,IAAI,CAAC,CAAC,CAAC,CAAC,IAAI,CAAC,CAAC,CAAC,CAAC,EAAE;SAC1B,EACD,OAAO,CACR,CAAC;QAEF,MAAM,EAAE,YAAY,EAAE,GAAG,IAAI,CAAC,SAAS,EAAE,CAAC;QAC1C,IAAI,YAAY,IAAI,OAAO,EAAE;YAC3B,IAAA,wCAAsB,EACpB,GAAG,EAAE,CAAC,YAAY,CAAC,IAAI,EAAE,EAAE,KAAK,EAAE,OAAO,EAAE,CAAC,EAC5C,CAAC,CAAC,EAAE;gBACF,IAAI,CAAC;oBAAE,IAAI,CAAC,KAAK,CAAC,KAAK,CAAC,oBAAoB,EAAE,CAAC,CAAC,CAAC;YACnD,CAAC,EACD,IAAI,CACL,CAAC;SACH;QAED,OAAO,IAAI,CAAC;IACd,CAAC;IAEO,kBAAkB,CAAC,KAAa,EAAE,OAAgB;;QACxD,MAAM,IAAI,GAAG,IAAI,CAAC,MAAM,CAAC,SAAS,CAAC,KAAK,EAAE;YACxC,IAAI,EAAE,cAAQ,CAAC,QAAQ;YACvB,UAAU,EAAE;gBACV,CAAC,gDAAyB,CAAC,EAAE,OAAO;gBACpC,CAAC,qDAA8B,CAAC,EAAE,KAAK;aACxC;SACF,CAAC,CAAC;QAEH,OAAO,CAAC,OAAO,GAAG,MAAA,OAAO,CAAC,OAAO,mCAAI,EAAE,CAAC;QACxC,iBAAW,CAAC,MAAM,CAAC,WAAK,CAAC,OAAO,CAAC,aAAO,CAAC,MAAM,EAAE,EAAE,IAAI,CAAC,EAAE,OAAO,CAAC,OAAO,CAAC,CAAC;QAE3E,MAAM,EAAE,YAAY,EAAE,GAAG,IAAI,CAAC,SAAS,EAAE,CAAC;QAC1C,IAAI,YAAY,EAAE;YAChB,IAAA,wCAAsB,EACpB,GAAG,EAAE,CAAC,YAAY,CAAC,IAAI,EAAE,EAAE,KAAK,EAAE,OAAO,EAAE,CAAC,EAC5C,CAAC,CAAC,EAAE;gBACF,IAAI,CAAC;oBAAE,IAAI,CAAC,KAAK,CAAC,KAAK,CAAC,oBAAoB,EAAE,CAAC,CAAC,CAAC;YACnD,CAAC,EACD,IAAI,CACL,CAAC;SACH;QAED,OAAO,IAAI,CAAC;IACd,CAAC;CACF;AA/VD,wDA+VC","sourcesContent":["/*\n * Copyright The OpenTelemetry Authors, Aspecto\n *\n * Licensed under the Apache License, Version 2.0 (the \"License\");\n * you may not use this file except in compliance with the License.\n * You may obtain a copy of the License at\n *\n *      https://www.apache.org/licenses/LICENSE-2.0\n *\n * Unless required by applicable law or agreed to in writing, software\n * distributed under the License is distributed on an \"AS IS\" BASIS,\n * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n * See the License for the specific language governing permissions and\n * limitations under the License.\n */\n\nimport {\n  SpanKind,\n  Span,\n  SpanStatusCode,\n  Context,\n  propagation,\n  Link,\n  trace,\n  context,\n  ROOT_CONTEXT,\n} from '@opentelemetry/api';\nimport {\n  MESSAGINGOPERATIONVALUES_PROCESS,\n  MESSAGINGOPERATIONVALUES_RECEIVE,\n  SEMATTRS_MESSAGING_SYSTEM,\n  SEMATTRS_MESSAGING_DESTINATION,\n  SEMATTRS_MESSAGING_OPERATION,\n} from '@opentelemetry/semantic-conventions';\nimport type * as kafkaJs from 'kafkajs';\nimport type {\n  EachBatchHandler,\n  EachMessageHandler,\n  Producer,\n  RecordMetadata,\n  Message,\n  ConsumerRunConfig,\n  KafkaMessage,\n  Consumer,\n} from 'kafkajs';\nimport { KafkaJsInstrumentationConfig } from './types';\n/** @knipignore */\nimport { PACKAGE_NAME, PACKAGE_VERSION } from './version';\nimport { bufferTextMapGetter } from './propagator';\nimport {\n  InstrumentationBase,\n  InstrumentationNodeModuleDefinition,\n  safeExecuteInTheMiddle,\n  isWrapped,\n} from '@opentelemetry/instrumentation';\n\nexport class KafkaJsInstrumentation extends InstrumentationBase<KafkaJsInstrumentationConfig> {\n  constructor(config: KafkaJsInstrumentationConfig = {}) {\n    super(PACKAGE_NAME, PACKAGE_VERSION, config);\n  }\n\n  protected init() {\n    const unpatch = (moduleExports: typeof kafkaJs) => {\n      if (isWrapped(moduleExports?.Kafka?.prototype.producer)) {\n        this._unwrap(moduleExports.Kafka.prototype, 'producer');\n      }\n      if (isWrapped(moduleExports?.Kafka?.prototype.consumer)) {\n        this._unwrap(moduleExports.Kafka.prototype, 'consumer');\n      }\n    };\n\n    const module = new InstrumentationNodeModuleDefinition(\n      'kafkajs',\n      ['>=0.1.0 <3'],\n      (moduleExports: typeof kafkaJs) => {\n        unpatch(moduleExports);\n        this._wrap(\n          moduleExports?.Kafka?.prototype,\n          'producer',\n          this._getProducerPatch()\n        );\n        this._wrap(\n          moduleExports?.Kafka?.prototype,\n          'consumer',\n          this._getConsumerPatch()\n        );\n\n        return moduleExports;\n      },\n      unpatch\n    );\n    return module;\n  }\n\n  private _getConsumerPatch() {\n    const instrumentation = this;\n    return (original: kafkaJs.Kafka['consumer']) => {\n      return function consumer(\n        this: kafkaJs.Kafka,\n        ...args: Parameters<kafkaJs.Kafka['consumer']>\n      ) {\n        const newConsumer: Consumer = original.apply(this, args);\n\n        if (isWrapped(newConsumer.run)) {\n          instrumentation._unwrap(newConsumer, 'run');\n        }\n\n        instrumentation._wrap(\n          newConsumer,\n          'run',\n          instrumentation._getConsumerRunPatch()\n        );\n\n        return newConsumer;\n      };\n    };\n  }\n\n  private _getProducerPatch() {\n    const instrumentation = this;\n    return (original: kafkaJs.Kafka['producer']) => {\n      return function consumer(\n        this: kafkaJs.Kafka,\n        ...args: Parameters<kafkaJs.Kafka['producer']>\n      ) {\n        const newProducer: Producer = original.apply(this, args);\n\n        if (isWrapped(newProducer.sendBatch)) {\n          instrumentation._unwrap(newProducer, 'sendBatch');\n        }\n        instrumentation._wrap(\n          newProducer,\n          'sendBatch',\n          instrumentation._getProducerSendBatchPatch()\n        );\n\n        if (isWrapped(newProducer.send)) {\n          instrumentation._unwrap(newProducer, 'send');\n        }\n        instrumentation._wrap(\n          newProducer,\n          'send',\n          instrumentation._getProducerSendPatch()\n        );\n\n        return newProducer;\n      };\n    };\n  }\n\n  private _getConsumerRunPatch() {\n    const instrumentation = this;\n    return (original: Consumer['run']) => {\n      return function run(\n        this: Consumer,\n        ...args: Parameters<Consumer['run']>\n      ): ReturnType<Consumer['run']> {\n        const config = args[0];\n        if (config?.eachMessage) {\n          if (isWrapped(config.eachMessage)) {\n            instrumentation._unwrap(config, 'eachMessage');\n          }\n          instrumentation._wrap(\n            config,\n            'eachMessage',\n            instrumentation._getConsumerEachMessagePatch()\n          );\n        }\n        if (config?.eachBatch) {\n          if (isWrapped(config.eachBatch)) {\n            instrumentation._unwrap(config, 'eachBatch');\n          }\n          instrumentation._wrap(\n            config,\n            'eachBatch',\n            instrumentation._getConsumerEachBatchPatch()\n          );\n        }\n        return original.call(this, config);\n      };\n    };\n  }\n\n  private _getConsumerEachMessagePatch() {\n    const instrumentation = this;\n    return (original: ConsumerRunConfig['eachMessage']) => {\n      return function eachMessage(\n        this: unknown,\n        ...args: Parameters<EachMessageHandler>\n      ): Promise<void> {\n        const payload = args[0];\n        const propagatedContext: Context = propagation.extract(\n          ROOT_CONTEXT,\n          payload.message.headers,\n          bufferTextMapGetter\n        );\n        const span = instrumentation._startConsumerSpan(\n          payload.topic,\n          payload.message,\n          MESSAGINGOPERATIONVALUES_PROCESS,\n          propagatedContext\n        );\n\n        const eachMessagePromise = context.with(\n          trace.setSpan(propagatedContext, span),\n          () => {\n            return original!.apply(this, args);\n          }\n        );\n        return instrumentation._endSpansOnPromise([span], eachMessagePromise);\n      };\n    };\n  }\n\n  private _getConsumerEachBatchPatch() {\n    return (original: ConsumerRunConfig['eachBatch']) => {\n      const instrumentation = this;\n      return function eachBatch(\n        this: unknown,\n        ...args: Parameters<EachBatchHandler>\n      ): Promise<void> {\n        const payload = args[0];\n        // https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/trace/semantic_conventions/messaging.md#topic-with-multiple-consumers\n        const receivingSpan = instrumentation._startConsumerSpan(\n          payload.batch.topic,\n          undefined,\n          MESSAGINGOPERATIONVALUES_RECEIVE,\n          ROOT_CONTEXT\n        );\n        return context.with(\n          trace.setSpan(context.active(), receivingSpan),\n          () => {\n            const spans = payload.batch.messages.map(\n              (message: KafkaMessage) => {\n                const propagatedContext: Context = propagation.extract(\n                  ROOT_CONTEXT,\n                  message.headers,\n                  bufferTextMapGetter\n                );\n                const spanContext = trace\n                  .getSpan(propagatedContext)\n                  ?.spanContext();\n                let origSpanLink: Link | undefined;\n                if (spanContext) {\n                  origSpanLink = {\n                    context: spanContext,\n                  };\n                }\n                return instrumentation._startConsumerSpan(\n                  payload.batch.topic,\n                  message,\n                  MESSAGINGOPERATIONVALUES_PROCESS,\n                  undefined,\n                  origSpanLink\n                );\n              }\n            );\n            const batchMessagePromise: Promise<void> = original!.apply(\n              this,\n              args\n            );\n            spans.unshift(receivingSpan);\n            return instrumentation._endSpansOnPromise(\n              spans,\n              batchMessagePromise\n            );\n          }\n        );\n      };\n    };\n  }\n\n  private _getProducerSendBatchPatch() {\n    const instrumentation = this;\n    return (original: Producer['sendBatch']) => {\n      return function sendBatch(\n        this: Producer,\n        ...args: Parameters<Producer['sendBatch']>\n      ): ReturnType<Producer['sendBatch']> {\n        const batch = args[0];\n        const messages = batch.topicMessages || [];\n        const spans: Span[] = messages\n          .map(topicMessage =>\n            topicMessage.messages.map(message =>\n              instrumentation._startProducerSpan(topicMessage.topic, message)\n            )\n          )\n          .reduce((acc, val) => acc.concat(val), []);\n\n        const origSendResult: Promise<RecordMetadata[]> = original.apply(\n          this,\n          args\n        );\n        return instrumentation._endSpansOnPromise(spans, origSendResult);\n      };\n    };\n  }\n\n  private _getProducerSendPatch() {\n    const instrumentation = this;\n    return (original: Producer['send']) => {\n      return function send(\n        this: Producer,\n        ...args: Parameters<Producer['send']>\n      ): ReturnType<Producer['send']> {\n        const record = args[0];\n        const spans: Span[] = record.messages.map(message => {\n          return instrumentation._startProducerSpan(record.topic, message);\n        });\n\n        const origSendResult: Promise<RecordMetadata[]> = original.apply(\n          this,\n          args\n        );\n        return instrumentation._endSpansOnPromise(spans, origSendResult);\n      };\n    };\n  }\n\n  private _endSpansOnPromise<T>(\n    spans: Span[],\n    sendPromise: Promise<T>\n  ): Promise<T> {\n    return Promise.resolve(sendPromise)\n      .catch(reason => {\n        let errorMessage: string;\n        if (typeof reason === 'string') errorMessage = reason;\n        else if (\n          typeof reason === 'object' &&\n          Object.prototype.hasOwnProperty.call(reason, 'message')\n        )\n          errorMessage = reason.message;\n\n        spans.forEach(span =>\n          span.setStatus({\n            code: SpanStatusCode.ERROR,\n            message: errorMessage,\n          })\n        );\n\n        throw reason;\n      })\n      .finally(() => {\n        spans.forEach(span => span.end());\n      });\n  }\n\n  private _startConsumerSpan(\n    topic: string,\n    message: KafkaMessage | undefined,\n    operation: string,\n    context: Context | undefined,\n    link?: Link\n  ) {\n    const span = this.tracer.startSpan(\n      topic,\n      {\n        kind: SpanKind.CONSUMER,\n        attributes: {\n          [SEMATTRS_MESSAGING_SYSTEM]: 'kafka',\n          [SEMATTRS_MESSAGING_DESTINATION]: topic,\n          [SEMATTRS_MESSAGING_OPERATION]: operation,\n        },\n        links: link ? [link] : [],\n      },\n      context\n    );\n\n    const { consumerHook } = this.getConfig();\n    if (consumerHook && message) {\n      safeExecuteInTheMiddle(\n        () => consumerHook(span, { topic, message }),\n        e => {\n          if (e) this._diag.error('consumerHook error', e);\n        },\n        true\n      );\n    }\n\n    return span;\n  }\n\n  private _startProducerSpan(topic: string, message: Message) {\n    const span = this.tracer.startSpan(topic, {\n      kind: SpanKind.PRODUCER,\n      attributes: {\n        [SEMATTRS_MESSAGING_SYSTEM]: 'kafka',\n        [SEMATTRS_MESSAGING_DESTINATION]: topic,\n      },\n    });\n\n    message.headers = message.headers ?? {};\n    propagation.inject(trace.setSpan(context.active(), span), message.headers);\n\n    const { producerHook } = this.getConfig();\n    if (producerHook) {\n      safeExecuteInTheMiddle(\n        () => producerHook(span, { topic, message }),\n        e => {\n          if (e) this._diag.error('producerHook error', e);\n        },\n        true\n      );\n    }\n\n    return span;\n  }\n}\n"]}