New admin method for fetching group offsets for multiple topics

[nirga]

Fixes #989

This PR introduces a new method called fetchConsumerGroupOffsets that mimics the behavior fetchOffsets (which I now deprecated), but adds support for querying multiple topics at once, or omitting the topics parameter altogether and getting the group offsets in all consumed topics.

I chose to re-implement this method instead of changing fetchOffsets because this introduces an API breaking change: the result is now a list of topics and offsets (instead of just the offsets as it was in the original function).

[Nevon]

Isn't topics an array? in checks for the existence of a key in an object, not whether an item is in an array. Maybe you meant topics.includes(response.topic)?

[nirga]

Yeah it was just a draft PR and I'm not super familiar with javascript :)
I've removed this line altogether since I realized why would response not always include exactly the topics we requested?

[Nevon]

What's with the empty array here?

[nirga]

It needs to be passed to offsetFetch so that it will fetch all topics (it actually sets it in the protocol it sends to Kafka from what I've seen).

[Nevon]

I see two options:

1. Create a new method, as is done in this PR.
2. Change the signature of the old method so that it has two optional parameters, topic: string and topics: string[]. Internally, we could normalize the arguments so that topic: "foo" becomes topics: ["foo"], to avoid having to have a bunch of duplicated code. We could eventually deprecate the topic argument in favor of topics. If topic and topics are both defined, we could throw an error.

I think I'm in favor of option 2 more than option 1, because it limits the number of methods on the client and migrating from the old signature to the new is very simple. Do you see any big problem with going with that approach, @nirga?

@tulios / @JaapRood: Any input from you two?

[roeezolantz]

In the previous implementation if topic wasn't passed an exception was thrown

    if (!topic) {
      throw new KafkaJSNonRetriableError(`Invalid topic ${topic}`)
    }

but here you're setting topics's default value to pass your exception throwing ([]), so you're actually not validating that this param was sent here

[nirga]

mm yeah but I think it's kinda weird to force users to set topics=[] if they want to just fetch all of topics of the consumer instead of just calling it with fetchConsumerGroupOffsets({groupId}).

[nirga]

@Nevon I agree with you. I moved the changes back to fetchOffset and added logic go support passing either topic or topics (tweaking the returned value accordingly). I've also updated the docs.

[Nevon]

I'm not sure how to write the JSDoc for this, now that the interplay between topic and topics is more complicated.

Both topic and topics are optional. If neither is set, the default is for topics to get the value []. Do you know if there's any way to express that in JSDoc, @ankon? Not a big deal, but I just don't know.

[ankon]

Interesting, from the top of my head: I think this might be too much to express with JSDoc. There could be a way to do it in the TypeScript definition, and then you could do magic with @type here.

But it also might not be needed: If these are optional, then from the callers point that's all they need to know. The specific defaulting could just be considered an implementation detail.

I guess I would just go with

/**
 * Fetch offsets for a topic or multiple topics
 *
 * Note that `topics` will be defaulted to `[]` if neither `topic` nor `topics` is provided.
 * 
 * @param {object} options
 * @param {string} options.groupId
 * @param {string} [options.topic]
 * @param {string[]} [options.topics]
 * @param {boolean} [optins.resolveOffsets=false]
 */

I guess the reason you need both topic and topics is that the function is part of the API, i.e. you're providing backwards-compatibility. Could this not be achieved by adding a new function (say fetchTopicOffsets), and then eventually deprecate the old one?

---

Another approach for consideration, closer to what the .d.ts would do given your later comment :)

/**
 * @param {import("../../types").FetchOffsetsTopicOptions|import("../../types").FetchOffsetsTopicsOptions} options
 * @returns {Promise<...>}
 */

[nirga]

I added comments as @ankon suggested, inline in the parameter list as well.

[Nevon]

Github won't let me create a suggestion that goes outside of the lines of the diff, but there's an issue here where the return type of this function is different depending on the input, so we need to be a little more clever here. There's probably some fancy-pants thing you can do with conditional types, but we can also just use plain old method overloading:

interface FetchOffsetsBaseOptions {
  groupId: string,
  resolveOffsets?: boolean
}

export interface FetchOffsetsTopicOptions extends FetchOffsetsBaseOptions {
  topic: string
}

export interface FetchOffsetsTopicsOptions extends FetchOffsetsBaseOptions {
  topics?: string[]
}

export type Admin = {
  /**
   * @deprecated "topic: string" property now replaced by "topics: string[]"
   */
  fetchOffsets(options: FetchOffsetsTopicOptions): Promise<Array<PartitionOffset & { metadata: string | null }>>,
  fetchOffsets(options: FetchOffsetsTopicsOptions): Promise<Array<{ topic: string, partitions: Array<PartitionOffset & { metadata: string | null }>}>>,
}

This has a few benefits. The first is that the type system will prevent admin.fetchOffsets({ topic: 'foo', topics: ['bar'] }), because the topic and topics properties are mutually exclusive. The second is that it allows us to define the relationship between the input type and the return type. The return type is a bit of an abomination. If we didn't return metadata it could be simplified to just Promise<TopicOffsets[]> and Promise<PartitionOffset[]>, but alas, here we are.

[nirga]

Done. I've opted to keep the options parameter inline though, but moved the Array<PartitionOffset & { metadata: string | null }> to a new type. LMK if you prefer the options to be extracted to a type as well.

[Nevon]

Looks good overall. Just some fairly minor things to fix.

[nirga]

@Nevon Fixed everything. Thanks so much for your review and responsiveness!

[Nevon]

Great, thanks for your contribution! Sorry you had to wait so long for a review.

[tulios]

I see two options:

1. Create a new method, as is done in this PR.
2. Change the signature of the old method so that it has two optional parameters, topic: string and topics: string[]. Internally, we could normalize the arguments so that topic: "foo" becomes topics: ["foo"], to avoid having to have a bunch of duplicated code. We could eventually deprecate the topic argument in favor of topics. If topic and topics are both defined, we could throw an error.

I think I'm in favor of option 2 more than option 1, because it limits the number of methods on the client and migrating from the old signature to the new is very simple. Do you see any big problem with going with that approach, @nirga?

@tulios / @JaapRood: Any input from you two?

I guess you were too fast @Nevon 😅
Glad that the PR was merged.