lucasbru commented on code in PR #22636:
URL: https://github.com/apache/kafka/pull/22636#discussion_r3453152126
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java:
##########
@@ -120,6 +131,11 @@ public ApiResult<CoordinatorKey, StreamsGroupDescription>
handleResponse(
final Set<AclOperation> authorizedOperations =
validAclOperations(describedGroup.authorizedOperations());
+ final StreamsGroupTopologyDescriptionStatus
topologyDescriptionStatus =
+
StreamsGroupTopologyDescriptionStatus.forId(describedGroup.topologyDescriptionStatus());
Review Comment:
`forId(byte)` throws `IllegalArgumentException` for any byte not in
{0,1,2,3}, and there is no try/catch around this call inside `handleResponse`.
If a broker sends an unknown status code (e.g. a future version adds status 4),
the exception propagates through `AdminApiDriver.onResponse` ->
`KafkaAdminClient.call.fail`, which since `IllegalArgumentException` is not
`RetriableException` immediately fails every key in the batch. Should be caught
and translated into a per-group failure (put into `failed`), or `forId` should
have a safe fallback for unknown values.
##########
clients/src/testFixtures/java/org/apache/kafka/clients/admin/MockAdminClient.java:
##########
@@ -1470,9 +1472,27 @@ public synchronized DeleteShareGroupsResult
deleteShareGroups(Collection<String>
throw new UnsupportedOperationException("Not implemented yet");
}
+ /**
+ * Registers a {@link StreamsGroupDescription} to be returned by {@link
#describeStreamsGroups}.
+ */
+ public synchronized void
addStreamsGroupDescription(StreamsGroupDescription description) {
+ streamsGroupDescriptions.put(description.groupId(), description);
+ }
+
@Override
public synchronized DescribeStreamsGroupsResult
describeStreamsGroups(Collection<String> groupIds, DescribeStreamsGroupsOptions
options) {
Review Comment:
`options.includeTopologyDescription()` is never checked. A test that
registers a description with `status = AVAILABLE` and calls
`describeStreamsGroups` without `includeTopologyDescription(true)` will get
`AVAILABLE` back, which a real broker would never do (it returns
`NOT_REQUESTED`). Tests using this mock cannot catch callers that forget to set
the option. Should probably return `NOT_REQUESTED` when the option is false.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java:
##########
@@ -189,6 +207,84 @@ private Map<String,
StreamsGroupSubtopologyDescription.TopicInfo> convertTopicIn
));
}
+ private Optional<StreamsGroupTopologyDescription>
convertTopologyDescription(
+ final StreamsGroupTopologyDescriptionStatus status,
+ final StreamsGroupDescribeResponseData.TopologyDescription
topologyDescription) {
+ // The description is present if and only if the status is AVAILABLE.
+ if (status != StreamsGroupTopologyDescriptionStatus.AVAILABLE ||
topologyDescription == null) {
+ return Optional.empty();
+ }
+ final List<StreamsGroupTopologyDescription.Subtopology> subtopologies
= topologyDescription.subtopologies().stream()
+ .map(this::convertTopologySubtopology)
+ .collect(Collectors.toList());
+ final List<StreamsGroupTopologyDescription.GlobalStore> globalStores =
topologyDescription.globalStores().stream()
+ .map(this::convertGlobalStore)
+ .collect(Collectors.toList());
+ return Optional.of(new StreamsGroupTopologyDescription(subtopologies,
globalStores));
+ }
+
+ private StreamsGroupTopologyDescription.Subtopology
convertTopologySubtopology(
+ final
StreamsGroupDescribeResponseData.TopologyDescriptionSubtopology subtopology) {
+ final Map<String, Set<String>> predecessors =
reconstructPredecessors(subtopology.nodes());
+ final List<StreamsGroupTopologyDescription.Node> nodes =
subtopology.nodes().stream()
+ .map(node -> convertTopologyNode(node, predecessors))
+ .collect(Collectors.toList());
+ return new
StreamsGroupTopologyDescription.Subtopology(subtopology.subtopologyId(), nodes);
+ }
+
+ private StreamsGroupTopologyDescription.GlobalStore convertGlobalStore(
+ final
StreamsGroupDescribeResponseData.TopologyDescriptionGlobalStore globalStore) {
+ // The source and processor of a global store form a single unit;
reconstruct predecessors across the pair.
+ final List<StreamsGroupDescribeResponseData.TopologyDescriptionNode>
pair =
+ List.of(globalStore.source(), globalStore.processor());
+ final Map<String, Set<String>> predecessors =
reconstructPredecessors(pair);
+ final StreamsGroupTopologyDescription.Node source =
convertTopologyNode(globalStore.source(), predecessors);
+ final StreamsGroupTopologyDescription.Node processor =
convertTopologyNode(globalStore.processor(), predecessors);
+ if (!(source instanceof StreamsGroupTopologyDescription.Source)
+ || !(processor instanceof
StreamsGroupTopologyDescription.Processor)) {
Review Comment:
`convertGlobalStore` throws `IllegalStateException` when the
source/processor node has an unexpected type, and `convertTopologyNode`'s
default switch arm does the same for unknown `nodeType` values. Both are called
from inside the `handleResponse` group loop with no catch. Same propagation
path as the `forId` issue: the exception escapes `handleResponse` and fails
every key in the batch rather than just the affected group. These should put
the offending group in `failed` and continue.
--
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]
