chore: additionalGlobalConfig

Author: joebowbeer

packages/modules/kafka/src/kafka-container-7.test.ts

@@ -84,17 +84,18 @@ describe("KafkaContainer", { timeout: 240_000 }, () => {
         })
         .start();
 
-      await assertMessageProducedAndConsumed(container, {
-        brokers: [`${container.getHost()}:${container.getMappedPort(9096)}`],
-        sasl: {
-          username: "app-user",
-          password: "userPassword",
-          mechanism: "scram-sha-512",
-        },
-        ssl: {
-          ca: [fs.readFileSync(path.resolve(certificatesDir, "kafka.client.truststore.pem"))],
+      await assertMessageProducedAndConsumed(container,
+        {
+          brokers: [`${container.getHost()}:${container.getMappedPort(9096)}`],
+          ssl: true,
         },
-      });
+        {
+          "sasl.username": "app-user",
+          "sasl.password": "userPassword",
+          "sasl.mechanism": "scram-sha-512",
+          ssl_ca: [fs.readFileSync(path.resolve(certificatesDir, "kafka.client.truststore.pem"))],
+        }
+      );
       // }
     });
 
@@ -121,17 +122,18 @@ describe("KafkaContainer", { timeout: 240_000 }, () => {
         })
         .start();
 
-      await assertMessageProducedAndConsumed(container, {
-        brokers: [`${container.getHost()}:${container.getMappedPort(9096)}`],
-        sasl: {
-          username: "app-user",
-          password: "userPassword",
-          mechanism: "scram-sha-512",
-        },
-        ssl: {
-          ca: [fs.readFileSync(path.resolve(certificatesDir, "kafka.client.truststore.pem"))],
+      await assertMessageProducedAndConsumed(container,
+        {
+          brokers: [`${container.getHost()}:${container.getMappedPort(9096)}`],
+          ssl: true,
         },
-      });
+        {
+          "sasl.username": "app-user",
+          "sasl.password": "userPassword",
+          "sasl.mechanism": "scram-sha-512",
+          ssl_ca: [fs.readFileSync(path.resolve(certificatesDir, "kafka.client.truststore.pem"))],
+        }
+      );
     });
 
     it(`should connect within Docker network`, async () => {

packages/modules/kafka/src/kafka-container-latest.test.ts

@@ -55,17 +55,18 @@ describe("KafkaContainer", { timeout: 240_000 }, () => {
 
     await using container = await new KafkaContainer(IMAGE).withSaslSslListener(saslConfig).start();
 
-    await assertMessageProducedAndConsumed(container, {
-      brokers: [`${container.getHost()}:${container.getMappedPort(9096)}`],
-      sasl: {
-        username: "app-user",
-        password: "userPassword",
-        mechanism: "scram-sha-512",
-      },
-      ssl: {
-        ca: [fs.readFileSync(path.resolve(certificatesDir, "kafka.client.truststore.pem"))],
+    await assertMessageProducedAndConsumed(container,
+      {
+        brokers: [`${container.getHost()}:${container.getMappedPort(9096)}`],
+        ssl: true,
       },
-    });
+      {
+        "sasl.username": "app-user",
+        "sasl.password": "userPassword",
+        "sasl.mechanism": "scram-sha-512",
+        ssl_ca: [fs.readFileSync(path.resolve(certificatesDir, "kafka.client.truststore.pem"))],
+      }
+    );
     // }
   });
 

packages/modules/kafka/src/test-helper.ts

@@ -1,18 +1,20 @@
-import { KafkaJS } from "@confluentinc/kafka-javascript";
+import { GlobalConfig, KafkaJS } from "@confluentinc/kafka-javascript";
 import { StartedKafkaContainer } from "./kafka-container";
 
 // kafkaTestHelper {
 export async function assertMessageProducedAndConsumed(
   container: StartedKafkaContainer,
-  additionalConfig: Partial<KafkaJS.KafkaConfig> = {}
+  additionalKafkaConfig: Partial<KafkaJS.KafkaConfig> = {},
+  additionalGlobalConfig: Partial<GlobalConfig> = {}
 ) {
   const brokers = [`${container.getHost()}:${container.getMappedPort(9093)}`];
   const kafka = new KafkaJS.Kafka({
     kafkaJS: {
       logLevel: KafkaJS.logLevel.NOTHING,
       brokers: brokers,
-      ...additionalConfig,
+      ...additionalKafkaConfig,
     },
+    ...additionalGlobalConfig,
   });
 
   const producer = kafka.producer();