Skip to content

KAFKA-14500; [5/N] Implement JoinGroup protocol in new GroupCoordinator #13870

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dajac merged 14 commits into from
Jul 19, 2023
Merged

KAFKA-14500; [5/N] Implement JoinGroup protocol in new GroupCoordinator #13870

Changes from 1 commit
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
242d6a5
Implement JoinGroup in new Coordinator
jeffkbkim Jun 16, 2023
9dbb770
unify MockCoordinatorTimer
jeffkbkim Jul 12, 2023
2f55300
tidy
jeffkbkim Jul 12, 2023
dc6979e
checkstyle
jeffkbkim Jul 13, 2023
23209ce
address comments
jeffkbkim Jul 14, 2023
c4c74cb
address comments
jeffkbkim Jul 17, 2023
bf2ea19
checkstyle
jeffkbkim Jul 17, 2023
2797b9b
log only on incomplete response futures
jeffkbkim Jul 17, 2023
a9bb3c3
build fix
jeffkbkim Jul 17, 2023
4de7890
only log when not kafka exception
jeffkbkim Jul 17, 2023
17459f4
address comments
jeffkbkim Jul 18, 2023
45c99f6
fix test
jeffkbkim Jul 18, 2023
7b82894
Trigger build
jeffkbkim Jul 18, 2023
1d21547
fix test
jeffkbkim Jul 18, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions ...coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
*/
package org.apache.kafka.coordinator.group;

import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.errors.InvalidFetchSizeException;
Expand Down Expand Up @@ -301,10 +302,12 @@ public CompletableFuture<JoinGroupResponseData> joinGroup(
topicPartitionFor(request.groupId()),
coordinator -> coordinator.genericGroupJoin(context, request, responseFuture)
).exceptionally(exception -> {
if (!responseFuture.isDone()) {
if (!(exception instanceof KafkaException)) {
log.error("Request {} hit an unexpected exception: {}",
Comment thread
dajac marked this conversation as resolved.
Outdated
Comment thread
jeffkbkim marked this conversation as resolved.
Outdated
request, exception.getMessage());

}

if (!responseFuture.isDone()) {
responseFuture.complete(new JoinGroupResponseData()
.setErrorCode(Errors.forException(exception).code()));
}
Expand Down