fix(instrumentation-kafkajs): fix instr to work with [email protected] and earlier

The tests broke on [email protected] and earlier.
The instrumentation crashed on [email protected] and earlier.

Refs: open-telemetry#2784 (comment)
Fixes: open-telemetry#2784

Author: trentm

plugins/node/instrumentation-kafkajs/README.md

@@ -69,7 +69,7 @@ This package uses `@opentelemetry/semantic-conventions` version `1.30+`, which i
 | --------------------- | ------------------------------------- | ------------------------------------------------------------ |
 | Consumer              | `messaging.process.duration`          | Duration of processing operation. [1]                        |
 | Consumer              | `messaging.client.consumed.messages`  | Number of messages that were delivered to the application.   |
-| Consumer and Producer | `messaging.client.operation.duration` | Number of messages that were delivered to the application.   |
+| Consumer and Producer | `messaging.client.operation.duration` | Number of messages that were delivered to the application. (Only emitted for [email protected] and later.)   |
 | Producer              | `messaging.client.sent.messages`      | Number of messages producer attempted to send to the broker. |
 
 **[1] `messaging.process.duration`:** In the context of `eachBatch`, this metric will be emitted once for each message but the value reflects the duration of the entire batch.

plugins/node/instrumentation-kafkajs/src/instrumentation.ts

@@ -251,10 +251,13 @@ export class KafkaJsInstrumentation extends InstrumentationBase<KafkaJsInstrumen
   private _setKafkaEventListeners(kafkaObj: KafkaEventEmitter) {
     if (kafkaObj[EVENT_LISTENERS_SET]) return;
 
-    kafkaObj.on(
-      kafkaObj.events.REQUEST,
-      this._recordClientDurationMetric.bind(this)
-    );
+    // The REQUEST Consumer event was added in [email protected].
+    if (kafkaObj.events?.REQUEST) {
+      kafkaObj.on(
+        kafkaObj.events.REQUEST,
+        this._recordClientDurationMetric.bind(this)
+      );
+    }
 
     kafkaObj[EVENT_LISTENERS_SET] = true;
   }

plugins/node/instrumentation-kafkajs/test/kafkajs.test.ts

@@ -220,9 +220,7 @@ describe('instrumentation-kafkajs', () => {
         );
         instrumentation.disable();
         instrumentation.enable();
-        producer = kafka.producer({
-          createPartitioner: kafkajs.Partitioners.LegacyPartitioner,
-        });
+        producer = kafka.producer();
       }
       beforeEach(() => {
         initializeProducer();
@@ -479,9 +477,7 @@ describe('instrumentation-kafkajs', () => {
         });
         instrumentation.disable();
         instrumentation.enable();
-        producer = kafka.producer({
-          createPartitioner: kafkajs.Partitioners.LegacyPartitioner,
-        });
+        producer = kafka.producer();
       });
 
       it('error in send create failed span', async () => {
@@ -634,9 +630,7 @@ describe('instrumentation-kafkajs', () => {
         instrumentation.disable();
         instrumentation.setConfig(config);
         instrumentation.enable();
-        producer = kafka.producer({
-          createPartitioner: kafkajs.Partitioners.LegacyPartitioner,
-        });
+        producer = kafka.producer();
       });
 
       it('producer hook add span attribute with value from message', async () => {
@@ -671,9 +665,7 @@ describe('instrumentation-kafkajs', () => {
         instrumentation.disable();
         instrumentation.setConfig(config);
         instrumentation.enable();
-        producer = kafka.producer({
-          createPartitioner: kafkajs.Partitioners.LegacyPartitioner,
-        });
+        producer = kafka.producer();
       });
 
       it('producer hook add span attribute with value from message', async () => {
@@ -1227,9 +1219,7 @@ describe('instrumentation-kafkajs', () => {
       storeRunConfig();
       instrumentation.disable();
       instrumentation.enable();
-      producer = kafka.producer({
-        createPartitioner: kafkajs.Partitioners.LegacyPartitioner,
-      });
+      producer = kafka.producer();
       consumer = kafka.consumer({ groupId: 'testing-group-id' });
     });