Added listGroup method to admin interface

docs/Admin.md

@@ -322,3 +322,31 @@ Example response:
     throttleTime: 0,
 }
 ```
+
+## <a name="list-groups"></a> List groups
+
+List groups available on the broker.
+
+```javascript
+await admin.listGroups()
+```
+
+Example:
+
+```javascript
+const { ResourceTypes } = require('kafkajs')
+
+await admin.listGroups()
+```
+
+Example response:
+
+```javascript
+{
+    errorCode: 0,
+    throttleTime: 0,
+    groups: [
+        {groupId: 'testgroup', protocolType: 'consumer'}
+    ]
+}
+```

src/admin/__tests__/deleteTopics.spec.js

@@ -102,7 +102,9 @@ describe('Admin', () => {
       )
 
       expect(broker.deleteTopics).toHaveBeenCalledTimes(1)
-      expect(loggerInstance.error).toHaveBeenCalledWith(
+      expect(
+        loggerInstance.error
+      ).toHaveBeenCalledWith(
         'Could not delete topics, check if "delete.topic.enable" is set to "true" (the default value is "false") or increase the timeout',
         { error: 'The request timed out', retryCount: 0, retryTime: expect.any(Number) }
       )

src/admin/__tests__/listGroups.spec.js

@@ -0,0 +1,78 @@
+const createAdmin = require('../index')
+const createProducer = require('../../producer')
+const createConsumer = require('../../consumer')
+
+const {
+  createCluster,
+  newLogger,
+  createTopic,
+  createModPartitioner,
+  waitForMessages,
+  secureRandom,
+  waitForConsumerToJoinGroup,
+} = require('testHelpers')
+
+describe('Admin', () => {
+  describe('listGroups', () => {
+    let admin, topicName, groupId, cluster, consumer, producer
+
+    test('list groups', async () => {
+      topicName = `test-topic-${secureRandom()}`
+      groupId = `consumer-group-id-${secureRandom()}`
+
+      await createTopic({ topic: topicName })
+
+      cluster = createCluster()
+      producer = createProducer({
+        cluster,
+        createPartitioner: createModPartitioner,
+        logger: newLogger(),
+      })
+
+      consumer = createConsumer({
+        cluster,
+        groupId,
+        maxWaitTimeInMs: 100,
+        logger: newLogger(),
+      })
+
+      admin = createAdmin({ cluster: cluster, logger: newLogger() })
+
+      await admin.connect()
+
+      jest.spyOn(cluster, 'refreshMetadataIfNecessary')
+
+      await consumer.connect()
+      await producer.connect()
+      await consumer.subscribe({ topic: topicName, fromBeginning: true })
+
+      const messagesConsumed = []
+      consumer.run({ eachMessage: async event => messagesConsumed.push(event) })
+      await waitForConsumerToJoinGroup(consumer)
+
+      const messages = Array(1)
+        .fill()
+        .map(() => {
+          const value = secureRandom()
+          return { key: `key-${value}`, value: `value-${value}` }
+        })
+
+      await producer.send({ acks: 1, topic: topicName, messages })
+      await waitForMessages(messagesConsumed, { number: messages.length })
+
+      const listGroupResponse = await admin.listGroups()
+
+      let foundGroupId = ''
+
+      for (const group of listGroupResponse.groups) {
+        if (group.groupId === groupId) foundGroupId = groupId
+      }
+
+      expect(foundGroupId).toEqual(groupId)
+
+      await admin.disconnect()
+      await consumer.disconnect()
+      await producer.disconnect()
+    })
+  })
+})

src/admin/index.js

@@ -595,6 +595,39 @@ module.exports = ({
     }
   }
 
+  /**
+   * List groups in a broker
+   *
+   * @return {Promise<ListGroups>}
+   *
+   * @typedef {Object} ListGroups
+   * @property {number} errorCode
+   * @property {Array<ListGroup>} groups
+   *
+   * @typedef {Object} ListGroup
+   * @property {string} groupId
+   * @property {string} protocolType
+   */
+  const listGroups = async () => {
+    const retrier = createRetry(retry)
+
+    return retrier(async (bail, retryCount, retryTime) => {
+      try {
+        await cluster.refreshMetadata()
+        const broker = await cluster.findControllerBroker()
+        const response = await broker.listGroups()
+        return response
+      } catch (e) {
+        if (e.type === 'NOT_CONTROLLER') {
+          logger.warn('Could not list groups', { error: e.message, retryCount, retryTime })
+          throw e
+        }
+
+        bail(e)
+      }
+    })
+  }
+
   /**
    * @param {string} eventName
    * @param {Function} listener
@@ -637,5 +670,6 @@ module.exports = ({
     alterConfigs,
     on,
     logger: getLogger,
+    listGroups,
   }
 }

src/broker/__tests__/addPartitionsToTxn.spec.js

@@ -73,7 +73,10 @@ describe('Broker > AddPartitionsToTxn', () => {
       errors: [
         {
           topic: topicName,
-          partitionErrors: [{ errorCode: 0, partition: 1 }, { errorCode: 0, partition: 2 }],
+          partitionErrors: [
+            { errorCode: 0, partition: 1 },
+            { errorCode: 0, partition: 2 },
+          ],
         },
       ],
     })

src/broker/__tests__/deleteTopics.spec.js

@@ -42,9 +42,10 @@ describe('Broker > deleteTopics', () => {
 
     expect(response).toEqual({
       throttleTime: 0,
-      topicErrors: [{ topic: topicName1, errorCode: 0 }, { topic: topicName2, errorCode: 0 }].sort(
-        topicNameComparator
-      ),
+      topicErrors: [
+        { topic: topicName1, errorCode: 0 },
+        { topic: topicName2, errorCode: 0 },
+      ].sort(topicNameComparator),
     })
   })
 })

src/broker/__tests__/txnOffsetCommit.spec.js

@@ -92,7 +92,10 @@ describe('Broker > TxnOffsetCommit', () => {
       topics: [
         {
           topic: topicName,
-          partitions: [{ partition: 0, offset: 0 }, { partition: 1, offset: 0 }],
+          partitions: [
+            { partition: 0, offset: 0 },
+            { partition: 1, offset: 0 },
+          ],
         },
       ],
     })
@@ -102,7 +105,10 @@ describe('Broker > TxnOffsetCommit', () => {
       topics: [
         {
           topic: topicName,
-          partitions: [{ errorCode: 0, partition: 0 }, { errorCode: 0, partition: 1 }],
+          partitions: [
+            { errorCode: 0, partition: 0 },
+            { errorCode: 0, partition: 1 },
+          ],
         },
       ],
     })
@@ -117,7 +123,10 @@ describe('Broker > TxnOffsetCommit', () => {
       topics: [
         {
           topic: topicName,
-          partitions: [{ partition: 0, offset: 0 }, { partition: 1, offset: 0 }],
+          partitions: [
+            { partition: 0, offset: 0 },
+            { partition: 1, offset: 0 },
+          ],
         },
       ],
     })

src/broker/index.js

@@ -661,6 +661,16 @@ module.exports = class Broker {
     )
   }
 
+  /**
+   * Send request for list of groups
+   * @public
+   * @returns {Promise}
+   */
+  async listGroups() {
+    const listGroups = this.lookupRequest(apiKeys.ListGroups, requests.ListGroups)
+    return await this.connection.send(listGroups())
+  }
+
   /***
    * @private
    */

src/consumer/offsetManager/__tests__/initializeConsumerOffsets.spec.js

@@ -51,7 +51,10 @@ describe('Consumer > OffsetMananger > initializeConsumerOffsets', () => {
       },
       {
         topic: 'topic-name2',
-        partitions: [{ partition: 0, offset: '1' }, { partition: 1, offset: '2' }],
+        partitions: [
+          { partition: 0, offset: '1' },
+          { partition: 1, offset: '2' },
+        ],
       },
     ])
   })

src/producer/index.spec.js

@@ -557,7 +557,10 @@ describe('Producer', () => {
       const topicMessages = [
         {
           topic: topicName,
-          messages: [{ key: 'key-1', value: 'value-1' }, { key: 'key-2', value: 'value-2' }],
+          messages: [
+            { key: 'key-1', value: 'value-1' },
+            { key: 'key-2', value: 'value-2' },
+          ],
         },
         {
           topic: topicName,

src/protocol/requests/addPartitionsToTxn/v0/response.spec.js

@@ -9,7 +9,10 @@ describe('Protocol > Requests > AddPartitionsToTxn > v0', () => {
       errors: [
         {
           topic: 'test-topic-f6bab978bdca094688e3-37015-ca4f7ad4-5dcc-4bb9-9853-b1e4c4ed78a7',
-          partitionErrors: [{ errorCode: 0, partition: 1 }, { errorCode: 0, partition: 2 }],
+          partitionErrors: [
+            { errorCode: 0, partition: 1 },
+            { errorCode: 0, partition: 2 },
+          ],
         },
       ],
     })
@@ -23,7 +26,10 @@ describe('Protocol > Requests > AddPartitionsToTxn > v0', () => {
       errors: [
         {
           topic: 'test-topic',
-          partitionErrors: [{ errorCode: 0, partition: 1 }, { errorCode: 49, partition: 2 }],
+          partitionErrors: [
+            { errorCode: 0, partition: 1 },
+            { errorCode: 49, partition: 2 },
+          ],
         },
       ],
     }

src/protocol/requests/index.js

@@ -18,7 +18,7 @@ const requests = {
   LeaveGroup: require('./leaveGroup'),
   SyncGroup: require('./syncGroup'),
   DescribeGroups: require('./describeGroups'),
-  ListGroups: {},
+  ListGroups: require('./listGroups'),
   SaslHandshake: require('./saslHandshake'),
   ApiVersions: require('./apiVersions'),
   CreateTopics: require('./createTopics'),

src/protocol/requests/listGroups/index.js

@@ -0,0 +1,27 @@
+const versions = {
+  0: () => {
+    const request = require('./v0/request')
+    const response = require('./v0/response')
+    return { request: request(), response }
+  },
+  1: () => {
+    const request = require('./v1/request')
+    const response = require('./v1/response')
+    return { request: request(), response }
+  },
+  2: () => {
+    const request = require('./v2/request')
+    const response = require('./v2/response')
+    return { request: request(), response }
+  },
+  3: () => {
+    const request = require('./v3/request')
+    const response = require('./v3/response')
+    return { request: request(), response }
+  },
+}
+
+module.exports = {
+  versions: Object.keys(versions),
+  protocol: ({ version }) => versions[version],
+}

src/protocol/requests/listGroups/v0/request.js

@@ -0,0 +1,17 @@
+const Encoder = require('../../../encoder')
+const { ListGroups: apiKey } = require('../../apiKeys')
+
+/**
+ * ListGroups Request (Version: 0)
+ */
+
+/**
+ */
+module.exports = () => ({
+  apiKey,
+  apiVersion: 0,
+  apiName: 'ListGroups',
+  encode: async () => {
+    return new Encoder()
+  },
+})

src/protocol/requests/listGroups/v0/response.js

@@ -0,0 +1,41 @@
+const Decoder = require('../../../decoder')
+const { failure, createErrorFromCode } = require('../../../error')
+
+/**
+ * ListGroups Response (Version: 0) => error_code [groups]
+ *   throttle_time_ms => INT32
+ *   error_code => INT16
+ *   groups => group_id protocol_type
+ *     group_id => STRING
+ *     protocol_type => STRING
+ */
+
+const decodeGroup = decoder => ({
+  groupId: decoder.readString(),
+  protocolType: decoder.readString(),
+})
+
+const decode = async rawData => {
+  const decoder = new Decoder(rawData)
+  const errorCode = decoder.readInt16()
+  const groups = decoder.readArray(decodeGroup)
+
+  return {
+    errorCode,
+    groups,
+  }
+}
+
+const parse = async data => {
+  if (failure(data.errorCode)) {
+    throw createErrorFromCode(data.errorCode)
+  }
+
+  return data
+}
+
+module.exports = {
+  decodeGroup,
+  decode,
+  parse,
+}