fetchOffsets = async ({ groupId, topics, resolveOffsets = false }) => {
    if (!groupId) {
      throw new KafkaJSNonRetriableError(`Invalid groupId ${groupId}`)
    }

    if (!topics) {
      topics = []
    }

    if (!Array.isArray(topics)) {
      throw new KafkaJSNonRetriableError('Expected topics array to be set')
    }

    const coordinator = await cluster.findGroupCoordinator({ groupId })
    const topicsToFetch = await Promise.all(
      topics.map(async topic => {
        const partitions = await findTopicPartitions(cluster, topic)
        const partitionsToFetch = partitions.map(partition => ({ partition }))
        return { topic, partitions: partitionsToFetch }
      })
    )
    let { responses: consumerOffsets } = await coordinator.offsetFetch({
      groupId,
      topics: topicsToFetch,
    })

    if (resolveOffsets) {
      consumerOffsets = await Promise.all(
        consumerOffsets.map(async ({ topic, partitions }) => {
          const indexedOffsets = indexByPartition(await fetchTopicOffsets(topic))
          const recalculatedPartitions = partitions.map(({ offset, partition, ...props }) => {
            let resolvedOffset = offset
            if (Number(offset) === EARLIEST_OFFSET) {
              resolvedOffset = indexedOffsets[partition].low
            }
            if (Number(offset) === LATEST_OFFSET) {
              resolvedOffset = indexedOffsets[partition].high
            }
            return {
              partition,
              offset: resolvedOffset,
              ...props,
            }
          })

          await setOffsets({ groupId, topic, partitions: recalculatedPartitions })

          return {
            topic,
            partitions: recalculatedPartitions,
          }
        })
      )
    }

    return consumerOffsets.map(({ topic, partitions }) => {
      const completePartitions = partitions.map(({ partition, offset, metadata }) => ({
        partition,
        offset,
        metadata: metadata || null,
      }))

      return { topic, partitions: completePartitions }
    })
  }
Fetch offsets for a topic or multiple topics

Note: set either topic or topics but not both.

@param {string} groupId
@param {string[]} topics - list of topics to fetch offsets for, defaults to `[]` which fetches all topics for `groupId`.
@param {boolean} [resolveOffsets=false]
@return {Promise}