List consumer group offsets

CHANGELOG.md

@@ -1,10 +1,11 @@
-# confluent-kafka-javascript v0.3.1
+# confluent-kafka-javascript v0.4.0
 
-v0.3.1 is a limited availability maintenance release. It is supported for all usage.
+v0.4.0 is a limited availability feature release. It is supported for all usage.
 
 ## Enhancements
 
 1. Fixes an issue where headers were not passed correctly to the `eachBatch` callback (#130).
+2. Add support for an Admin API to list a consumer group's offsets (#49).
 
 
 # confluent-kafka-javascript v0.3.0

MIGRATION.md

@@ -331,6 +331,8 @@ The admin-client only has support for a limited subset of methods, with more to
   * The `describeGroups` method is supported with additional `timeout` and `includeAuthorizedOperations` options.
     A number of additional properties have been added to the returned groups.
   * The `deleteGroups` method is supported with an additional `timeout` option.
+  * The `fetchOffsets` method is supported with additional `timeout` and 
+  `requireStableOffsets` option but `resolveOffsets` option is not yet supported.
 
 ### Using the Schema Registry
 

examples/kafkajs/admin/fetch-offsets.js

@@ -0,0 +1,96 @@
+const { Kafka } = require('@confluentinc/kafka-javascript').KafkaJS;
+const { parseArgs } = require('node:util');
+
+async function fetchOffsets() {
+  const args = parseArgs({
+    allowPositionals: true,
+    options: {
+      'bootstrap-servers': {
+        type: 'string',
+        short: 'b',
+        default: 'localhost:9092',
+      },
+      'timeout': {
+        type: 'string',
+        short: 'm',
+        default: '5000',
+      },
+      'require-stable-offsets': {
+        type: 'boolean',
+        short: 'r',
+        default: false,
+      },
+    },
+  });
+
+  const {
+    'bootstrap-servers': bootstrapServers,
+    timeout,
+    'require-stable-offsets': requireStableOffsets,
+  } = args.values;
+
+  const [groupId, ...rest] = args.positionals;
+
+  if (!groupId) {
+    console.error('Group ID is required');
+    process.exit(1);
+  }
+
+  const kafka = new Kafka({
+    kafkaJS: {
+      brokers: [bootstrapServers],
+    },
+  });
+
+  const admin = kafka.admin();
+  await admin.connect();
+
+  try {
+    // Parse topics and partitions from remaining arguments
+    const topicInput = parseTopicsAndPartitions(rest);
+
+    // Fetch offsets for the specified consumer group
+    const offsets = await admin.fetchOffsets({
+      groupId: groupId,
+      topics: topicInput,
+      requireStableOffsets,
+      timeout: Number(timeout),
+    });
+
+    console.log(`Offsets for Consumer Group "${groupId}":`, JSON.stringify(offsets, null, 2));
+  } catch (err) {
+    console.error('Error fetching consumer group offsets:', err);
+  } finally {
+    await admin.disconnect();
+  }
+}
+
+// Helper function to parse topics and partitions from arguments
+function parseTopicsAndPartitions(args) {
+  if (args.length === 0) return undefined;
+
+  const topicInput = [];
+  let i = 0;
+
+  while (i < args.length) {
+    const topic = args[i];
+    i++;
+
+    const partitions = [];
+    while (i < args.length && !isNaN(args[i])) {
+      partitions.push(Number(args[i]));
+      i++;
+    }
+
+    // Add topic with partitions (or an empty array if no partitions specified)
+    if (partitions.length > 0) {
+      topicInput.push({ topic, partitions });
+    } else {
+      topicInput.push(topic); // Add as a string if no partitions specified
+    }
+  }
+
+  return topicInput;
+}
+
+fetchOffsets();

lib/admin.js

@@ -437,3 +437,50 @@ AdminClient.prototype.listTopics = function (options, cb) {
     }
   });
 };
+
+/**
+ * List offsets for topic partition(s) for consumer group(s).
+ *
+ * @param {import("../../types/rdkafka").ListGroupOffsets} listGroupOffsets - The list of groupId, partitions to fetch offsets for.
+ *                                                                            If partitions is null, list offsets for all partitions
+ *                                                                            in the group.
+ * @param {number?} options.timeout - The request timeout in milliseconds.
+ *                                    May be unset (default: 5000)
+ * @param {boolean?} options.requireStableOffsets - Whether broker should return stable offsets
+ *                                                  (transaction-committed). (default: false)
+ *
+ * @param {function} cb - The callback to be executed when finished.
+ */
+AdminClient.prototype.listConsumerGroupOffsets = function (listGroupOffsets, options, cb) {
+
+  if (!this._isConnected) {
+    throw new Error('Client is disconnected');
+  }
+
+  if (!listGroupOffsets[0].groupId) {
+    throw new Error('groupId must be provided');
+  }
+
+
+  if (!Object.hasOwn(options, 'timeout')) {
+    options.timeout = 5000;
+  }
+
+  if (!Object.hasOwn(options, 'requireStableOffsets')) {
+    options.requireStableOffsets = false;
+  }
+
+  this._client.listConsumerGroupOffsets(listGroupOffsets, options, function (err, offsets) {
+    if (err) {
+      if (cb) {
+        cb(LibrdKafkaError.create(err));
+      }
+      return;
+    }
+
+    if (cb) {
+      cb(null, offsets);
+    }
+  });
+};
+

