You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "DL1231 (via GitHub)" <gi...@apache.org> on 2025/06/03 02:58:17 UTC
[PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
DL1231 opened a new pull request, #19885:
URL: https://github.com/apache/kafka/pull/19885
Introduce Topic ID in existing OffsetFetch API, the patch only contains the client side changes.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #19885:
URL: https://github.com/apache/kafka/pull/19885#issuecomment-3522805441
Hey @DL1231 , now that we merged the commit side let's try to push this one. Ping me for another review pass once you address the remaining comments. Thanks a lot, almost there!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #19885:
URL: https://github.com/apache/kafka/pull/19885#discussion_r2536014053
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -1124,22 +1123,24 @@ private void onSuccess(final long currentTimeMs,
var failedRequestRegistered = false;
for (var topic : response.topics()) {
+ // If the topic id is used, the topic name is empty in the response.
+ String topicName = topic.name().isEmpty() ? metadata.topicNames().get(topic.topicId()) : topic.name();
for (var partition : topic.partitions()) {
var tp = new TopicPartition(
- topic.name(),
+ topicName,
partition.partitionIndex()
);
var error = Errors.forCode(partition.errorCode());
- if (error != Errors.NONE) {
+ if (error != Errors.NONE || topicName == null) {
log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
Review Comment:
Thanks for pointing this out. I have updated the PR to separate the two failure scenarios.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #19885:
URL: https://github.com/apache/kafka/pull/19885#discussion_r2536009271
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -751,6 +751,59 @@ public void testOffsetFetchRequestEnsureDuplicatedRequestSucceed() {
assertEmptyPendingRequests(commitRequestManager);
}
+ @Test
+ public void testOffsetFetchRequestShouldSucceedWithTopicId() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+ Uuid topicId = Uuid.randomUuid();
+ when(metadata.topicIds()).thenReturn(Map.of("t1", topicId));
+ when(metadata.topicNames()).thenReturn(Map.of(topicId, "t1"));
+ Set<TopicPartition> partitions = new HashSet<>();
+ partitions.add(new TopicPartition("t1", 0));
+
+ List<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> futures = sendAndVerifyDuplicatedOffsetFetchRequests(
+ commitRequestManager,
+ partitions,
+ 2,
+ Errors.NONE,
+ true,
+ topicId);
+ futures.forEach(f -> {
+ assertTrue(f.isDone());
+ assertFalse(f.isCompletedExceptionally());
+ });
+ // expecting the buffers to be emptied after being completed successfully
+ commitRequestManager.poll(0);
+ assertEmptyPendingRequests(commitRequestManager);
+ }
+
+ @Test
+ public void testOffsetFetchRequestShouldFailedWithTopicIdWhenMetadataUnknownResponseTopicId() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+ Uuid topicId = Uuid.randomUuid();
+ when(metadata.topicIds()).thenReturn(Map.of("t1", topicId));
+ // Mock the scenario where the topicID from the response is not in the metadata.
+ when(metadata.topicNames()).thenReturn(Map.of());
+ Set<TopicPartition> partitions = new HashSet<>();
+ partitions.add(new TopicPartition("t1", 0));
+
+ List<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> futures = sendAndVerifyDuplicatedOffsetFetchRequests(
+ commitRequestManager,
+ partitions,
+ 2,
+ Errors.NONE,
+ true,
+ topicId);
+ futures.forEach(f -> {
+ assertTrue(f.isDone());
+ assertTrue(f.isCompletedExceptionally());
+ });
Review Comment:
Done.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19885:
URL: https://github.com/apache/kafka/pull/19885#discussion_r2491866942
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -983,38 +984,52 @@ public boolean sameRequest(final OffsetFetchRequestState request) {
}
public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
- List<OffsetFetchRequestData.OffsetFetchRequestTopics> topics = requestedPartitions.stream()
- .collect(Collectors.groupingBy(TopicPartition::topic))
- .entrySet()
- .stream()
- .map(entry -> new OffsetFetchRequestData.OffsetFetchRequestTopics()
+ Map<String, Uuid> topicIds = metadata.topicIds();
+ boolean canUseTopicIds = true;
+ List<OffsetFetchRequestData.OffsetFetchRequestTopics> topics = new ArrayList<>();
+ Map<String, List<TopicPartition>> tps = requestedPartitions.stream().collect(Collectors.groupingBy(TopicPartition::topic));
+ for (Map.Entry<String, List<TopicPartition>> entry : tps.entrySet()) {
+ String topic = entry.getKey();
+ Uuid topicId = topicIds.getOrDefault(topic, Uuid.ZERO_UUID);
+ if (Uuid.ZERO_UUID.equals(topicId)) {
+ canUseTopicIds = false;
+ }
+ topics.add(new OffsetFetchRequestData.OffsetFetchRequestTopics()
.setName(entry.getKey())
+ .setTopicId(topicId)
.setPartitionIndexes(entry.getValue().stream()
.map(TopicPartition::partition)
- .collect(Collectors.toList())))
- .collect(Collectors.toList());
+ .collect(Collectors.toList())));
+ }
+ boolean finalCanUseTopicIds = canUseTopicIds;
OffsetFetchRequest.Builder builder = memberInfo.memberEpoch
- .map(epoch -> OffsetFetchRequest.Builder.forTopicNames(
- new OffsetFetchRequestData()
+ .map(epoch -> {
+ OffsetFetchRequestData data = new OffsetFetchRequestData()
.setRequireStable(true)
.setGroups(List.of(
new OffsetFetchRequestData.OffsetFetchRequestGroup()
.setGroupId(groupId)
.setMemberId(memberInfo.memberId)
.setMemberEpoch(epoch)
- .setTopics(topics))),
- throwOnFetchStableOffsetUnsupported))
+ .setTopics(topics)));
+ return finalCanUseTopicIds
+ ? OffsetFetchRequest.Builder.forTopicIdsOrNames(data, throwOnFetchStableOffsetUnsupported, true)
+ : OffsetFetchRequest.Builder.forTopicNames(data, throwOnFetchStableOffsetUnsupported);
+ })
// Building request without passing member ID/epoch to leave the logic to choose
// default values when not present on the request builder.
- .orElseGet(() -> OffsetFetchRequest.Builder.forTopicNames(
- new OffsetFetchRequestData()
+ .orElseGet(() -> {
+ OffsetFetchRequestData data = new OffsetFetchRequestData()
.setRequireStable(true)
.setGroups(List.of(
new OffsetFetchRequestData.OffsetFetchRequestGroup()
.setGroupId(groupId)
- .setTopics(topics))),
- throwOnFetchStableOffsetUnsupported));
+ .setTopics(topics)));
+ return finalCanUseTopicIds
+ ? OffsetFetchRequest.Builder.forTopicIdsOrNames(data, throwOnFetchStableOffsetUnsupported, true)
+ : OffsetFetchRequest.Builder.forTopicNames(data, throwOnFetchStableOffsetUnsupported);
+ });
Review Comment:
this whole section is getting too complicated, let's try to simplify.
We just one to create the builder, the only diff seems to be around the group data, that should include epoch or not. So what about extracting that bit and leaving the rest as a common code, no dups? Something along these lines:
```
OffsetFetchRequestData.OffsetFetchRequestGroup groupData = new OffsetFetchRequestData.OffsetFetchRequestGroup()
.setGroupId(groupId)
.setTopics(topics);
if (memberInfo.memberEpoch.isPresent()) {
groupData = groupData.setMemberId(memberInfo.memberId)
.setMemberEpoch(memberInfo.memberEpoch.get());
}
OffsetFetchRequestData data = new OffsetFetchRequestData()
.setRequireStable(true)
.setGroups(List.of(groupData));
OffsetFetchRequest.Builder builder = finalCanUseTopicIds
? OffsetFetchRequest.Builder.forTopicIdsOrNames(data, throwOnFetchStableOffsetUnsupported, true)
: OffsetFetchRequest.Builder.forTopicNames(data, throwOnFetchStableOffsetUnsupported);
```
Would that work? Not sure if I'm missing any other diff
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19885:
URL: https://github.com/apache/kafka/pull/19885#discussion_r2491866942
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -983,38 +984,52 @@ public boolean sameRequest(final OffsetFetchRequestState request) {
}
public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
- List<OffsetFetchRequestData.OffsetFetchRequestTopics> topics = requestedPartitions.stream()
- .collect(Collectors.groupingBy(TopicPartition::topic))
- .entrySet()
- .stream()
- .map(entry -> new OffsetFetchRequestData.OffsetFetchRequestTopics()
+ Map<String, Uuid> topicIds = metadata.topicIds();
+ boolean canUseTopicIds = true;
+ List<OffsetFetchRequestData.OffsetFetchRequestTopics> topics = new ArrayList<>();
+ Map<String, List<TopicPartition>> tps = requestedPartitions.stream().collect(Collectors.groupingBy(TopicPartition::topic));
+ for (Map.Entry<String, List<TopicPartition>> entry : tps.entrySet()) {
+ String topic = entry.getKey();
+ Uuid topicId = topicIds.getOrDefault(topic, Uuid.ZERO_UUID);
+ if (Uuid.ZERO_UUID.equals(topicId)) {
+ canUseTopicIds = false;
+ }
+ topics.add(new OffsetFetchRequestData.OffsetFetchRequestTopics()
.setName(entry.getKey())
+ .setTopicId(topicId)
.setPartitionIndexes(entry.getValue().stream()
.map(TopicPartition::partition)
- .collect(Collectors.toList())))
- .collect(Collectors.toList());
+ .collect(Collectors.toList())));
+ }
+ boolean finalCanUseTopicIds = canUseTopicIds;
OffsetFetchRequest.Builder builder = memberInfo.memberEpoch
- .map(epoch -> OffsetFetchRequest.Builder.forTopicNames(
- new OffsetFetchRequestData()
+ .map(epoch -> {
+ OffsetFetchRequestData data = new OffsetFetchRequestData()
.setRequireStable(true)
.setGroups(List.of(
new OffsetFetchRequestData.OffsetFetchRequestGroup()
.setGroupId(groupId)
.setMemberId(memberInfo.memberId)
.setMemberEpoch(epoch)
- .setTopics(topics))),
- throwOnFetchStableOffsetUnsupported))
+ .setTopics(topics)));
+ return finalCanUseTopicIds
+ ? OffsetFetchRequest.Builder.forTopicIdsOrNames(data, throwOnFetchStableOffsetUnsupported, true)
+ : OffsetFetchRequest.Builder.forTopicNames(data, throwOnFetchStableOffsetUnsupported);
+ })
// Building request without passing member ID/epoch to leave the logic to choose
// default values when not present on the request builder.
- .orElseGet(() -> OffsetFetchRequest.Builder.forTopicNames(
- new OffsetFetchRequestData()
+ .orElseGet(() -> {
+ OffsetFetchRequestData data = new OffsetFetchRequestData()
.setRequireStable(true)
.setGroups(List.of(
new OffsetFetchRequestData.OffsetFetchRequestGroup()
.setGroupId(groupId)
- .setTopics(topics))),
- throwOnFetchStableOffsetUnsupported));
+ .setTopics(topics)));
+ return finalCanUseTopicIds
+ ? OffsetFetchRequest.Builder.forTopicIdsOrNames(data, throwOnFetchStableOffsetUnsupported, true)
+ : OffsetFetchRequest.Builder.forTopicNames(data, throwOnFetchStableOffsetUnsupported);
+ });
Review Comment:
this whole section is getting too complicated, let's try to simplify.
We just want to create the builder, the only diff seems to be around the group data (with epoch or not). So what about extracting that bit and leaving the rest as a common code, no dups? Something along these lines:
```
OffsetFetchRequestData.OffsetFetchRequestGroup groupData = new OffsetFetchRequestData.OffsetFetchRequestGroup()
.setGroupId(groupId)
.setTopics(topics);
if (memberInfo.memberEpoch.isPresent()) {
groupData = groupData.setMemberId(memberInfo.memberId)
.setMemberEpoch(memberInfo.memberEpoch.get());
}
OffsetFetchRequestData data = new OffsetFetchRequestData()
.setRequireStable(true)
.setGroups(List.of(groupData));
OffsetFetchRequest.Builder builder = finalCanUseTopicIds
? OffsetFetchRequest.Builder.forTopicIdsOrNames(data, throwOnFetchStableOffsetUnsupported, true)
: OffsetFetchRequest.Builder.forTopicNames(data, throwOnFetchStableOffsetUnsupported);
```
Would that work? Not sure if I'm missing any other diff
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #19885:
URL: https://github.com/apache/kafka/pull/19885#issuecomment-3466005679
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19885:
URL: https://github.com/apache/kafka/pull/19885#discussion_r2528122589
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -1124,9 +1123,11 @@ private void onSuccess(final long currentTimeMs,
var failedRequestRegistered = false;
for (var topic : response.topics()) {
+ // If the topic id is used, the topic name is empty in the response.
+ String topicName = topic.name().isEmpty() ? metadata.topicNames().get(topic.topicId()) : topic.name();
Review Comment:
after this, we could still have empty `topicName` if the `topicID` from the response is not in the metadata anymore right? And in this case of OffsetFetch that would be problematic because we do need a topic name to return along with the offsets.
One option that comes to mind is to fail in the same way we do when we get `UNKNOWN_TOPIC_ID` , failing the future with "topic does not exist" (ln 1143). wdyt?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #19885:
URL: https://github.com/apache/kafka/pull/19885#discussion_r2531313841
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -1124,9 +1123,11 @@ private void onSuccess(final long currentTimeMs,
var failedRequestRegistered = false;
for (var topic : response.topics()) {
+ // If the topic id is used, the topic name is empty in the response.
+ String topicName = topic.name().isEmpty() ? metadata.topicNames().get(topic.topicId()) : topic.name();
Review Comment:
Good catch! I've updated the PR, please take another look when you get a chance, thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #19885:
URL: https://github.com/apache/kafka/pull/19885#issuecomment-2967755637
Hey @DL1231, just FYI, very busy now with the last fixes for 4.1 but I'll circle back to review this asap (and to the OffsetCommit one too). Thanks!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19885:
URL: https://github.com/apache/kafka/pull/19885#discussion_r2491866942
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -983,38 +984,52 @@ public boolean sameRequest(final OffsetFetchRequestState request) {
}
public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
- List<OffsetFetchRequestData.OffsetFetchRequestTopics> topics = requestedPartitions.stream()
- .collect(Collectors.groupingBy(TopicPartition::topic))
- .entrySet()
- .stream()
- .map(entry -> new OffsetFetchRequestData.OffsetFetchRequestTopics()
+ Map<String, Uuid> topicIds = metadata.topicIds();
+ boolean canUseTopicIds = true;
+ List<OffsetFetchRequestData.OffsetFetchRequestTopics> topics = new ArrayList<>();
+ Map<String, List<TopicPartition>> tps = requestedPartitions.stream().collect(Collectors.groupingBy(TopicPartition::topic));
+ for (Map.Entry<String, List<TopicPartition>> entry : tps.entrySet()) {
+ String topic = entry.getKey();
+ Uuid topicId = topicIds.getOrDefault(topic, Uuid.ZERO_UUID);
+ if (Uuid.ZERO_UUID.equals(topicId)) {
+ canUseTopicIds = false;
+ }
+ topics.add(new OffsetFetchRequestData.OffsetFetchRequestTopics()
.setName(entry.getKey())
+ .setTopicId(topicId)
.setPartitionIndexes(entry.getValue().stream()
.map(TopicPartition::partition)
- .collect(Collectors.toList())))
- .collect(Collectors.toList());
+ .collect(Collectors.toList())));
+ }
+ boolean finalCanUseTopicIds = canUseTopicIds;
OffsetFetchRequest.Builder builder = memberInfo.memberEpoch
- .map(epoch -> OffsetFetchRequest.Builder.forTopicNames(
- new OffsetFetchRequestData()
+ .map(epoch -> {
+ OffsetFetchRequestData data = new OffsetFetchRequestData()
.setRequireStable(true)
.setGroups(List.of(
new OffsetFetchRequestData.OffsetFetchRequestGroup()
.setGroupId(groupId)
.setMemberId(memberInfo.memberId)
.setMemberEpoch(epoch)
- .setTopics(topics))),
- throwOnFetchStableOffsetUnsupported))
+ .setTopics(topics)));
+ return finalCanUseTopicIds
+ ? OffsetFetchRequest.Builder.forTopicIdsOrNames(data, throwOnFetchStableOffsetUnsupported, true)
+ : OffsetFetchRequest.Builder.forTopicNames(data, throwOnFetchStableOffsetUnsupported);
+ })
// Building request without passing member ID/epoch to leave the logic to choose
// default values when not present on the request builder.
- .orElseGet(() -> OffsetFetchRequest.Builder.forTopicNames(
- new OffsetFetchRequestData()
+ .orElseGet(() -> {
+ OffsetFetchRequestData data = new OffsetFetchRequestData()
.setRequireStable(true)
.setGroups(List.of(
new OffsetFetchRequestData.OffsetFetchRequestGroup()
.setGroupId(groupId)
- .setTopics(topics))),
- throwOnFetchStableOffsetUnsupported));
+ .setTopics(topics)));
+ return finalCanUseTopicIds
+ ? OffsetFetchRequest.Builder.forTopicIdsOrNames(data, throwOnFetchStableOffsetUnsupported, true)
+ : OffsetFetchRequest.Builder.forTopicNames(data, throwOnFetchStableOffsetUnsupported);
+ });
Review Comment:
this whole section is getting too complicated, let's try to simplify.
We just want to create the builder, the only diff seems to be around the group data, that should include epoch or not. So what about extracting that bit and leaving the rest as a common code, no dups? Something along these lines:
```
OffsetFetchRequestData.OffsetFetchRequestGroup groupData = new OffsetFetchRequestData.OffsetFetchRequestGroup()
.setGroupId(groupId)
.setTopics(topics);
if (memberInfo.memberEpoch.isPresent()) {
groupData = groupData.setMemberId(memberInfo.memberId)
.setMemberEpoch(memberInfo.memberEpoch.get());
}
OffsetFetchRequestData data = new OffsetFetchRequestData()
.setRequireStable(true)
.setGroups(List.of(groupData));
OffsetFetchRequest.Builder builder = finalCanUseTopicIds
? OffsetFetchRequest.Builder.forTopicIdsOrNames(data, throwOnFetchStableOffsetUnsupported, true)
: OffsetFetchRequest.Builder.forTopicNames(data, throwOnFetchStableOffsetUnsupported);
```
Would that work? Not sure if I'm missing any other diff
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on PR #19885:
URL: https://github.com/apache/kafka/pull/19885#issuecomment-2968692406
Thanks for the heads-up, appreciate you circling back when you can. Good luck with the final pushes! @lianetm
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19885:
URL: https://github.com/apache/kafka/pull/19885#discussion_r2538567773
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -751,6 +751,56 @@ public void testOffsetFetchRequestEnsureDuplicatedRequestSucceed() {
assertEmptyPendingRequests(commitRequestManager);
}
+ @Test
+ public void testOffsetFetchRequestShouldSucceedWithTopicId() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+ Uuid topicId = Uuid.randomUuid();
+ when(metadata.topicIds()).thenReturn(Map.of("t1", topicId));
+ when(metadata.topicNames()).thenReturn(Map.of(topicId, "t1"));
+ Set<TopicPartition> partitions = new HashSet<>();
+ partitions.add(new TopicPartition("t1", 0));
+
+ List<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> futures = sendAndVerifyDuplicatedOffsetFetchRequests(
+ commitRequestManager,
+ partitions,
+ 2,
+ Errors.NONE,
+ true,
+ topicId);
+ futures.forEach(f -> {
+ assertTrue(f.isDone());
+ assertFalse(f.isCompletedExceptionally());
+ });
+ // expecting the buffers to be emptied after being completed successfully
+ commitRequestManager.poll(0);
+ assertEmptyPendingRequests(commitRequestManager);
+ }
+
+ @Test
+ public void testOffsetFetchRequestShouldFailWithTopicIdWhenMetadataUnknownResponseTopicId() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+ Uuid topicId = Uuid.randomUuid();
+ when(metadata.topicIds()).thenReturn(Map.of("t1", topicId));
+ // Mock the scenario where the topicID from the response is not in the metadata.
+ when(metadata.topicNames()).thenReturn(Map.of());
+ Set<TopicPartition> partitions = new HashSet<>();
+ partitions.add(new TopicPartition("t1", 0));
+
+ List<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> futures = sendAndVerifyDuplicatedOffsetFetchRequests(
+ commitRequestManager,
+ partitions,
+ 1,
+ Errors.UNKNOWN_TOPIC_OR_PARTITION,
Review Comment:
this generates a response with an error at the group level, which is not exactly the scenario we want to test.
We are after the case where we use topic IDs, the response has NONE error, but topic ID is not in metadata anymore. On fetch that is expected to fail with `KafkaException` (not retriable, sorry I mentioned testRetriable before, that's the case for commit but not for fetch). So we just need NONE and assert that the future completes exceptionally with `KafkaException` to cover our scenario I expect.
https://github.com/apache/kafka/blob/0491e1124eaa163a7ed7c139ab477c9f4a7b71c9/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java#L1142
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #19885:
URL: https://github.com/apache/kafka/pull/19885#discussion_r2538729446
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -751,6 +751,56 @@ public void testOffsetFetchRequestEnsureDuplicatedRequestSucceed() {
assertEmptyPendingRequests(commitRequestManager);
}
+ @Test
+ public void testOffsetFetchRequestShouldSucceedWithTopicId() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+ Uuid topicId = Uuid.randomUuid();
+ when(metadata.topicIds()).thenReturn(Map.of("t1", topicId));
+ when(metadata.topicNames()).thenReturn(Map.of(topicId, "t1"));
+ Set<TopicPartition> partitions = new HashSet<>();
+ partitions.add(new TopicPartition("t1", 0));
+
+ List<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> futures = sendAndVerifyDuplicatedOffsetFetchRequests(
+ commitRequestManager,
+ partitions,
+ 2,
+ Errors.NONE,
+ true,
+ topicId);
+ futures.forEach(f -> {
+ assertTrue(f.isDone());
+ assertFalse(f.isCompletedExceptionally());
+ });
+ // expecting the buffers to be emptied after being completed successfully
+ commitRequestManager.poll(0);
+ assertEmptyPendingRequests(commitRequestManager);
+ }
+
+ @Test
+ public void testOffsetFetchRequestShouldFailWithTopicIdWhenMetadataUnknownResponseTopicId() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+ Uuid topicId = Uuid.randomUuid();
+ when(metadata.topicIds()).thenReturn(Map.of("t1", topicId));
+ // Mock the scenario where the topicID from the response is not in the metadata.
+ when(metadata.topicNames()).thenReturn(Map.of());
+ Set<TopicPartition> partitions = new HashSet<>();
+ partitions.add(new TopicPartition("t1", 0));
+
+ List<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> futures = sendAndVerifyDuplicatedOffsetFetchRequests(
+ commitRequestManager,
+ partitions,
+ 1,
+ Errors.UNKNOWN_TOPIC_OR_PARTITION,
Review Comment:
Thanks for the review. I've updated the PR. PTAL.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19885:
URL: https://github.com/apache/kafka/pull/19885#discussion_r2491976458
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -694,6 +694,29 @@ public void testOffsetFetchRequestEnsureDuplicatedRequestSucceed() {
assertEmptyPendingRequests(commitRequestManager);
}
+ @Test
+ public void testOffsetFetchRequestShouldSucceedWithTopicId() {
Review Comment:
this is nice, but we're still missing coverage for the core change right? (request builder using topic IDs if available)
I expect we actually need to look into the generated request returned by poll, and validate the 2 core cases:
- request generated with v<10 and topic names (if topic IDs not in metadata)
- request generated with v >=10 and topic IDs (if topic IDs in metadata)
Makes sense?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19885:
URL: https://github.com/apache/kafka/pull/19885#discussion_r2534735107
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -1124,22 +1123,24 @@ private void onSuccess(final long currentTimeMs,
var failedRequestRegistered = false;
for (var topic : response.topics()) {
+ // If the topic id is used, the topic name is empty in the response.
+ String topicName = topic.name().isEmpty() ? metadata.topicNames().get(topic.topicId()) : topic.name();
for (var partition : topic.partitions()) {
var tp = new TopicPartition(
- topic.name(),
+ topicName,
partition.partitionIndex()
);
var error = Errors.forCode(partition.errorCode());
- if (error != Errors.NONE) {
+ if (error != Errors.NONE || topicName == null) {
log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
Review Comment:
this log will show weird if the case is error=none and topicName=null (tp without name and empty error message I expect). Should we differentiate?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -751,6 +751,59 @@ public void testOffsetFetchRequestEnsureDuplicatedRequestSucceed() {
assertEmptyPendingRequests(commitRequestManager);
}
+ @Test
+ public void testOffsetFetchRequestShouldSucceedWithTopicId() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+ Uuid topicId = Uuid.randomUuid();
+ when(metadata.topicIds()).thenReturn(Map.of("t1", topicId));
+ when(metadata.topicNames()).thenReturn(Map.of(topicId, "t1"));
+ Set<TopicPartition> partitions = new HashSet<>();
+ partitions.add(new TopicPartition("t1", 0));
+
+ List<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> futures = sendAndVerifyDuplicatedOffsetFetchRequests(
+ commitRequestManager,
+ partitions,
+ 2,
+ Errors.NONE,
+ true,
+ topicId);
+ futures.forEach(f -> {
+ assertTrue(f.isDone());
+ assertFalse(f.isCompletedExceptionally());
+ });
+ // expecting the buffers to be emptied after being completed successfully
+ commitRequestManager.poll(0);
+ assertEmptyPendingRequests(commitRequestManager);
+ }
+
+ @Test
+ public void testOffsetFetchRequestShouldFailedWithTopicIdWhenMetadataUnknownResponseTopicId() {
Review Comment:
typo, shouldFailWith
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -751,6 +751,59 @@ public void testOffsetFetchRequestEnsureDuplicatedRequestSucceed() {
assertEmptyPendingRequests(commitRequestManager);
}
+ @Test
+ public void testOffsetFetchRequestShouldSucceedWithTopicId() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+ Uuid topicId = Uuid.randomUuid();
+ when(metadata.topicIds()).thenReturn(Map.of("t1", topicId));
+ when(metadata.topicNames()).thenReturn(Map.of(topicId, "t1"));
+ Set<TopicPartition> partitions = new HashSet<>();
+ partitions.add(new TopicPartition("t1", 0));
+
+ List<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> futures = sendAndVerifyDuplicatedOffsetFetchRequests(
+ commitRequestManager,
+ partitions,
+ 2,
+ Errors.NONE,
+ true,
+ topicId);
+ futures.forEach(f -> {
+ assertTrue(f.isDone());
+ assertFalse(f.isCompletedExceptionally());
+ });
+ // expecting the buffers to be emptied after being completed successfully
+ commitRequestManager.poll(0);
+ assertEmptyPendingRequests(commitRequestManager);
+ }
+
+ @Test
+ public void testOffsetFetchRequestShouldFailedWithTopicIdWhenMetadataUnknownResponseTopicId() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+ Uuid topicId = Uuid.randomUuid();
+ when(metadata.topicIds()).thenReturn(Map.of("t1", topicId));
+ // Mock the scenario where the topicID from the response is not in the metadata.
+ when(metadata.topicNames()).thenReturn(Map.of());
+ Set<TopicPartition> partitions = new HashSet<>();
+ partitions.add(new TopicPartition("t1", 0));
+
+ List<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> futures = sendAndVerifyDuplicatedOffsetFetchRequests(
+ commitRequestManager,
+ partitions,
+ 2,
+ Errors.NONE,
+ true,
+ topicId);
+ futures.forEach(f -> {
+ assertTrue(f.isDone());
+ assertTrue(f.isCompletedExceptionally());
+ });
Review Comment:
should we extend this to validate it completes with the error we expect? (and all the other actions that we expect on error) we can maybe reuse validation we do for the error cases (that do not include topic ID, so don't cover this case) https://github.com/apache/kafka/blob/169e21199791d02d195a739044734b89847266af/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java#L769
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm merged PR #19885:
URL: https://github.com/apache/kafka/pull/19885
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15283:[2/2] Client support for OffsetFetch with topic ID [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on PR #19885:
URL: https://github.com/apache/kafka/pull/19885#issuecomment-3526079347
@lianetm Sorry for the delayed response. I’m not sure how I missed your review.
Thanks again for your feedback! I’ve updated the PR — please take another look when you have time.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org