lucliu1108 commented on code in PR #23566:
URL: https://github.com/apache/kafka/pull/23566#discussion_r4100068251
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java:
##########
@@ -275,6 +458,33 @@ public StreamsGroupHeartbeatResult
maybeSetTopologyDescriptionRequired(
return result;
}
+ /**
+ * Decorates a streams group heartbeat's result with {@link
#maybeSetTopologyDescriptionRequired},
+ * once the underlying write completes. No-ops when no plugin is
configured, without
+ * scheduling the extra stage. If decoration itself fails (e.g. because of
an unexpected
+ * response shape), the heartbeat has already committed durably, so the
failure is logged and
+ * the committed result is returned unmodified rather than translated into
an error; that
+ * would mask a successful broker-side state change behind a
client-visible failure.
+ */
+ public CompletableFuture<StreamsGroupHeartbeatResult>
decorateHeartbeatResult(
+ CompletableFuture<StreamsGroupHeartbeatResult> heartbeat,
+ String groupId,
+ int apiVersion,
+ int memberEpoch
+ ) {
+ if (!isPluginConfigured()) return heartbeat;
+ return heartbeat.thenApply(result -> {
+ try {
+ return maybeSetTopologyDescriptionRequired(result, groupId,
apiVersion, memberEpoch);
Review Comment:
Fix: change all 3 methods to private.
--
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]
