dongnuo123 commented on code in PR #15196:
URL: https://github.com/apache/kafka/pull/15196#discussion_r1453580051
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -524,37 +495,39 @@ public CompletableFuture<ListGroupsResponseData>
listGroups(
);
}
- final CompletableFuture<ListGroupsResponseData> future = new
CompletableFuture<>();
- final List<ListGroupsResponseData.ListedGroup> results = new
ArrayList<>();
final Set<TopicPartition> existingPartitionSet = runtime.partitions();
- final AtomicInteger cnt = new
AtomicInteger(existingPartitionSet.size());
if (existingPartitionSet.isEmpty()) {
return CompletableFuture.completedFuture(new
ListGroupsResponseData());
}
+ final
List<CompletableFuture<List<ListGroupsResponseData.ListedGroup>>> futures =
+ new ArrayList<>();
+
for (TopicPartition tp : existingPartitionSet) {
- runtime.scheduleReadOperation(
+ futures.add(runtime.scheduleReadOperation(
"list-groups",
tp,
(coordinator, lastCommittedOffset) ->
coordinator.listGroups(request.statesFilter(), lastCommittedOffset)
- ).handle((groups, exception) -> {
- if (exception == null) {
- synchronized (results) {
- results.addAll(groups);
- }
+ ).exceptionally(exception -> {
+ exception = Errors.maybeUnwrapException(exception);
+ if (exception instanceof NotCoordinatorException) {
+ return Collections.emptyList();
} else {
- if (!(exception instanceof NotCoordinatorException)) {
- future.complete(new
ListGroupsResponseData().setErrorCode(Errors.forException(exception).code()));
- }
- }
- if (cnt.decrementAndGet() == 0) {
- future.complete(new
ListGroupsResponseData().setGroups(results));
+ throw new CompletionException(exception);
}
- return null;
- });
+ }));
}
- return future;
+
+ return FutureUtils
+ .combineFutures(futures, ArrayList::new, List::addAll)
+ .thenApply(groups -> new
ListGroupsResponseData().setGroups(groups))
+ .exceptionally(exception -> handleOperationException(
+ "ListGroups",
+ request,
+ exception,
+ (error, __) -> new
ListGroupsResponseData().setErrorCode(error.code())
+ ));
Review Comment:
The existing implementation completes the future immediately if an error is
not `NotCoordinatorException`, and the new implementation adds an empty list
and continue listing groups in the following topic partitions. Is this our
desired behavior?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -1098,30 +1054,36 @@ private static boolean isGroupIdNotEmpty(String
groupId) {
return groupId != null && !groupId.isEmpty();
}
- /**
- * Handles the exception in the scheduleWriteOperation.
- * @return The Errors instance associated with the given exception.
- */
- private static Errors normalizeException(Throwable exception) {
- exception = Errors.maybeUnwrapException(exception);
-
- if (exception instanceof UnknownTopicOrPartitionException ||
- exception instanceof NotEnoughReplicasException ||
- exception instanceof TimeoutException) {
- return Errors.COORDINATOR_NOT_AVAILABLE;
- }
-
- if (exception instanceof NotLeaderOrFollowerException ||
- exception instanceof KafkaStorageException) {
- return Errors.NOT_COORDINATOR;
- }
-
- if (exception instanceof RecordTooLargeException ||
- exception instanceof RecordBatchTooLargeException ||
- exception instanceof InvalidFetchSizeException) {
- return Errors.UNKNOWN_SERVER_ERROR;
+ private <REQ, RSP> RSP handleOperationException(
Review Comment:
Do we need a javadoc for it?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
