Hangleton commented on code in PR #13240:
URL: https://github.com/apache/kafka/pull/13240#discussion_r1131047710
##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java:
##########
@@ -119,28 +131,52 @@ public boolean shouldClientThrottle(short version) {
return version >= 4;
}
+ public short version() {
+ return version;
+ }
+
public static class Builder {
OffsetCommitResponseData data = new OffsetCommitResponseData();
HashMap<String, OffsetCommitResponseTopic> byTopicName = new
HashMap<>();
+ private final TopicResolver topicResolver;
+ private final short version;
+
+ public Builder(TopicResolver topicResolver, short version) {
Review Comment:
Thanks for the answer. If I understand correctly, we would then have a
resolution of topic ids from topic-name-based persisted data, so this may not
prevent offsets from a topic to be provided as those of another topic with the
same name (defined at different point in time in the server)?
The resolution can be done in the group coordinator layer, assuming it has
access to the topic id resolved upstream by the request handler. Because we
want to preserve the same mapping used when request processing started, we need
to ensure the right ids are used within the adaptor's
`GroupCoordinator#commitOffsets` method(). Since the mapping returned from the
metadata cache depends on the snapshot used at the time the mapping is
requested, if the adaptor retrieves it from the metadata cache internally,
there is no guarantee the metadata is the same hence that the topic IDs
registered with the broker are the same.
This means that the topic ids need to be propagated from the request handler
(`KafkaApis`) to the coordinator adaptor somehow. Without a change in the
method and contract implemented by the coordinator, these ids could be
transferred via the `OffsetCommitRequestData` DTO directly, which means a
change in the API schema would be required prior to the change. Alternatively,
we may want to change the interface of the coordinator and change the signature
of the offset commit method to allow for the propagation of topic ids.
I may be missing the entire thing though?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
