Update comments and and examples

Author: milindl

examples/kafkajs/admin/dependent_admin.js

@@ -0,0 +1,78 @@
+// require('kafkajs') is replaced with require('@confluentinc/kafka-javascript').KafkaJS.
+const { Kafka } = require('@confluentinc/kafka-javascript').KafkaJS;
+
+async function adminFromConsumer() {
+    const kafka = new Kafka({
+      kafkaJS: {
+        brokers: ['localhost:9092'],
+      }
+    });
+
+    const consumer = kafka.consumer({
+        kafkaJS: {
+            groupId: 'test-group',
+            fromBeginning: true,
+        }
+    });
+
+    await consumer.connect();
+
+    // The consumer can be used as normal
+    await consumer.subscribe({ topic: 'test-topic' });
+    consumer.run({
+        eachMessage: async ({ topic, partition, message }) => {
+            console.log({
+                topic,
+                partition,
+                offset: message.offset,
+                key: message.key?.toString(),
+                value: message.value.toString(),
+            });
+        },
+    });
+
+    // And the same consumer can create an admin client - the consumer must have successfully
+    // been connected before the admin client can be created.
+    const admin = consumer.dependentAdmin();
+    await admin.connect();
+
+    // The admin client can be used until the consumer is connected.
+    const listTopicsResult = await admin.listTopics();
+    console.log(listTopicsResult);
+
+    await new Promise(resolve => setTimeout(resolve, 10000));
+
+    // Disconnect the consumer and admin clients in the correct order.
+    await admin.disconnect();
+    await consumer.disconnect();
+}
+
+async function adminFromProducer() {
+    const kafka = new Kafka({
+      kafkaJS: {
+        brokers: ['localhost:9092'],
+      }
+    });
+
+    const producer = kafka.producer({});
+
+    await producer.connect();
+
+    // The producer can be used as normal
+    await producer.send({ topic: 'test-topic', messages: [{ value: 'Hello!' }] });
+
+    // And the same producer can create an admin client - the producer must have successfully
+    // been connected before the admin client can be created.
+    const admin = producer.dependentAdmin();
+    await admin.connect();
+
+    // The admin client can be used until the producer is connected.
+    const listTopicsResult = await admin.listTopics();
+    console.log(listTopicsResult);
+
+    // Disconnect the producer and admin clients in the correct order.
+    await admin.disconnect();
+    await producer.disconnect();
+}
+
+adminFromProducer().then(() => adminFromConsumer()).catch(console.error);

examples/node-rdkafka/dependent_admin.js

@@ -0,0 +1,104 @@
+const Kafka = require('@confluentinc/kafka-javascript');
+const admin = require('../../lib/admin');
+
+const bootstrapServers = 'localhost:9092';
+
+function adminFromProducer() {
+    const producer = new Kafka.Producer({
+        'bootstrap.servers': bootstrapServers,
+        'dr_msg_cb': true,
+    });
+
+    const createAdminAndListTopics = () => {
+        // Create an admin client from the producer, which must be connected.
+        // Thus, this is called from the producer's 'ready' event.
+        const admin = Kafka.AdminClient.createFrom(producer);
+
+        // The admin client can be used until the producer is connected.
+        admin.listTopics((err, topics) => {
+            if (err) {
+                console.error(err);
+                return;
+            }
+            console.log("Topics: ", topics);
+            admin.disconnect();
+        });
+    };
+
+    producer.connect();
+
+    producer.on('ready', () => {
+        console.log("Producer is ready");
+        producer.setPollInterval(100);
+
+        // After the producer is ready, it can be used to create an admin client.
+        createAdminAndListTopics();
+
+        // It can also be used normally to produce messages.
+        producer.produce('test-topic', null, Buffer.from('Hello World!'), null, Date.now());
+    });
+
+    producer.on('event.error', (err) => {
+        console.error(err);
+    });
+
+    producer.on('delivery-report', (err, report) => {
+        console.log("Delivery report received:", report);
+    });
+
+    setTimeout(() => {
+        producer.disconnect();
+    }, 30000);
+}
+
+function adminFromConsumer() {
+    const consumer = new Kafka.KafkaConsumer({
+        'bootstrap.servers': bootstrapServers,
+        'group.id': 'test-group',
+        'auto.offset.reset': 'earliest',
+    });
+
+    const createAdminAndListTopics = () => {
+        // Create an admin client from the consumer, which must be connected.
+        // Thus, this is called from the consumer's 'ready' event.
+        const admin = Kafka.AdminClient.createFrom(consumer);
+
+        // The admin client can be used until the consumer is connected.
+        admin.listTopics((err, topics) => {
+            if (err) {
+                console.error(err);
+                return;
+            }
+            console.log("Topics: ", topics);
+            admin.disconnect();
+        });
+    };
+
+    consumer.connect();
+
+    consumer.on('ready', () => {
+        console.log("Consumer is ready");
+
+        // After the consumer is ready, it can be used to create an admin client.
+        createAdminAndListTopics();
+
+        // It can also be used normally to consume messages.
+        consumer.subscribe(['test-topic']);
+        consumer.consume();
+    });
+
+    consumer.on('data', (data) => {
+        console.log("Consumer:data", data);
+    });
+
+    consumer.on('event.error', (err) => {
+        console.error("Consumer:error", err);
+    });
+
+    setTimeout(() => {
+        consumer.disconnect();
+    }, 30000);
+}
+
+adminFromProducer();
+setTimeout(() => adminFromConsumer(), 35000);

lib/admin.js

@@ -84,8 +84,11 @@ function createAdminClient(conf, eventHandlers) {
  *
  * This is a factory method because it immediately starts an
  * active handle with the brokers.
+ *
+ * The producer or consumer being used must be connected.
+ * The client can only be used while the producer or consumer is connected.
+ * Logging and other events from this client will be emitted on the producer or consumer.
  * @param {import('../types/rdkafka').Producer | import('../types/rdkafka').KafkaConsumer} existingClient a producer or consumer to create the admin client from
- *                                                                                                        It must be connected.
  * @param {object} eventHandlers optional key value pairs of event handlers to attach to the client
  */
 function createAdminClientFrom(existingClient, eventHandlers) {

lib/kafkajs/_consumer.js

@@ -247,6 +247,10 @@ class Consumer {
 
   /**
    * Create a new admin client using the underlying connections of the consumer.
+   *
+   * The consumer must be connected before connecting the resulting admin client.
+   * The usage of the admin client is limited to the lifetime of the consumer.
+   * The consumer's logger is shared with the admin client.
    * @returns {import("../../types/kafkajs").Admin}
    */
   dependentAdmin() {

lib/kafkajs/_producer.js

@@ -93,7 +93,11 @@ class Producer {
   }
 
   /**
-   * Create a new admin client using the underlying connections of the consumer.
+   * Create a new admin client using the underlying connections of the producer.
+   *
+   * The producer must be connected before connecting the resulting admin client.
+   * The usage of the admin client is limited to the lifetime of the producer.
+   * The producer's logger is shared with the admin client.
    * @returns {import("../../types/kafkajs").Admin}
    */
   dependentAdmin() {