test: update confluent kafka tests to use dynamic topic (#6079)

Author: rochdev

packages/datadog-plugin-confluentinc-kafka-javascript/test/index.spec.js

@@ -1,5 +1,6 @@
 'use strict'
 
+const { randomUUID } = require('crypto')
 const { expect } = require('chai')
 const agent = require('../../dd-trace/test/plugins/agent')
 const { expectSomeSpan, withDefaults } = require('../../dd-trace/test/plugins/helpers')
@@ -10,14 +11,12 @@ const DataStreamsContext = require('../../dd-trace/src/datastreams/context')
 const { computePathwayHash } = require('../../dd-trace/src/datastreams/pathway')
 const { ENTRY_PARENT_HASH, DataStreamsProcessor } = require('../../dd-trace/src/datastreams/processor')
 
-const testTopic = 'test-topic'
-
-const getDsmPathwayHash = (isProducer, parentHash) => {
+const getDsmPathwayHash = (testTopic, isProducer, parentHash) => {
   let edgeTags
   if (isProducer) {
     edgeTags = ['direction:out', 'topic:' + testTopic, 'type:kafka']
   } else {
-    edgeTags = ['direction:in', 'group:test-group', 'topic:' + testTopic, 'type:kafka']
+    edgeTags = ['direction:in', 'group:test-group-confluent', 'topic:' + testTopic, 'type:kafka']
   }
 
   edgeTags.sort()
@@ -26,6 +25,7 @@ const getDsmPathwayHash = (isProducer, parentHash) => {
 
 describe('Plugin', () => {
   const module = '@confluentinc/kafka-javascript'
+  const groupId = 'test-group-confluent'
 
   describe('confluentinc-kafka-javascript', function () {
     this.timeout(30000)
@@ -36,11 +36,13 @@ describe('Plugin', () => {
 
     withVersions('confluentinc-kafka-javascript', module, (version) => {
       let kafka
+      let admin
       let tracer
       let Kafka
       let ConfluentKafka
       let messages
       let nativeApi
+      let testTopic
 
       describe('without configuration', () => {
         beforeEach(async () => {
@@ -60,11 +62,24 @@ describe('Plugin', () => {
           kafka = new Kafka({
             kafkaJS: {
               clientId: `kafkajs-test-${version}`,
-              brokers: ['127.0.0.1:9092']
+              brokers: ['127.0.0.1:9092'],
+              logLevel: ConfluentKafka.logLevel.WARN
             }
           })
+          testTopic = `test-topic-${randomUUID()}`
+          admin = kafka.admin()
+          await admin.connect()
+          await admin.createTopics({
+            topics: [{
+              topic: testTopic,
+              numPartitions: 1,
+              replicationFactor: 1
+            }]
+          })
         })
 
+        afterEach(() => admin.disconnect())
+
         describe('kafkaJS api', () => {
           describe('producer', () => {
             it('should be instrumented', async () => {
@@ -74,7 +89,7 @@ describe('Plugin', () => {
                 meta: {
                   'span.kind': 'producer',
                   component: 'confluentinc-kafka-javascript',
-                  'messaging.destination.name': 'test-topic',
+                  'messaging.destination.name': testTopic,
                   'messaging.kafka.bootstrap.servers': '127.0.0.1:9092'
                 },
                 metrics: {
@@ -125,7 +140,7 @@ describe('Plugin', () => {
             beforeEach(async () => {
               messages = [{ key: 'key1', value: 'test2' }]
               consumer = kafka.consumer({
-                kafkaJS: { groupId: 'test-group' }
+                kafkaJS: { groupId, fromBeginning: true, autoCommit: false }
               })
               await consumer.connect()
               await consumer.subscribe({ topic: testTopic })
@@ -142,7 +157,7 @@ describe('Plugin', () => {
                 meta: {
                   'span.kind': 'consumer',
                   component: 'confluentinc-kafka-javascript',
-                  'messaging.destination.name': 'test-topic'
+                  'messaging.destination.name': testTopic
                 },
                 resource: testTopic,
                 error: 0,
@@ -151,7 +166,7 @@ describe('Plugin', () => {
 
               const consumerReceiveMessagePromise = new Promise(resolve => {
                 consumer.run({
-                  eachMessage: async () => {
+                  eachMessage: () => {
                     resolve()
                   }
                 })
@@ -221,7 +236,7 @@ describe('Plugin', () => {
                   [ERROR_STACK]: fakeError.stack,
                   'span.kind': 'consumer',
                   component: 'confluentinc-kafka-javascript',
-                  'messaging.destination.name': 'test-topic'
+                  'messaging.destination.name': testTopic
                 },
                 resource: testTopic,
                 error: 1,
@@ -344,7 +359,10 @@ describe('Plugin', () => {
             beforeEach(async () => {
               nativeConsumer = new Consumer({
                 'bootstrap.servers': '127.0.0.1:9092',
-                'group.id': 'test-group'
+                'group.id': groupId,
+                'enable.auto.commit': false,
+              }, {
+                'auto.offset.reset': 'earliest'
               })
 
               await new Promise((resolve, reject) => {
@@ -491,15 +509,15 @@ describe('Plugin', () => {
             tracer.use('confluentinc-kafka-javascript', { dsmEnabled: true })
             messages = [{ key: 'key1', value: 'test2' }]
             consumer = kafka.consumer({
-              kafkaJS: { groupId: 'test-group', fromBeginning: false }
+              kafkaJS: { groupId, fromBeginning: true }
             })
             await consumer.connect()
             await consumer.subscribe({ topic: testTopic })
           })
 
-          before(() => {
-            expectedProducerHash = getDsmPathwayHash(true, ENTRY_PARENT_HASH)
-            expectedConsumerHash = getDsmPathwayHash(false, expectedProducerHash)
+          beforeEach(() => {
+            expectedProducerHash = getDsmPathwayHash(testTopic, true, ENTRY_PARENT_HASH)
+            expectedConsumerHash = getDsmPathwayHash(testTopic, false, expectedProducerHash)
           })
 
           afterEach(async () => {
@@ -617,24 +635,22 @@ describe('Plugin', () => {
                     partition,
                     offset: Number(message.offset)
                   }
-                  // Signal that we've processed a message
                   messageProcessedResolve()
                 }
               })
 
-              consumerRunPromise.catch(() => {})
+              await consumerRunPromise
 
               // wait for the message to be processed before continuing
-              await sendMessages(kafka, testTopic, messages).then(
-                async () => await messageProcessedPromise
-              )
+              await sendMessages(kafka, testTopic, messages)
+              await messageProcessedPromise
 
               for (const call of setOffsetSpy.getCalls()) {
                 expect(call.args[0]).to.not.have.property('type', 'kafka_commit')
               }
 
               const newConsumer = kafka.consumer({
-                kafkaJS: { groupId: 'test-group', autoCommit: false }
+                kafkaJS: { groupId, fromBeginning: true, autoCommit: false }
               })
               await newConsumer.connect()
               await sendMessages(kafka, testTopic, [{ key: 'key1', value: 'test2' }])
@@ -648,12 +664,11 @@ describe('Plugin', () => {
 
               // Check our work
               const runArg = setOffsetSpy.lastCall.args[0]
-              expect(setOffsetSpy).to.be.calledOnce
               expect(runArg).to.have.property('offset', commitMeta.offset)
               expect(runArg).to.have.property('partition', commitMeta.partition)
               expect(runArg).to.have.property('topic', commitMeta.topic)
               expect(runArg).to.have.property('type', 'kafka_commit')
-              expect(runArg).to.have.property('consumer_group', 'test-group')
+              expect(runArg).to.have.property('consumer_group', groupId)
             })
 
             it('Should add backlog on producer response', async () => {

packages/dd-trace/test/setup/services/kafka.js

@@ -9,7 +9,7 @@ const kafka = new Kafka({
 })
 const admin = kafka.admin()
 const producer = kafka.producer()
-const consumer = kafka.consumer({ groupId: 'test-group' })
+const consumer = kafka.consumer({ groupId: 'setup-group' })
 const topic = 'test-topic'
 const messages = [{ key: 'setup', value: 'test' }]