Skip to content

KAFKA-6789: Handle COORDINATOR_LOAD_IN_PROGRESS, COORDINATOR_NOT_AVAILABLE retriable errors in AdminClient API #5578

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
hachikuji merged 3 commits into from
May 9, 2019
Merged

KAFKA-6789: Handle COORDINATOR_LOAD_IN_PROGRESS, COORDINATOR_NOT_AVAILABLE retriable errors in AdminClient API #5578

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c58d157
KAFKA-6789: Handle COORDINATOR_LOAD_IN_PROGRESS, COORDINATOR_NOT_AVAI…
omkreddy Mar 8, 2019
75dedba
Address review comments
omkreddy May 9, 2019
1d60ed8
Address review comment
omkreddy May 9, 2019
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
116 changes: 56 additions & 60 deletions clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2551,7 +2551,7 @@ FindCoordinatorRequest.Builder createRequest(int timeoutMs) {
void handleResponse(AbstractResponse abstractResponse) {
final FindCoordinatorResponse fcResponse = (FindCoordinatorResponse) abstractResponse;

if (handleFindCoordinatorError(fcResponse, futures.get(groupId)))
if (handleGroupRequestError(fcResponse.error(), futures.get(groupId)))
return;

final long nowDescribeConsumerGroups = time.milliseconds();
Expand All @@ -2577,38 +2577,37 @@ void handleResponse(AbstractResponse abstractResponse) {
.findFirst().get();

final Errors groupError = Errors.forCode(describedGroup.errorCode());
if (groupError != Errors.NONE) {
// TODO: KAFKA-6789, we can retry based on the error code
future.completeExceptionally(groupError.exception());
} else {
final String protocolType = describedGroup.protocolType();
if (protocolType.equals(ConsumerProtocol.PROTOCOL_TYPE) || protocolType.isEmpty()) {
final List<DescribedGroupMember> members = describedGroup.members();
final List<MemberDescription> memberDescriptions = new ArrayList<>(members.size());
final Set<AclOperation> authorizedOperations = validAclOperations(describedGroup.authorizedOperations());
for (DescribedGroupMember groupMember : members) {
Set<TopicPartition> partitions = Collections.emptySet();
if (groupMember.memberAssignment().length > 0) {
final PartitionAssignor.Assignment assignment = ConsumerProtocol.
deserializeAssignment(ByteBuffer.wrap(groupMember.memberAssignment()));
partitions = new HashSet<>(assignment.partitions());
}
final MemberDescription memberDescription =
new MemberDescription(groupMember.memberId(),
groupMember.clientId(),
groupMember.clientHost(),
new MemberAssignment(partitions));
memberDescriptions.add(memberDescription);

if (handleGroupRequestError(groupError, future))
return;

final String protocolType = describedGroup.protocolType();
if (protocolType.equals(ConsumerProtocol.PROTOCOL_TYPE) || protocolType.isEmpty()) {
final List<DescribedGroupMember> members = describedGroup.members();
final List<MemberDescription> memberDescriptions = new ArrayList<>(members.size());
final Set<AclOperation> authorizedOperations = validAclOperations(describedGroup.authorizedOperations());
for (DescribedGroupMember groupMember : members) {
Set<TopicPartition> partitions = Collections.emptySet();
if (groupMember.memberAssignment().length > 0) {
final PartitionAssignor.Assignment assignment = ConsumerProtocol.
deserializeAssignment(ByteBuffer.wrap(groupMember.memberAssignment()));
partitions = new HashSet<>(assignment.partitions());
}
final ConsumerGroupDescription consumerGroupDescription =
new ConsumerGroupDescription(groupId, protocolType.isEmpty(),
memberDescriptions,
describedGroup.protocolData(),
ConsumerGroupState.parse(describedGroup.groupState()),
fcResponse.node(),
authorizedOperations);
future.complete(consumerGroupDescription);
final MemberDescription memberDescription =
new MemberDescription(groupMember.memberId(),
groupMember.clientId(),
groupMember.clientHost(),
new MemberAssignment(partitions));
memberDescriptions.add(memberDescription);
}
final ConsumerGroupDescription consumerGroupDescription =
new ConsumerGroupDescription(groupId, protocolType.isEmpty(),
memberDescriptions,
describedGroup.protocolData(),
ConsumerGroupState.parse(describedGroup.groupState()),
fcResponse.node(),
authorizedOperations);
future.complete(consumerGroupDescription);
}
}

Expand Down Expand Up @@ -2641,11 +2640,10 @@ private Set<AclOperation> validAclOperations(final int authorizedOperations) {
.collect(Collectors.toSet());
}

private boolean handleFindCoordinatorError(FindCoordinatorResponse response, KafkaFutureImpl<?> future) {
Errors error = response.error();
if (error.exception() instanceof RetriableException) {
private boolean handleGroupRequestError(Errors error, KafkaFutureImpl<?> future) {
if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS || error == Errors.COORDINATOR_NOT_AVAILABLE) {
Comment thread
hachikuji marked this conversation as resolved.
Outdated
throw error.exception();
} else if (response.hasError()) {

@shunge shunge Aug 31, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the same, under the hood:

    public boolean hasError() {
        return this.error != Errors.NONE;
    }

So I think you can keep this?

} else if (error != Errors.NONE) {
future.completeExceptionally(error.exception());
return true;
Comment thread
omkreddy marked this conversation as resolved.
Outdated
}
Expand Down Expand Up @@ -2797,7 +2795,7 @@ FindCoordinatorRequest.Builder createRequest(int timeoutMs) {
void handleResponse(AbstractResponse abstractResponse) {
final FindCoordinatorResponse response = (FindCoordinatorResponse) abstractResponse;

if (handleFindCoordinatorError(response, groupOffsetListingFuture))
if (handleGroupRequestError(response.error(), groupOffsetListingFuture))
return;

final long nowListConsumerGroupOffsets = time.milliseconds();
Expand All @@ -2815,26 +2813,25 @@ void handleResponse(AbstractResponse abstractResponse) {
final OffsetFetchResponse response = (OffsetFetchResponse) abstractResponse;
final Map<TopicPartition, OffsetAndMetadata> groupOffsetsListing = new HashMap<>();

if (response.hasError()) {
groupOffsetListingFuture.completeExceptionally(response.error().exception());
} else {
for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry :
response.responseData().entrySet()) {
final TopicPartition topicPartition = entry.getKey();
OffsetFetchResponse.PartitionData partitionData = entry.getValue();
final Errors error = partitionData.error;

if (error == Errors.NONE) {
final Long offset = partitionData.offset;
final String metadata = partitionData.metadata;
final Optional<Integer> leaderEpoch = partitionData.leaderEpoch;
groupOffsetsListing.put(topicPartition, new OffsetAndMetadata(offset, leaderEpoch, metadata));
} else {
log.warn("Skipping return offset for {} due to error {}.", topicPartition, error);
}
if (handleGroupRequestError(response.error(), groupOffsetListingFuture))
return;

for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry :
response.responseData().entrySet()) {
final TopicPartition topicPartition = entry.getKey();
OffsetFetchResponse.PartitionData partitionData = entry.getValue();
final Errors error = partitionData.error;

if (error == Errors.NONE) {
final Long offset = partitionData.offset;
final String metadata = partitionData.metadata;
final Optional<Integer> leaderEpoch = partitionData.leaderEpoch;
groupOffsetsListing.put(topicPartition, new OffsetAndMetadata(offset, leaderEpoch, metadata));
} else {
log.warn("Skipping return offset for {} due to error {}.", topicPartition, error);
}
groupOffsetListingFuture.complete(groupOffsetsListing);
}
groupOffsetListingFuture.complete(groupOffsetsListing);
}

@Override
Expand Down Expand Up @@ -2891,7 +2888,7 @@ FindCoordinatorRequest.Builder createRequest(int timeoutMs) {
void handleResponse(AbstractResponse abstractResponse) {
final FindCoordinatorResponse response = (FindCoordinatorResponse) abstractResponse;

if (handleFindCoordinatorError(response, futures.get(groupId)))
if (handleGroupRequestError(response.error(), futures.get(groupId)))
return;

final long nowDeleteConsumerGroups = time.milliseconds();
Expand All @@ -2912,11 +2909,10 @@ void handleResponse(AbstractResponse abstractResponse) {
KafkaFutureImpl<Void> future = futures.get(groupId);
final Errors groupError = response.get(groupId);

if (groupError != Errors.NONE) {
future.completeExceptionally(groupError.exception());
} else {
future.complete(null);
}
if (handleGroupRequestError(groupError, future))
return;

future.complete(null);
}

@Override
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/requests/ListGroupsResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ public static Schema[] schemaVersions() {
/**
* Possible error codes:
*
* COORDINATOR_LOADING_IN_PROGRESS (14)
* COORDINATOR_LOAD_IN_PROGRESS (14)
* COORDINATOR_NOT_AVAILABLE (15)
* AUTHORIZATION_FAILED (29)
*/
Expand Down
68 changes: 60 additions & 8 deletions clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -942,7 +942,7 @@ public void testListConsumerGroups() throws Exception {
Collections.emptySet(),
Collections.emptySet(), nodes.get(0));

try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster)) {
try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster, AdminClientConfig.RETRIES_CONFIG, "2")) {
env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());

// Empty metadata response should be retried
Expand Down Expand Up @@ -972,8 +972,8 @@ public void testListConsumerGroups() throws Exception {
// handle retriable errors
env.kafkaClient().prepareResponseFrom(
new ListGroupsResponse(
Errors.COORDINATOR_NOT_AVAILABLE,
Collections.emptyList()
Errors.COORDINATOR_NOT_AVAILABLE,
Collections.emptyList()
),
node1);
env.kafkaClient().prepareResponseFrom(
Expand Down Expand Up @@ -1076,9 +1076,37 @@ public void testDescribeConsumerGroups() throws Exception {
try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster)) {
env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());

env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE, env.cluster().controller()));
//Retriable FindCoordinatorResponse errors should be retried
env.kafkaClient().prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.COORDINATOR_NOT_AVAILABLE, Node.noNode()));
env.kafkaClient().prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.COORDINATOR_LOAD_IN_PROGRESS, Node.noNode()));

env.kafkaClient().prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, env.cluster().controller()));

DescribeGroupsResponseData data = new DescribeGroupsResponseData();

//Retriable errors should be retried

@shunge shunge Aug 31, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

NIT: extra whitespace

data.groups().add(DescribeGroupsResponse.groupMetadata(
"group-0",
Errors.COORDINATOR_LOAD_IN_PROGRESS,
"",
"",
"",
Collections.emptyList(),
Collections.emptySet()));
env.kafkaClient().prepareResponse(new DescribeGroupsResponse(data));

data = new DescribeGroupsResponseData();
data.groups().add(DescribeGroupsResponse.groupMetadata(
"group-0",
Errors.COORDINATOR_NOT_AVAILABLE,
"",
"",
"",
Collections.emptyList(),
Collections.emptySet()));
env.kafkaClient().prepareResponse(new DescribeGroupsResponse(data));

data = new DescribeGroupsResponseData();
TopicPartition myTopicPartition0 = new TopicPartition("my_topic", 0);
TopicPartition myTopicPartition1 = new TopicPartition("my_topic", 1);
TopicPartition myTopicPartition2 = new TopicPartition("my_topic", 2);
Expand Down Expand Up @@ -1143,7 +1171,14 @@ public void testDescribeConsumerGroupOffsets() throws Exception {
try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster)) {
env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());

env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE, env.cluster().controller()));
//Retriable FindCoordinatorResponse errors should be retried
env.kafkaClient().prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.COORDINATOR_NOT_AVAILABLE, Node.noNode()));

