Eos 201 list offsets v2 #209
Conversation
src/broker/index.js
Outdated
| const listOffsets = this.lookupRequest(apiKeys.ListOffsets, requests.ListOffsets) | ||
| return await this.connection.send(listOffsets({ replicaId, topics })) | ||
| return await this.connection.send(listOffsets({ replicaId, isolationLevel, topics })) |
There was a problem hiding this comment.
I would suggest doing the normalization between versions here, rather than in the cluster. We don't really have an established pattern for where to normalize. We have done it in the protocol in some place (unfortunately I can't remember where...), but I think we should try to keep our protocol true to the Kafka protocol. But we shouldn't let that bleed through too far, like the cluster, so I think the broker is the best place for it that we have at the moment.
There was a problem hiding this comment.
Ok. I do like currently that the broker is a thin wrapper to execute the protocols. I also agree that we should not handle the normalization in the protocols (it's done for fetch with record sets vs messages, and it's giving me some trouble now that we're working with transactions). I think ideally there'd be a wrapper around the broker that did normalization. But I'm fine moving into the broker for now, just didn't want to introduce a new pattern without consensus 😄
|
Looks good. 👏 Just don't forget that we need v1 as well. |
|
@Nevon Are you just generally noting that we'd like to add ListOffsets v1? Or are you saying that it's necessary for this PR? I may have time to add regardless. |
|
KafkaJS generally assumes that all the versions in between the ranges are implemented so that we will need v1. I just checked, and it's similar to v2, but we can do it in another PR if you want. |
Resolves #201
Implement protocol
ListOffsetsv2, so that we can now provide an isolation level. For now continue with the default of "read_comitted".The v2 response format is divergent from v0, as it returns a single offset rather than an array. AFAIK there's no pattern for normalizing responses (I'm unsure if that's even desirable), so I decided to handle it in an ad-hoc fashion within the cluster module.
Made the base branch
eosso that this can land with the transactional producer & consumer.