Update examples for kafkaJs: block

Author: milindl

examples/kafkajs/consumer.js

@@ -1,16 +1,27 @@
-const { Kafka, logLevel } = require('../..').KafkaJS;
+const { Kafka } = require('../..').KafkaJS
 //const { Kafka } = require('kafkajs')
 
 async function consumerStart() {
-    let consumer;
-    var stopped = false;
+  let consumer;
+  var stopped = false;
 
-    const kafka = new Kafka({
-        brokers: ['localhost:9092'],
-    });
+  const kafka = new Kafka({
+    kafkaJs: {
+      brokers: ['localhost:9092'],
+      ssl: true,
+      connectionTimeout: 5000,
+      sasl: {
+        mechanism: 'plain',
+        username: '<fill>',
+        password: '<fill>',
+      },
+    }
+  });
 
-    consumer = kafka.consumer({
-      groupId: 'test-group22' + Math.random(),
+  consumer = kafka.consumer({
+    kafkaJs: {
+      groupId: 'test-group',
+      autoCommit: false,
       rebalanceListener: {
         onPartitionsAssigned: async (assignment) => {
           console.log(`Assigned partitions ${JSON.stringify(assignment)}`);
@@ -24,95 +35,90 @@ async function consumerStart() {
           }
         }
       },
-      rdKafka: {
-        globalConfig: {
-        'debug': 'conf',
-        'enable.auto.commit': false,
-        'auto.offset.reset': 'error'
-        },
-        topicConfig: {
-        },
-      }
-    });
+    },
+
+    /* Properties from librdkafka can also be used */
+    'auto.commit.interval.ms': 6000,
+  });
+
+  await consumer.connect();
+  console.log("Connected successfully");
 
-    await consumer.connect();
-    console.log("Connected successfully");
+  await consumer.subscribe({
+    topics: [
+      "topic2"
+    ]
+  })
 
-    await consumer.subscribe({
-      topics: [
-        "test-topic"
-      ]
-    })
+  // Batch consumer, commit and seek example
+  var batch = 0;
+  consumer.run({
+    eachMessage: async ({ topic, partition, message }) => {
+      console.log({
+        topic,
+        partition,
+        offset: message.offset,
+        key: message.key?.toString(),
+        value: message.value.toString(),
+      })
 
-    // Batch consumer, commit and seek example
-    var batch = 0;
-    consumer.run({
-      eachMessage: async ({ topic, partition, message }) => {
-        console.log({
+      if (++batch % 100 == 0) {
+        await consumer.seek({
           topic,
           partition,
-          offset: message.offset,
-          key: message.key?.toString(),
-          value: message.value.toString(),
-        })
-
-        if (++batch % 100 == 0) {
-          await consumer.seek({
-            topic,
-            partition,
-            offset: -2
-          });
-          await consumer.commitOffsets();
-          batch = 0;
-        }
-      },
-    });
+          offset: -2
+        });
+        await consumer.commitOffsets();
+        batch = 0;
+      }
+    },
+  });
 
-    // Pause/Resume example
-    const pauseResumeLoop = async () => {
-      let paused = false;
-      let ticks = 0;
-      while (!stopped) {
-        await new Promise((resolve) => setTimeout(resolve, 100));
-        if (stopped)
-          break;
+  // Pause/Resume example
+  const pauseResumeLoop = async () => {
+    let paused = false;
+    let ticks = 0;
+    while (!stopped) {
+      await new Promise((resolve) => setTimeout(resolve, 100));
+      if (stopped)
+        break;
 
-        ticks++;
-        if (ticks == 200) {
-          ticks = 0;
-          const assignment = consumer.assignment();
-          if (paused) {
-            console.log(`Resuming partitions ${JSON.stringify(assignment)}`)
-            consumer.resume(assignment);
-          } else {
-            console.log(`Pausing partitions ${JSON.stringify(assignment)}`);
-            consumer.pause(assignment);
-          }
-          paused = !paused;
+      ticks++;
+      if (ticks == 200) {
+        ticks = 0;
+        const assignment = consumer.assignment();
+        if (paused) {
+          console.log(`Resuming partitions ${JSON.stringify(assignment)}`)
+          consumer.resume(assignment);
+        } else {
+          console.log(`Pausing partitions ${JSON.stringify(assignment)}`);
+          consumer.pause(assignment);
         }
+        paused = !paused;
       }
     }
+  }
 
-    if (consumer.assignment) {
-      // KafkaJS doesn't have assignment()
-      pauseResumeLoop()
-    }
+  if (consumer.assignment()) {
+    // KafkaJS doesn't have assignment()
+    pauseResumeLoop()
+  }
 
-    // Disconnect example
-    const disconnect = () => {
-      process.off('SIGINT', disconnect);
-      process.off('SIGTERM', disconnect);
-      stopped = true;
-      consumer.commitOffsets()
+  // Disconnect example
+  const disconnect = () => {
+    process.off('SIGINT', disconnect);
+    process.off('SIGTERM', disconnect);
+    stopped = true;
+    consumer.commitOffsets()
       .finally(() =>
         consumer.disconnect()
       )
       .finally(() =>
         console.log("Disconnected successfully")
       );
-    }
-    process.on('SIGINT', disconnect);
-    process.on('SIGTERM', disconnect);
+  }
+  process.on('SIGINT', disconnect);
+  process.on('SIGTERM', disconnect);
 }
 
 consumerStart()

examples/kafkajs/eos.js

@@ -3,26 +3,28 @@ const { Kafka } = require('../..').KafkaJS
 
 async function eosStart() {
     const kafka = new Kafka({
-        brokers: ['<fill>'],
-        ssl: true,
-        sasl: {
-            mechanism: 'plain',
-            username: '<fill>',
-            password: '<fill>',
+        kafkaJs: {
+            brokers: ['<fill>'],
+            ssl: true,
+            sasl: {
+                mechanism: 'plain',
+                username: '<fill>',
+                password: '<fill>',
+            }
         }
     });
 
     const consumer = kafka.consumer({
-        groupId: 'groupId',
-        rdKafka: {
-            globalConfig: {
-                "enable.auto.commit": false,
-            }
-        },
+        kafkaJs: {
+            groupId: 'groupId',
+            autoCommit: false,
+        }
     });
 
     const producer = kafka.producer({
-        transactionalId: 'txid'
+        kafkaJs: {
+            transactionalId: 'txid'
+        }
     });
 
     await consumer.connect();
@@ -66,9 +68,7 @@ async function eosStart() {
                         {
                             topic,
                             partitions: [
-                                /* The message.offset indicates current offset, so we need to add 1 to it, since committed offset denotes
-                                 * the next offset to consume. */
-                                { partition, offset: message.offset + 1 },
+                                { partition, offset: message.offset },
                             ],
                         }
                     ],

examples/kafkajs/producer.js

@@ -3,12 +3,14 @@ const { Kafka } = require('../..').KafkaJS
 
 async function producerStart() {
     const kafka = new Kafka({
-        brokers: ['<fill>'],
-        ssl: true,
-        sasl: {
-            mechanism: 'plain',
-            username: '<fill>',
-            password: '<fill>',
+        kafkaJs: {
+            brokers: ['<fill>'],
+            ssl: true,
+            sasl: {
+                mechanism: 'plain',
+                username: '<fill>',
+                password: '<fill>',
+            },
         }
     });
 
@@ -19,16 +21,16 @@ async function producerStart() {
     console.log("Connected successfully");
 
     const res = []
-    for(let i = 0; i < 50; i++) {
+    for (let i = 0; i < 50; i++) {
         res.push(producer.send({
             topic: 'topic2',
             messages: [
-                {value: 'v222', partition: 0},
-                {value: 'v11', partition: 0, key: 'x'},
+                { value: 'v222', partition: 0 },
+                { value: 'v11', partition: 0, key: 'x' },
             ]
         }));
     }
-    await Promise.allSettled(res);
+    await Promise.all(res);
 
     await producer.disconnect();