lianetm commented on code in PR #19642:
URL: https://github.com/apache/kafka/pull/19642#discussion_r2076053266
##########
clients/src/test/java/org/apache/kafka/clients/admin/internals/ListConsumerGroupOffsetsHandlerTest.java:
##########
@@ -233,117 +279,226 @@ public void testFailedHandleResponse() {
@Test
public void testFailedHandleResponseWithMultipleGroups() {
- Map<String, Errors> errorMap = new HashMap<>();
- errorMap.put(groupZero, Errors.GROUP_AUTHORIZATION_FAILED);
- errorMap.put(groupOne, Errors.GROUP_ID_NOT_FOUND);
- errorMap.put(groupTwo, Errors.INVALID_GROUP_ID);
- Map<String, Class<? extends Throwable>> groupToExceptionMap = new
HashMap<>();
- groupToExceptionMap.put(groupZero, GroupAuthorizationException.class);
- groupToExceptionMap.put(groupOne, GroupIdNotFoundException.class);
- groupToExceptionMap.put(groupTwo, InvalidGroupIdException.class);
- assertFailedForMultipleGroups(groupToExceptionMap,
- handleWithErrorWithMultipleGroups(errorMap, batchedRequestMap));
+ var errorMap = Map.of(
+ group0, Errors.GROUP_AUTHORIZATION_FAILED,
+ group1, Errors.GROUP_ID_NOT_FOUND,
+ group2, Errors.INVALID_GROUP_ID
+ );
+ var groupToExceptionMap = Map.of(
+ group0, (Class<? extends Throwable>)
GroupAuthorizationException.class,
+ group1, (Class<? extends Throwable>)
GroupIdNotFoundException.class,
+ group2, (Class<? extends Throwable>) InvalidGroupIdException.class
+ );
+ assertFailedForMultipleGroups(
+ groupToExceptionMap,
+ handleWithErrorWithMultipleGroups(errorMap, multiGroupSpecs)
+ );
}
private OffsetFetchResponse buildResponse(Errors error) {
return new OffsetFetchResponse(
- throttleMs,
- Collections.singletonMap(groupZero, error),
- Collections.singletonMap(groupZero, new HashMap<>()));
- }
-
- private OffsetFetchResponse buildResponseWithMultipleGroups(
- Map<String, Errors> errorMap,
- Map<String, Map<TopicPartition, PartitionData>> responseData
- ) {
- return new OffsetFetchResponse(throttleMs, errorMap, responseData);
Review Comment:
`throttleMs` doesn't seem needed after this (on ln 152)
##########
core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala:
##########
@@ -1609,15 +1608,19 @@ class AuthorizerIntegrationTest extends
AbstractAuthorizerIntegrationTest {
// without describe permission on the topic, we shouldn't be able to fetch
offsets
val offsetFetchRequest = createOffsetFetchRequestAllPartitions
var offsetFetchResponse =
connectAndReceive[OffsetFetchResponse](offsetFetchRequest)
- assertEquals(Errors.NONE, offsetFetchResponse.groupLevelError(group))
- assertTrue(offsetFetchResponse.partitionDataMap(group).isEmpty)
Review Comment:
why removing this? this check is what ensures that we're not returning the
offsets if no auth right?
##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java:
##########
@@ -60,221 +55,126 @@
public class OffsetFetchResponse extends AbstractResponse {
public static final long INVALID_OFFSET = -1L;
public static final String NO_METADATA = "";
- public static final PartitionData UNKNOWN_PARTITION = new
PartitionData(INVALID_OFFSET,
-
Optional.empty(),
-
NO_METADATA,
-
Errors.UNKNOWN_TOPIC_OR_PARTITION);
- public static final PartitionData UNAUTHORIZED_PARTITION = new
PartitionData(INVALID_OFFSET,
-
Optional.empty(),
-
NO_METADATA,
-
Errors.TOPIC_AUTHORIZATION_FAILED);
+
private static final List<Errors> PARTITION_ERRORS = Arrays.asList(
- Errors.UNKNOWN_TOPIC_OR_PARTITION, Errors.TOPIC_AUTHORIZATION_FAILED);
+ Errors.UNKNOWN_TOPIC_OR_PARTITION,
+ Errors.TOPIC_AUTHORIZATION_FAILED,
+ Errors.UNSTABLE_OFFSET_COMMIT
Review Comment:
Is there a reason why we need this error here? this one is expected only if
v>7 (requires stable), but `PARTITION_ERRORS` is used only if v<2 (when we want
to bubble up errors)
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
