Add examples and tests for non-promisified OAUTHBEARER token refresh callback

Author: milindl

e2e/oauthbearer_cb.spec.js

@@ -0,0 +1,82 @@
+/*
+ * confluent-kafka-javascript - Node.js wrapper  for RdKafka C/C++ library
+ *
+ * Copyright (c) 2024 Confluent, Inc.
+ *
+ * This software may be modified and distributed under the terms
+ * of the MIT license.  See the LICENSE.txt file for details.
+ */
+
+var Kafka = require('../');
+var t = require('assert');
+
+var eventListener = require('./listener');
+
+var kafkaBrokerList = process.env.KAFKA_HOST || 'localhost:9092';
+
+describe('Client with oauthbearer_cb', function () {
+  const oauthbearer_config = 'key=value';
+  let oauthbearer_cb_called = 0;
+  let oauthbearer_cb = function (config, cb) {
+    console.log("Called oauthbearer_cb with given config: " + config);
+    t.equal(config, oauthbearer_config);
+    oauthbearer_cb_called++;
+
+    // The broker is not expected to be configured for oauthbearer authentication.
+    // We just want to make sure that token refresh callback is triggered.
+    cb(new Error('oauthbearer_cb error'), null);
+  };
+
+  const commonConfig = {
+    'metadata.broker.list': kafkaBrokerList,
+    'debug': 'all',
+    'security.protocol': 'SASL_PLAINTEXT',
+    'sasl.mechanisms': 'OAUTHBEARER',
+    'oauthbearer_token_refresh_cb': oauthbearer_cb,
+    'sasl.oauthbearer.config': oauthbearer_config,
+  }
+
+  const checkClient = function (client, done) {
+    eventListener(client);
+
+    client.on('error', function (e) {
+      t.match(e.message, /oauthbearer_cb error/);
+    });
+
+    client.connect();
+
+    // We don't actually expect the connection to succeed, but we want to
+    // make sure that the oauthbearer_cb is called so give it a couple seconds.
+    setTimeout(() => {
+      client.disconnect();
+      client = null;
+      t.equal(oauthbearer_cb_called >= 1, true);
+      done();
+    }, 2000);
+  }
+
+  beforeEach(function (done) {
+    oauthbearer_cb_called = 0;
+    done();
+  });
+
+  it('as producer', function (done) {
+    let producer = new Kafka.Producer(commonConfig);
+    checkClient(producer, done);
+    producer = null;
+  }).timeout(2500);
+
+  it('as consumer', function (done) {
+    const config = Object.assign({ 'group.id': 'gid' }, commonConfig);
+    let consumer = new Kafka.KafkaConsumer(config);
+    checkClient(consumer, done);
+    consumer = null;
+  }).timeout(2500);
+
+  it('as admin', function (done) {
+    let admin = new Kafka.AdminClient.create(commonConfig);
+    checkClient(admin, done);
+    admin = null;
+  }).timeout(2500);
+
+});

examples/node-rdkafka/oauthbearer_callback_authentication/oauthbearer_callback_authentication.js

@@ -0,0 +1,82 @@
+const Kafka = require('@confluentinc/kafka-javascript');
+var jwt = require('jsonwebtoken');
+
+// This example uses the Producer for demonstration purposes.
+// It is the same whether you use a Consumer/AdminClient.
+
+function token_refresh(oauthbearer_config /* string - passed from config */, cb) {
+    console.log("Called token_refresh with given config: " + oauthbearer_config);
+    // At this point, we can use the information in the token, make
+    // some API calls, fetch something from a file...
+    // For the illustration, everything is hard-coded.
+    const principal = 'admin';
+    // In seconds - needed by jsonwebtoken library
+    const exp_seconds = Math.floor(Date.now() / 1000) + (60 * 60);
+    // In milliseconds - needed by kafka-javascript.
+    const exp_ms = exp_seconds * 1000;
+
+    // For illustration, we're not signing our JWT (algorithm: none).
+    // For production uses-cases, it should be signed.
+    const tokenValue = jwt.sign(
+        { 'sub': principal, exp: exp_seconds, 'scope': 'requiredScope' }, '', { algorithm: 'none' });
+
+    // SASL extensions can be passed as Map or key/value pairs in an object.
+    const extensions = {
+        traceId: '123'
+    };
+
+    // The callback is called with the new token, its lifetime, and the principal.
+    // The extensions are optional and may be omitted.
+    console.log("Finished token_refresh, triggering callback: with tokenValue: " +
+        tokenValue.slice(0, 10) + "..., lifetime: " + exp_ms +
+        ", principal: " + principal + ", extensions: " + JSON.stringify(extensions));
+    cb(
+        // If no token could be fetched or an error occurred, a new Error can be
+        // and passed as the first parameter and the second parameter omitted.
+        null,
+        { tokenValue, lifetime: exp_ms, principal, extensions });
+}
+
+function run() {
+    const producer = new Kafka.Producer({
+        'metadata.broker.list': 'localhost:60125',
+        'dr_cb': true,
+        // 'debug': 'all'
+
+        // Config important for OAUTHBEARER:
+        'security.protocol': 'SASL_PLAINTEXT',
+        'sasl.mechanisms': 'OAUTHBEARER',
+        'sasl.oauthbearer.config': 'someConfigPropertiesKey=value',
+        'oauthbearer_token_refresh_cb': token_refresh,
+    });
+
+    producer.connect();
+
+    producer.on('event.log', (event) => {
+        console.log(event);
+    });
+
+    producer.on('ready', () => {
+        console.log('Producer is ready!');
+        producer.setPollInterval(1000);
+        console.log("Producing message.");
+        producer.produce(
+            'topic',
+            null, // partition - let partitioner choose
+            Buffer.from('messageValue'),
+            'messageKey',
+        );
+    });
+
+    producer.on('error', (err) => {
+        console.error("Encountered error in producer: " + err.message);
+    });
+
+    producer.on('delivery-report', function (err, report) {
+        console.log('delivery-report: ' + JSON.stringify(report));
+        // since we just want to produce one message, close shop.
+        producer.disconnect();
+    });
+}
+
+run();

examples/node-rdkafka/oauthbearer_callback_authentication/package-lock.json

@@ -0,0 +1,15 @@
+{
+  "name": "oauthbearer_callback_authentication",
+  "version": "1.0.0",
+  "description": "",
+  "main": "oauthbearer_callback_authentication.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "author": "",
+  "license": "MIT",
+  "dependencies": {
+    "@confluentinc/kafka-javascript": "file:../../..",
+    "jsonwebtoken": "^9.0.2"
+  }
+}