Feature Request: Fetch all partitions of a topic once

[lp247]

Hi, in our Kafka use cases where we return batches of messages over a REST API, we need to be able to fetch messages from all partitions for a specific topic only once, without having an infinite stream of messages. Seemingly, this library only provides ways to either create an infinite stream of messages with consumer.consume or fetch topic messages of a single node with consumer.fetch. Neither approach satisfies our requirement completely, and to implement a solution, we would have to duplicate a lot of library code.

I would like to request a feature that allows us to fetch all leaders of all topics very similarly to what is done in _read of theMessageStream class. I am willing to create a PR myself, but I would need some guidance on how to best implement it.

I was looking into the code and found two possible ways:

1. Extend consumer.consume with an option singleFetch or similar. Pass this option to the constructed MessageStream. Inside _read after fetching all the partition leaders, end the stream (by pushing null I guess?) if said option is set.
2. Create a new function, consumer.fetchAll, which is quite similar to consumer.fetch but includes checking and looping over all the partition leaders like in _read.

Let me know what you think.

[ShogunPanda]

Hi!
If I understand your use case correctly, I can think of two ways of having this today without any modification in the library:

for await (const message of stream) {
  // Your stuff

  break
}

or

stream.once('data', () => {
  // Your stuff

  stream.close()
})

In both cases, only one message is processed. The only important thing here is probably to disable autocommit when creating the stream.

Anyway, if you would like to send a PR for this, my suggestion would be to add an option to consume that in place gets propagated to MessageStream. After the first this.push, the MessageStream autocloses itself.

[patrykwegrzyn]

This might actually work nicely if consumer.assign (alternative to subscription) is in play — manual assignment offers greater control over the consumer and helps avoid junk consumer groups in some usecases.

[patrykwegrzyn]

Why not handle it with something like this instead? Feels like a much lighter lift compared to modifying the library internals

export const fetchAll = async (topic) => {
  const consumer = new Consumer({
    clientId: "batch-fetcher",
    groupId: "test-group",
    bootstrapBrokers: ["localhost:9992"],
    deserializers: stringDeserializers,
    autocommit: false,
  });
  const latest = await consumer.listOffsets({
    topics: [topic],
    timestamp: BigInt(ListOffsetTimestamps.LATEST),
  });

  const endOffsets = latest.get(topic);

  const stream = await consumer.consume({
    topics: [topic],
    mode: MessagesStreamModes.EARLIEST,
  });

  const seen = new Map();
  const messages = [];

  for await (const msg of stream) {
    messages.push(msg);
    seen.set(msg.partition, msg.offset);
    const done = endOffsets.every(
      (offset, partition) => seen.get(partition) + 1n === offset
    );

    if (done) {
      await stream.close();
      break;
    }
  }

  await consumer.close();
  return messages;
};


`