env.kafkaClient().prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, env.cluster().controller()));

//Retriable errors should be retried

@shunge shunge Aug 31, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

NIT: same, whitespace

env.kafkaClient().prepareResponse(new OffsetFetchResponse(Errors.COORDINATOR_NOT_AVAILABLE, Collections.emptyMap()));
env.kafkaClient().prepareResponse(new OffsetFetchResponse(Errors.COORDINATOR_LOAD_IN_PROGRESS, Collections.emptyMap()));

TopicPartition myTopicPartition0 = new TopicPartition("my_topic", 0);
TopicPartition myTopicPartition1 = new TopicPartition("my_topic", 1);
Expand Down Expand Up @@ -1192,9 +1227,9 @@ public void testDeleteConsumerGroups() throws Exception {

env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE, env.cluster().controller()));

final Map<String, Errors> response = new HashMap<>();
response.put("group-0", Errors.NONE);
env.kafkaClient().prepareResponse(new DeleteGroupsResponse(response));
final Map<String, Errors> validResponse = new HashMap<>();
validResponse.put("group-0", Errors.NONE);
env.kafkaClient().prepareResponse(new DeleteGroupsResponse(validResponse));

final DeleteConsumerGroupsResult result = env.adminClient().deleteConsumerGroups(groupIds);

