dajac commented on a change in pull request #11019:
URL: https://github.com/apache/kafka/pull/11019#discussion_r669398821
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandler.java
##########
@@ -97,54 +108,79 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetDeleteResponse response = (OffsetDeleteResponse)
abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new
HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new
HashMap<>();
+ final Map<CoordinatorKey, Throwable> failed = new HashMap<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
final Errors error = Errors.forCode(response.data().errorCode());
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleGroupError(groupId, error, failed, groupsToUnmap,
groupsToRetry);
} else {
- final Map<TopicPartition, Errors> partitions = new HashMap<>();
- response.data().topics().forEach(topic ->
+ final Map<TopicPartition, Errors> partitionResults = new
HashMap<>();
+ response.data().topics().forEach(topic ->
topic.partitions().forEach(partition -> {
Errors partitionError =
Errors.forCode(partition.errorCode());
- if (!handleError(groupId, partitionError, failed,
unmapped)) {
- partitions.put(new TopicPartition(topic.name(),
partition.partitionIndex()), partitionError);
- }
+
+ partitionResults.put(new TopicPartition(topic.name(),
partition.partitionIndex()), partitionError);
})
);
- if (!partitions.isEmpty())
- completed.put(groupId, partitions);
+
+ completed.put(groupId, partitionResults);
Review comment:
Could we directly return here as well?
```
return new ApiResult<>(Collections.singletonMap(groupId, partitionResults),
Collections.emptyList(), Collections.emptyList()) ;
```
I think that it will make the error handling a bit more explicit.
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandler.java
##########
@@ -97,54 +108,79 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetDeleteResponse response = (OffsetDeleteResponse)
abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new
HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new
HashMap<>();
+ final Map<CoordinatorKey, Throwable> failed = new HashMap<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
final Errors error = Errors.forCode(response.data().errorCode());
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleGroupError(groupId, error, failed, groupsToUnmap,
groupsToRetry);
Review comment:
It seems that `groupsToRetry` is not really necessary in this case.
Moreover, we could directly return in the branch as we don't expect errors in
the partitions.
```
if (error != Errors.NONE) {
final Map<CoordinatorKey, Throwable> failed = new HashMap<>();
final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
handleGroupError(groupId, error, failed, groupsToUnmap);
return new ApiResult<>(Collections.emptyMap(), failed, new
ArrayList<>(groupsToUnmap);
}
```
`groupId` will be either in `failed` or in `groupsToUnmap` after the call to
`handleGroupError`.
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandler.java
##########
@@ -97,54 +108,79 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetDeleteResponse response = (OffsetDeleteResponse)
abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new
HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new
HashMap<>();
+ final Map<CoordinatorKey, Throwable> failed = new HashMap<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
final Errors error = Errors.forCode(response.data().errorCode());
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleGroupError(groupId, error, failed, groupsToUnmap,
groupsToRetry);
} else {
- final Map<TopicPartition, Errors> partitions = new HashMap<>();
- response.data().topics().forEach(topic ->
+ final Map<TopicPartition, Errors> partitionResults = new
HashMap<>();
+ response.data().topics().forEach(topic ->
topic.partitions().forEach(partition -> {
Errors partitionError =
Errors.forCode(partition.errorCode());
- if (!handleError(groupId, partitionError, failed,
unmapped)) {
- partitions.put(new TopicPartition(topic.name(),
partition.partitionIndex()), partitionError);
- }
+
+ partitionResults.put(new TopicPartition(topic.name(),
partition.partitionIndex()), partitionError);
})
);
- if (!partitions.isEmpty())
- completed.put(groupId, partitions);
+
+ completed.put(groupId, partitionResults);
+ }
+
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ completed,
+ failed,
+ Collections.emptyList()
+ );
+ } else {
+ // retry the request, so don't send completed/failed results back
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
}
- return new ApiResult<>(completed, failed, unmapped);
}
- private boolean handleError(
+ private void handleGroupError(
CoordinatorKey groupId,
Errors error,
Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
case GROUP_AUTHORIZATION_FAILED:
case GROUP_ID_NOT_FOUND:
case INVALID_GROUP_ID:
- log.error("Received non retriable error for group {} in
`DeleteConsumerGroupOffsets` response", groupId,
- error.exception());
+ case NON_EMPTY_GROUP:
+ log.debug("`OffsetDelete` request for group id {} failed due
to error {}.", groupId, error);
Review comment:
nit: `groupId` -> `groupId.idValue`. There are few other cases.
##########
File path:
clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandlerTest.java
##########
@@ -67,48 +69,88 @@ public void testBuildRequest() {
@Test
public void testSuccessfulHandleResponse() {
Map<TopicPartition, Errors> responseData =
Collections.singletonMap(t0p0, Errors.NONE);
- assertCompleted(handleWithError(Errors.NONE), responseData);
+ assertCompleted(handleWithGroupError(Errors.NONE), responseData);
}
@Test
public void testUnmappedHandleResponse() {
- assertUnmapped(handleWithError(Errors.NOT_COORDINATOR));
+ assertUnmapped(handleWithGroupError(Errors.NOT_COORDINATOR));
+ assertUnmapped(handleWithGroupError(Errors.COORDINATOR_NOT_AVAILABLE));
}
@Test
public void testRetriableHandleResponse() {
- assertRetriable(handleWithError(Errors.COORDINATOR_LOAD_IN_PROGRESS));
- assertRetriable(handleWithError(Errors.COORDINATOR_NOT_AVAILABLE));
+
assertRetriable(handleWithGroupError(Errors.COORDINATOR_LOAD_IN_PROGRESS));
}
@Test
- public void testFailedHandleResponse() {
- assertFailed(GroupAuthorizationException.class,
handleWithError(Errors.GROUP_AUTHORIZATION_FAILED));
- assertFailed(GroupIdNotFoundException.class,
handleWithError(Errors.GROUP_ID_NOT_FOUND));
- assertFailed(InvalidGroupIdException.class,
handleWithError(Errors.INVALID_GROUP_ID));
+ public void testFailedHandleResponseWithGroupError() {
+ assertGroupFailed(GroupAuthorizationException.class,
handleWithGroupError(Errors.GROUP_AUTHORIZATION_FAILED));
+ assertGroupFailed(GroupIdNotFoundException.class,
handleWithGroupError(Errors.GROUP_ID_NOT_FOUND));
+ assertGroupFailed(InvalidGroupIdException.class,
handleWithGroupError(Errors.INVALID_GROUP_ID));
+ assertGroupFailed(GroupNotEmptyException.class,
handleWithGroupError(Errors.NON_EMPTY_GROUP));
}
- private OffsetDeleteResponse buildResponse(Errors error) {
+ @Test
+ public void testFailedHandleResponseWithPartitionError() {
+ assertPartitionFailed(Collections.singletonMap(t0p0,
Errors.GROUP_SUBSCRIBED_TO_TOPIC),
+ handleWithPartitionError(Errors.GROUP_SUBSCRIBED_TO_TOPIC));
+ assertPartitionFailed(Collections.singletonMap(t0p0,
Errors.TOPIC_AUTHORIZATION_FAILED),
+ handleWithPartitionError(Errors.TOPIC_AUTHORIZATION_FAILED));
+ assertPartitionFailed(Collections.singletonMap(t0p0,
Errors.UNKNOWN_TOPIC_OR_PARTITION),
+ handleWithPartitionError(Errors.UNKNOWN_TOPIC_OR_PARTITION));
+ }
+
+ private OffsetDeleteResponse buildGroupErrorResponse(Errors error) {
+ OffsetDeleteResponse response = new OffsetDeleteResponse(
+ new OffsetDeleteResponseData()
+ .setErrorCode(error.code()));
+ if (error == Errors.NONE) {
+ response.data()
+ .setThrottleTimeMs(0)
+ .setTopics(new
OffsetDeleteResponseTopicCollection(singletonList(
+ new OffsetDeleteResponseTopic()
+ .setName("t0")
+ .setPartitions(new
OffsetDeleteResponsePartitionCollection(singletonList(
+ new OffsetDeleteResponsePartition()
+ .setPartitionIndex(0)
+ .setErrorCode(error.code())
+ ).iterator()))
+ ).iterator()));
+ }
+ return response;
+ }
+
+ private OffsetDeleteResponse buildPartitionErrorResponse(Errors error) {
OffsetDeleteResponse response = new OffsetDeleteResponse(
- new OffsetDeleteResponseData()
- .setThrottleTimeMs(0)
- .setTopics(new
OffsetDeleteResponseTopicCollection(singletonList(
- new OffsetDeleteResponseTopic()
- .setName("t0")
- .setPartitions(new
OffsetDeleteResponsePartitionCollection(singletonList(
- new OffsetDeleteResponsePartition()
- .setPartitionIndex(0)
- .setErrorCode(error.code())
- ).iterator()))
- ).iterator())));
+ new OffsetDeleteResponseData()
+ .setThrottleTimeMs(0)
+ .setTopics(new
OffsetDeleteResponseTopicCollection(singletonList(
+ new OffsetDeleteResponseTopic()
+ .setName("t0")
Review comment:
ditto.
##########
File path:
clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandlerTest.java
##########
@@ -149,4 +191,21 @@ private void assertFailed(
assertEquals(singleton(key), result.failedKeys.keySet());
assertTrue(expectedExceptionType.isInstance(result.failedKeys.get(key)));
}
-}
+
+ private void assertPartitionFailed(
+ Map<TopicPartition, Errors> expectedResult,
+ AdminApiHandler.ApiResult<CoordinatorKey, Map<TopicPartition, Errors>>
result
+ ) {
+ CoordinatorKey key = CoordinatorKey.byGroupId(groupId);
+ assertEquals(singleton(key), result.completedKeys.keySet());
+
+ // verify the completed value is expected result
+ Collection<Map<TopicPartition, Errors>> completeCollection =
result.completedKeys.values();
+ assertEquals(1, completeCollection.size());
+ Map<TopicPartition, Errors> completeMap =
completeCollection.iterator().next();
+ assertEquals(expectedResult, completeMap);
+
+ assertEquals(emptyList(), result.unmappedKeys);
+ assertEquals(emptySet(), result.failedKeys.keySet());
+ }
+}
Review comment:
nit: Could we add the empty line back?
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandler.java
##########
@@ -97,54 +108,79 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetDeleteResponse response = (OffsetDeleteResponse)
abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new
HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new
HashMap<>();
+ final Map<CoordinatorKey, Throwable> failed = new HashMap<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
final Errors error = Errors.forCode(response.data().errorCode());
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleGroupError(groupId, error, failed, groupsToUnmap,
groupsToRetry);
} else {
- final Map<TopicPartition, Errors> partitions = new HashMap<>();
- response.data().topics().forEach(topic ->
+ final Map<TopicPartition, Errors> partitionResults = new
HashMap<>();
+ response.data().topics().forEach(topic ->
topic.partitions().forEach(partition -> {
Errors partitionError =
Errors.forCode(partition.errorCode());
- if (!handleError(groupId, partitionError, failed,
unmapped)) {
- partitions.put(new TopicPartition(topic.name(),
partition.partitionIndex()), partitionError);
- }
+
+ partitionResults.put(new TopicPartition(topic.name(),
partition.partitionIndex()), partitionError);
})
);
- if (!partitions.isEmpty())
- completed.put(groupId, partitions);
+
+ completed.put(groupId, partitionResults);
+ }
+
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ completed,
+ failed,
+ Collections.emptyList()
+ );
+ } else {
+ // retry the request, so don't send completed/failed results back
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
}
- return new ApiResult<>(completed, failed, unmapped);
}
- private boolean handleError(
+ private void handleGroupError(
CoordinatorKey groupId,
Errors error,
Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
case GROUP_AUTHORIZATION_FAILED:
case GROUP_ID_NOT_FOUND:
case INVALID_GROUP_ID:
- log.error("Received non retriable error for group {} in
`DeleteConsumerGroupOffsets` response", groupId,
- error.exception());
+ case NON_EMPTY_GROUP:
+ log.debug("`OffsetDelete` request for group id {} failed due
to error {}.", groupId, error);
failed.put(groupId, error.exception());
- return true;
+ break;
case COORDINATOR_LOAD_IN_PROGRESS:
+ // If the coordinator is in the middle of loading, then we
just need to retry
+ log.debug("`OffsetDelete` request for group {} failed because
the coordinator" +
+ " is still in the process of loading state. Will retry.",
groupId);
+ groupsToRetry.add(groupId);
+ break;
case COORDINATOR_NOT_AVAILABLE:
- return true;
case NOT_COORDINATOR:
- log.debug("DeleteConsumerGroupOffsets request for group {}
returned error {}. Will retry",
- groupId, error);
- unmapped.add(groupId);
- return true;
+ // If the coordinator is unavailable or there was a
coordinator change, then we unmap
+ // the key so that we retry the `FindCoordinator` request
+ log.debug("`OffsetDelete` request for group {} returned error
{}. " +
Review comment:
nit: `group` -> `group id`?
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandler.java
##########
@@ -97,54 +108,79 @@ public String apiName() {
Set<CoordinatorKey> groupIds,
AbstractResponse abstractResponse
) {
+ validateKeys(groupIds);
+
final OffsetDeleteResponse response = (OffsetDeleteResponse)
abstractResponse;
- Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new
HashMap<>();
- Map<CoordinatorKey, Throwable> failed = new HashMap<>();
- List<CoordinatorKey> unmapped = new ArrayList<>();
+ final Map<CoordinatorKey, Map<TopicPartition, Errors>> completed = new
HashMap<>();
+ final Map<CoordinatorKey, Throwable> failed = new HashMap<>();
+ final Set<CoordinatorKey> groupsToUnmap = new HashSet<>();
+ final Set<CoordinatorKey> groupsToRetry = new HashSet<>();
final Errors error = Errors.forCode(response.data().errorCode());
if (error != Errors.NONE) {
- handleError(groupId, error, failed, unmapped);
+ handleGroupError(groupId, error, failed, groupsToUnmap,
groupsToRetry);
} else {
- final Map<TopicPartition, Errors> partitions = new HashMap<>();
- response.data().topics().forEach(topic ->
+ final Map<TopicPartition, Errors> partitionResults = new
HashMap<>();
+ response.data().topics().forEach(topic ->
topic.partitions().forEach(partition -> {
Errors partitionError =
Errors.forCode(partition.errorCode());
- if (!handleError(groupId, partitionError, failed,
unmapped)) {
- partitions.put(new TopicPartition(topic.name(),
partition.partitionIndex()), partitionError);
- }
+
+ partitionResults.put(new TopicPartition(topic.name(),
partition.partitionIndex()), partitionError);
})
);
- if (!partitions.isEmpty())
- completed.put(groupId, partitions);
+
+ completed.put(groupId, partitionResults);
+ }
+
+ if (groupsToUnmap.isEmpty() && groupsToRetry.isEmpty()) {
+ return new ApiResult<>(
+ completed,
+ failed,
+ Collections.emptyList()
+ );
+ } else {
+ // retry the request, so don't send completed/failed results back
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ new ArrayList<>(groupsToUnmap)
+ );
}
- return new ApiResult<>(completed, failed, unmapped);
}
- private boolean handleError(
+ private void handleGroupError(
CoordinatorKey groupId,
Errors error,
Map<CoordinatorKey, Throwable> failed,
- List<CoordinatorKey> unmapped
+ Set<CoordinatorKey> groupsToUnmap,
+ Set<CoordinatorKey> groupsToRetry
) {
switch (error) {
case GROUP_AUTHORIZATION_FAILED:
case GROUP_ID_NOT_FOUND:
case INVALID_GROUP_ID:
- log.error("Received non retriable error for group {} in
`DeleteConsumerGroupOffsets` response", groupId,
- error.exception());
+ case NON_EMPTY_GROUP:
+ log.debug("`OffsetDelete` request for group id {} failed due
to error {}.", groupId, error);
failed.put(groupId, error.exception());
- return true;
+ break;
case COORDINATOR_LOAD_IN_PROGRESS:
+ // If the coordinator is in the middle of loading, then we
just need to retry
+ log.debug("`OffsetDelete` request for group {} failed because
the coordinator" +
Review comment:
nit: `group` -> `group id`?
##########
File path:
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
##########
@@ -3359,23 +3360,23 @@ public void testDeleteConsumerGroupOffsets() throws
Exception {
prepareFindCoordinatorResponse(Errors.NONE,
env.cluster().controller()));
env.kafkaClient().prepareResponse(new OffsetDeleteResponse(
- new OffsetDeleteResponseData()
- .setTopics(new
OffsetDeleteResponseTopicCollection(Stream.of(
- new OffsetDeleteResponseTopic()
- .setName("foo")
- .setPartitions(new
OffsetDeleteResponsePartitionCollection(Collections.singletonList(
- new OffsetDeleteResponsePartition()
- .setPartitionIndex(0)
- .setErrorCode(Errors.NONE.code())
- ).iterator())),
- new OffsetDeleteResponseTopic()
- .setName("bar")
- .setPartitions(new
OffsetDeleteResponsePartitionCollection(Collections.singletonList(
- new OffsetDeleteResponsePartition()
- .setPartitionIndex(0)
-
.setErrorCode(Errors.GROUP_SUBSCRIBED_TO_TOPIC.code())
- ).iterator()))
- ).collect(Collectors.toList()).iterator()))
+ new OffsetDeleteResponseData()
+ .setTopics(new
OffsetDeleteResponseTopicCollection(Stream.of(
+ new OffsetDeleteResponseTopic()
+ .setName("foo")
+ .setPartitions(new
OffsetDeleteResponsePartitionCollection(Collections.singletonList(
+ new OffsetDeleteResponsePartition()
+ .setPartitionIndex(0)
+ .setErrorCode(Errors.NONE.code())
+ ).iterator())),
+ new OffsetDeleteResponseTopic()
+ .setName("bar")
+ .setPartitions(new
OffsetDeleteResponsePartitionCollection(Collections.singletonList(
+ new OffsetDeleteResponsePartition()
+ .setPartitionIndex(0)
+
.setErrorCode(Errors.GROUP_SUBSCRIBED_TO_TOPIC.code())
+ ).iterator()))
+ ).collect(Collectors.toList()).iterator()))
Review comment:
nit: Is it really better like this? Personally, I prefer the previous
indentation.
##########
File path:
clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandlerTest.java
##########
@@ -67,48 +69,88 @@ public void testBuildRequest() {
@Test
public void testSuccessfulHandleResponse() {
Map<TopicPartition, Errors> responseData =
Collections.singletonMap(t0p0, Errors.NONE);
- assertCompleted(handleWithError(Errors.NONE), responseData);
+ assertCompleted(handleWithGroupError(Errors.NONE), responseData);
}
@Test
public void testUnmappedHandleResponse() {
- assertUnmapped(handleWithError(Errors.NOT_COORDINATOR));
+ assertUnmapped(handleWithGroupError(Errors.NOT_COORDINATOR));
+ assertUnmapped(handleWithGroupError(Errors.COORDINATOR_NOT_AVAILABLE));
}
@Test
public void testRetriableHandleResponse() {
- assertRetriable(handleWithError(Errors.COORDINATOR_LOAD_IN_PROGRESS));
- assertRetriable(handleWithError(Errors.COORDINATOR_NOT_AVAILABLE));
+
assertRetriable(handleWithGroupError(Errors.COORDINATOR_LOAD_IN_PROGRESS));
}
@Test
- public void testFailedHandleResponse() {
- assertFailed(GroupAuthorizationException.class,
handleWithError(Errors.GROUP_AUTHORIZATION_FAILED));
- assertFailed(GroupIdNotFoundException.class,
handleWithError(Errors.GROUP_ID_NOT_FOUND));
- assertFailed(InvalidGroupIdException.class,
handleWithError(Errors.INVALID_GROUP_ID));
+ public void testFailedHandleResponseWithGroupError() {
+ assertGroupFailed(GroupAuthorizationException.class,
handleWithGroupError(Errors.GROUP_AUTHORIZATION_FAILED));
+ assertGroupFailed(GroupIdNotFoundException.class,
handleWithGroupError(Errors.GROUP_ID_NOT_FOUND));
+ assertGroupFailed(InvalidGroupIdException.class,
handleWithGroupError(Errors.INVALID_GROUP_ID));
+ assertGroupFailed(GroupNotEmptyException.class,
handleWithGroupError(Errors.NON_EMPTY_GROUP));
}
- private OffsetDeleteResponse buildResponse(Errors error) {
+ @Test
+ public void testFailedHandleResponseWithPartitionError() {
+ assertPartitionFailed(Collections.singletonMap(t0p0,
Errors.GROUP_SUBSCRIBED_TO_TOPIC),
+ handleWithPartitionError(Errors.GROUP_SUBSCRIBED_TO_TOPIC));
+ assertPartitionFailed(Collections.singletonMap(t0p0,
Errors.TOPIC_AUTHORIZATION_FAILED),
+ handleWithPartitionError(Errors.TOPIC_AUTHORIZATION_FAILED));
+ assertPartitionFailed(Collections.singletonMap(t0p0,
Errors.UNKNOWN_TOPIC_OR_PARTITION),
+ handleWithPartitionError(Errors.UNKNOWN_TOPIC_OR_PARTITION));
+ }
+
+ private OffsetDeleteResponse buildGroupErrorResponse(Errors error) {
+ OffsetDeleteResponse response = new OffsetDeleteResponse(
+ new OffsetDeleteResponseData()
+ .setErrorCode(error.code()));
+ if (error == Errors.NONE) {
+ response.data()
+ .setThrottleTimeMs(0)
+ .setTopics(new
OffsetDeleteResponseTopicCollection(singletonList(
+ new OffsetDeleteResponseTopic()
+ .setName("t0")
Review comment:
nit: Could we rely on `t0p0` here for the name and the partition?
##########
File path:
clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandlerTest.java
##########
@@ -149,4 +191,21 @@ private void assertFailed(
assertEquals(singleton(key), result.failedKeys.keySet());
assertTrue(expectedExceptionType.isInstance(result.failedKeys.get(key)));
}
-}
+
+ private void assertPartitionFailed(
+ Map<TopicPartition, Errors> expectedResult,
+ AdminApiHandler.ApiResult<CoordinatorKey, Map<TopicPartition, Errors>>
result
+ ) {
+ CoordinatorKey key = CoordinatorKey.byGroupId(groupId);
+ assertEquals(singleton(key), result.completedKeys.keySet());
+
+ // verify the completed value is expected result
+ Collection<Map<TopicPartition, Errors>> completeCollection =
result.completedKeys.values();
+ assertEquals(1, completeCollection.size());
+ Map<TopicPartition, Errors> completeMap =
completeCollection.iterator().next();
+ assertEquals(expectedResult, completeMap);
Review comment:
You already assert that `completedKeys` only contains `key` so it seems
that we could just verify that `result.completedKeys.get(key)` is equal to
`expectedResult`, no?
--
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]
