lucasbru commented on code in PR #22627:
URL: https://github.com/apache/kafka/pull/22627#discussion_r3444255798
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java:
##########
@@ -299,6 +304,135 @@ private static Map.Entry<String, ApiError>
toFailureEntry(String groupId, Throwa
"Topology description plugin failed to delete the topology."));
}
+ /**
+ * Populate {@code TopologyDescription} and {@code
TopologyDescriptionStatus} on each
+ * {@code DescribedGroup} carried by {@code result}, calling {@code
plugin.getTopology} only
+ * for groups whose persisted {@code StoredDescriptionTopologyEpoch}
matches the group's
+ * current topology epoch. Returns a future that completes when every
per-group plugin call
+ * has settled; the future never completes exceptionally — per-group
errors fold into
+ * {@code TOPOLOGY_DESCRIPTION_STATUS_ERROR} on the corresponding
describedGroup.
+ *
+ * <p>Per-group status decisions:
+ * <ul>
+ * <li>{@code errorCode != NONE} — the group could not be resolved;
leave the status
+ * field at its {@code NOT_REQUESTED} default since the client
should consult the
+ * group's error code first.</li>
+ * <li>No plugin configured on this broker — every successful group
becomes
+ * {@code NOT_STORED}: from the client's perspective the broker
simply has no
+ * description to serve.</li>
+ * <li>{@code Topology} field is null on the response (group has not yet
declared a
+ * topology), {@code storedEpoch} is missing or {@code -1}, or
{@code storedEpoch}
+ * does not match {@code topology().epoch()} — {@code
NOT_STORED}.</li>
+ * <li>Plugin call returns null (the plugin no longer has the data, e.g.
backend wipe)
+ * — {@code NOT_STORED}.</li>
+ * <li>Plugin call completes exceptionally, throws synchronously, or
returns a null
+ * future (SPI contract violation) — {@code ERROR}. Conversion
failures from the
+ * returned POJO to the wire schema also fold into {@code ERROR} so
a single
+ * malformed plugin response cannot poison the rest of the
batch.</li>
+ * <li>Plugin call returns a non-null description — {@code AVAILABLE},
with the
+ * converted topology attached.</li>
+ * </ul>
+ */
+ public
CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>>
attachTopologyDescriptions(
+ StreamsGroupDescribeResult result
+ ) {
+ if (plugin.isEmpty()) {
+ for (StreamsGroupDescribeResponseData.DescribedGroup
describedGroup : result.describedGroups()) {
+ if (describedGroup.errorCode() == Errors.NONE.code()) {
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED);
+ }
+ }
+ return CompletableFuture.completedFuture(result.describedGroups());
+ }
+ final StreamsGroupTopologyDescriptionPlugin topologyDescriptionPlugin
= plugin.get();
+ List<CompletableFuture<Void>> pluginFutures = new ArrayList<>();
+ for (StreamsGroupDescribeResponseData.DescribedGroup describedGroup :
result.describedGroups()) {
+ CompletableFuture<Void> outcome =
maybeAttachOne(topologyDescriptionPlugin, describedGroup, result);
+ if (outcome != null) pluginFutures.add(outcome);
+ }
+ if (pluginFutures.isEmpty()) {
+ return CompletableFuture.completedFuture(result.describedGroups());
+ }
+ return CompletableFuture.allOf(pluginFutures.toArray(new
CompletableFuture<?>[0]))
+ .thenApply(unused -> result.describedGroups());
+ }
+
+ /**
+ * Inspect one describedGroup and, if eligible, fire {@code
plugin.getTopology}. Returns
+ * null when no plugin call is needed (status has already been decided
synchronously from
+ * the response shape); otherwise returns a future that completes once the
plugin call has
+ * been folded into the describedGroup's status / topology fields.
+ */
+ private CompletableFuture<Void> maybeAttachOne(
+ StreamsGroupTopologyDescriptionPlugin p,
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup,
+ StreamsGroupDescribeResult result
+ ) {
+ if (describedGroup.errorCode() != Errors.NONE.code()) {
+ return null;
+ }
+ if (describedGroup.topology() == null) {
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED);
+ return null;
+ }
+ Integer storedEpoch =
result.storedDescriptionTopologyEpochs().get(describedGroup.groupId());
+ if (storedEpoch == null || storedEpoch == -1 || storedEpoch !=
describedGroup.topology().epoch()) {
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED);
+ return null;
+ }
+ int topologyEpoch = describedGroup.topology().epoch();
+ CompletableFuture<StreamsGroupTopologyDescription> pluginFuture;
+ try {
+ pluginFuture = p.getTopology(describedGroup.groupId(),
topologyEpoch);
+ } catch (Exception e) {
+ // SPI contract violation: synchronous throw treated as ERROR.
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_ERROR);
+ return CompletableFuture.completedFuture(null);
+ }
+ if (pluginFuture == null) {
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_ERROR);
+ return CompletableFuture.completedFuture(null);
+ }
+ return pluginFuture.handle((topology, throwable) -> {
+ applyGetTopologyOutcome(describedGroup, topology, throwable);
+ return null;
+ });
+ }
+
+ private static void applyGetTopologyOutcome(
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup,
+ StreamsGroupTopologyDescription topology,
+ Throwable throwable
+ ) {
+ if (throwable != null) {
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_ERROR);
+ return;
+ }
+ if (topology == null) {
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED);
+ return;
+ }
+ try {
+ describedGroup.setTopologyDescription(
+
StreamsGroupTopologyDescriptionConverter.toDescribeResponse(topology));
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_AVAILABLE);
+ } catch (Exception conversionError) {
+ // Defense in depth. With the current contract this catch is
unreachable: the
+ // StreamsGroupTopologyDescription record and its nested
Subtopology / Source /
Review Comment:
The comment is a bit long. And I wouldn't mention why something is not
tested in production code.
How about
```
// Defensive catch, should be unreachable in practice
```
--
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]
