Add schema registry example

Author: milindl

MIGRATION.md

@@ -301,8 +301,12 @@ producerRun().then(consumerRun).catch(console.error);
       does not support `replicaAssignment`.
     * The `deleteTopics` method is fully supported.
 
+### Using the Schema Registry
 
-#### Error Handling
+In case you are using the Schema Registry client at `kafkajs/confluent-schema-registry`, you will not need to make any changes to the usage.
+An example is made available [here](../examples/kafkajs/sr.js).
+
+### Error Handling
 
   **Action**: Convert any checks based on `instanceof` and `error.name` or to error
             checks based on `error.code` or `error.type`.

examples/kafkajs/admin.js

@@ -1,5 +1,7 @@
-const { Kafka } = require('../..').KafkaJS
-//const { Kafka } = require('kafkajs')
+// require('kafkajs') is replaced with require('confluent-kafka-js').KafkaJS.
+// Since this example is within the package itself, we use '../..', but code
+// will typically use 'confluent-kafka-js'.
+const { Kafka } = require('../..').KafkaJS;
 
 async function adminStart() {
   const kafka = new Kafka({

examples/kafkajs/consumer.js

@@ -1,5 +1,7 @@
-const { Kafka } = require('../..').KafkaJS
-//const { Kafka } = require('kafkajs')
+// require('kafkajs') is replaced with require('confluent-kafka-js').KafkaJS.
+// Since this example is within the package itself, we use '../..', but code
+// will typically use 'confluent-kafka-js'.
+const { Kafka } = require('../..').KafkaJS;
 
 async function consumerStart() {
   let consumer;

examples/kafkajs/eos.js

@@ -1,5 +1,7 @@
-const { Kafka } = require('../..').KafkaJS
-//const { Kafka } = require('kafkajs')
+// require('kafkajs') is replaced with require('confluent-kafka-js').KafkaJS.
+// Since this example is within the package itself, we use '../..', but code
+// will typically use 'confluent-kafka-js'.
+const { Kafka } = require('../..').KafkaJS;
 
 async function eosStart() {
     const kafka = new Kafka({

examples/kafkajs/producer.js

@@ -1,5 +1,7 @@
-const { Kafka } = require('../..').KafkaJS
-//const { Kafka } = require('kafkajs')
+// require('kafkajs') is replaced with require('confluent-kafka-js').KafkaJS.
+// Since this example is within the package itself, we use '../..', but code
+// will typically use 'confluent-kafka-js'.
+const { Kafka } = require('../..').KafkaJS;
 
 async function producerStart() {
     const kafka = new Kafka({

examples/kafkajs/sr.js

@@ -0,0 +1,108 @@
+// require('kafkajs') is replaced with require('confluent-kafka-js').KafkaJS.
+// Since this example is within the package itself, we use '../..', but code
+// will typically use 'confluent-kafka-js'.
+const { Kafka } = require('../..').KafkaJS;
+
+// Note: The kafkajs/confluent-schema-registry will need to be installed separately to run this example,
+//       as it isn't a dependency of confluent-kafka-js.
+const { SchemaRegistry, SchemaType } = require('@kafkajs/confluent-schema-registry');
+
+const registry = new SchemaRegistry({ host: '<fill>' })
+const kafka = new Kafka({
+    brokers: ['<fill>'],
+    clientId: 'example-consumer',
+})
+let consumer = kafka.consumer({ groupId: 'test-group' , fromBeginning: true, } );
+let producer = kafka.producer();
+
+const schemaA = {
+    type: 'record',
+    namespace: 'test',
+    name: 'A',
+    fields: [
+        { name: 'id', type: 'int' },
+        { name: 'b', type: 'test.B' },
+    ],
+};
+
+const schemaB = {
+    type: 'record',
+    namespace: 'test',
+    name: 'B',
+    fields: [{ name: 'id', type: 'int' }],
+};
+
+const topicName = 'test-topic';
+
+const run = async () => {
+    // Register schemaB.
+    await registry.register(
+        {
+            type: SchemaType.AVRO,
+            schema: JSON.stringify(schemaB),
+        },
+        { subject: 'Avro:B' },
+    );
+    const response = await registry.api.Subject.latestVersion({ subject: 'Avro:B' });
+    const { version } = JSON.parse(response.responseData);
+
+    // Register schemaA, which references schemaB.
+    const { id } = await registry.register(
+        {
+            type: SchemaType.AVRO,
+            schema: JSON.stringify(schemaA),
+            references: [
+                {
+                    name: 'test.B',
+                    subject: 'Avro:B',
+                    version,
+                },
+            ],
+        },
+        { subject: 'Avro:A' },
+    )
+
+    // Produce a message with schemaA.
+    await producer.connect()
+    const outgoingMessage = {
+        key: 'key',
+        value: await registry.encode(id, { id: 1, b: { id: 2 } })
+    }
+    await producer.send({
+        topic: topicName,
+        messages: [outgoingMessage]
+    });
+    console.log("Producer sent its message.")
+    await producer.disconnect();
+    producer = null;
+
+    await consumer.connect()
+    await consumer.subscribe({ topic: topicName })
+
+    let messageRcvd = false;
+    await consumer.run({
+        eachMessage: async ({ message }) => {
+            const decodedMessage = {
+                ...message,
+                value: await registry.decode(message.value)
+            };
+            console.log("Consumer recieved message.\nBefore decoding: " + JSON.stringify(message) + "\nAfter decoding: " + JSON.stringify(decodedMessage));
+            messageRcvd = true;
+        },
+    });
+
+    // Wait around until we get a message, and then disconnect.
+    while (!messageRcvd) {
+        await new Promise((resolve) => setTimeout(resolve, 100));
+    }
+
+    await consumer.disconnect();
+    consumer = null;
+}
+
+run().catch (async e => {
+    console.error(e);
+    consumer && await consumer.disconnect();
+    producer && await producer.disconnect();
+    process.exit(1);
+})

lib/kafkajs/_consumer.js

@@ -967,6 +967,7 @@ class Consumer {
 
   /**
    * Disconnects and cleans up the consumer.
+   * @note This cannot be called from within `eachMessage` callback of `Consumer.run`.
    * @returns {Promise<void>} a promise that resolves when the consumer has disconnected.
    */
   async disconnect() {