lucasbru commented on code in PR #22627:
URL: https://github.com/apache/kafka/pull/22627#discussion_r3444255792
##########
core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala:
##########
@@ -303,6 +305,9 @@ class AuthorizerIntegrationTest extends
AbstractAuthorizerIntegrationTest {
ApiKeys.ALTER_SHARE_GROUP_OFFSETS -> (shareGroupReadAcl ++ topicReadAcl),
ApiKeys.STREAMS_GROUP_HEARTBEAT -> (streamsGroupReadAcl ++
topicDescribeAcl),
ApiKeys.STREAMS_GROUP_DESCRIBE -> (streamsGroupDescribeAcl ++
topicDescribeAcl),
+ // KIP-1331: like offset commit, topology push is treated as READ on the
GROUP (not a
+ // modification), so apps deployed with READ-only group ACLs can still
push.
+ ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE -> streamsGroupReadAcl,
Review Comment:
Like for other API keys, we shouldn't put inline comments here that don't
add much to what the code already says.
##########
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
Review Comment:
Nit: duplicates the fan-out scaffold from
`invokeDeleteTopologies`/`invokeSetTopology`, but without
`Errors.maybeUnwrapException` — so a plugin returning
`CompletionException(null)` is handled differently here than in the set/delete
paths.
##########
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 /
+ // Processor / Sink records enforce non-null collections through
Objects.requireNonNull
+ // + List.copyOf in their canonical constructors, and the Node
sealed interface
+ // (Source | Processor | Sink) makes the converter's "unknown node
type" branch
+ // unreachable from any plugin response built via legal API. The
catch survives so
+ // that a future relaxation of the sealed permits (or a
contract-bypass via bytecode
+ // manipulation) folds into a per-group ERROR rather than
poisoning the rest of the
+ // batch. No direct test: Mockito explicitly refuses to mock
sealed interfaces, so
+ // the only way to drive this branch would be bytecode-level
forgery we are
+ // unwilling to ship in tests.
+ describedGroup.setTopologyDescription(null);
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_ERROR);
Review Comment:
`setTopologyDescription` is never called when `toDescribeResponse` throws —
Java evaluates the argument before invoking the method — so
`topologyDescription` is still `null` here. The explicit reset is dead code.
##########
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:
This comment is a bit long — something like "Defense in depth; unreachable
with the current sealed Node types. Not unit-tested: Mockito cannot mock sealed
interfaces." covers the same ground.
--
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]