lib/kafkajs/_admin.js

@@ -417,6 +417,123 @@ class Admin {
       });
     });
   }
+
+  /**
+   * Fetch the offsets for topic partition(s) for consumer group(s).
+   *
+   * @param {string} options.groupId - The group ID to fetch offsets for.
+   * @param {import("../../types/kafkajs").TopicInput} options.topics - The topics to fetch offsets for.
+   * @param {boolean} options.resolveOffsets - not yet implemented
+   * @param {number?} options.timeout - The request timeout in milliseconds.
+   *                                    May be unset (default: 5000)
+   * @param {boolean?} options.requireStableOffsets - Whether broker should return stable offsets
+   *                                                  (transaction-committed). (default: false)
+   *
+   * @returns {Promise<Array<topic: string, partitions: import('../../types/kafkajs').FetchOffsetsPartition>>}
+   */
+  async fetchOffsets(options = {}) {
+    if (this.#state !== AdminState.CONNECTED) {
+      throw new error.KafkaJSError("Admin client is not connected.", { code: error.ErrorCodes.ERR__STATE });
+    }
+
+    if (Object.hasOwn(options, "resolveOffsets")) {
+      throw new error.KafkaJSError("resolveOffsets is not yet implemented.", { code: error.ErrorCodes.ERR__NOT_IMPLEMENTED });
+    }
+
+    const { groupId, topics } = options;
+
+    if (!groupId) {
+      throw new error.KafkaJSError("groupId is required.", { code: error.ErrorCodes.ERR__INVALID_ARG });
+    }
+
+    let partitions = null;
+    let originalTopics = null;
+
+    /*
+      If the input is a list of topic string, the user expects us to
+      fetch offsets for all all partitions of all the input topics. In
+      librdkafka, we can only fetch offsets by topic partitions, or else,
+      we can fetch all of them. Thus, we must fetch offsets for all topic
+      partitions (by settings partitions to null) and filter by the topic strings later.
+    */
+    if (topics && Array.isArray(topics)) {
+      if (typeof topics[0] === 'string') {
+        originalTopics = topics;
+        partitions = null;
+      } else if (typeof topics[0] === 'object' && Array.isArray(topics[0].partitions)) {
+        partitions = topics.flatMap(topic => topic.partitions.map(partition => ({
+          topic: topic.topic,
+          partition
+        })));
+      } else {
+        throw new error.KafkaJSError("Invalid topics format.", { code: error.ErrorCodes.ERR__INVALID_ARG });
+      }
+    }
+
+    const listGroupOffsets = [{
+      groupId,
+      partitions
+    }];
+
+
+    return new Promise((resolve, reject) => {
+      this.#internalClient.listConsumerGroupOffsets(listGroupOffsets, options, (err, offsets) => {
+        if (err) {
+          reject(createKafkaJsErrorFromLibRdKafkaError(err));
+        } else {
+
+          /**
+           * Offsets is an array of group results, each containing a group id,
+           * an error and an array of partitions.
+           * We need to convert it to the required format of an array of topics, each
+           * containing an array of partitions.
+           */
+          const topicPartitionMap = new Map();
+
+          if (offsets.length !== 1) {
+            reject(new error.KafkaJSError("Unexpected number of group results."));
+            return;
+          }
+
+          const groupResult = offsets[0];
+
+          if (groupResult.error) {
+            reject(createKafkaJsErrorFromLibRdKafkaError(groupResult.error));
+            return;
+          }
+
+          // Traverse the partitions and group them by topic
+          groupResult.partitions.forEach(partitionObj => {
+            const { topic, partition, offset, leaderEpoch, metadata, error } = partitionObj;
+            const fetchOffsetsPartition = {
+              partition: partition,
+              offset: String(offset),
+              metadata: metadata || null,
+              leaderEpoch: leaderEpoch || null,
+              error: error || null
+            };
+
+            // Group partitions by topic
+            if (!topicPartitionMap.has(topic)) {
+              topicPartitionMap.set(topic, []);
+            }
+            topicPartitionMap.get(topic).push(fetchOffsetsPartition);
+          });
+
+          // Convert the map back to the desired array format
+          let convertedOffsets = Array.from(topicPartitionMap, ([topic, partitions]) => ({
+            topic,
+            partitions
+          }));
+
+          if (originalTopics !== null) {
+            convertedOffsets = convertedOffsets.filter(convertedOffset => originalTopics.includes(convertedOffset.topic));
+          }
+          resolve(convertedOffsets);
+        }
+      });
+    });
+  }
 }
 
 module.exports = {