Expand All @@ -1207,6 +1242,23 @@ public void testDeleteConsumerGroups() throws Exception {
final DeleteConsumerGroupsResult errorResult = env.adminClient().deleteConsumerGroups(groupIds);
TestUtils.assertFutureError(errorResult.deletedGroups().get("group-0"), GroupAuthorizationException.class);

//Retriable errors should be retried
env.kafkaClient().prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, env.cluster().controller()));

final Map<String, Errors> errorResponse1 = new HashMap<>();
errorResponse1.put("group-0", Errors.COORDINATOR_NOT_AVAILABLE);
env.kafkaClient().prepareResponse(new DeleteGroupsResponse(errorResponse1));

final Map<String, Errors> errorResponse2 = new HashMap<>();
errorResponse2.put("group-0", Errors.COORDINATOR_LOAD_IN_PROGRESS);
env.kafkaClient().prepareResponse(new DeleteGroupsResponse(errorResponse2));

env.kafkaClient().prepareResponse(new DeleteGroupsResponse(validResponse));

final DeleteConsumerGroupsResult errorResult1 = env.adminClient().deleteConsumerGroups(groupIds);

final KafkaFuture<Void> errorResults = errorResult1.deletedGroups().get("group-0");
assertNull(errorResults.get());
}
}

Expand Down