lucasbru commented on code in PR #22639:
URL: https://github.com/apache/kafka/pull/22639#discussion_r3457795494
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java:
##########
@@ -351,6 +352,12 @@ public String toString() {
private final AtomicLong acceptableRecoveryLag = new AtomicLong(-1);
+ private final
AtomicReference<StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription>
wireTopologyDescription = new AtomicReference<>(null);
+
+ private final AtomicReference<String> memberId = new AtomicReference<>();
Review Comment:
`StreamsGroupHeartbeatRequestManager` reads `membershipManager.memberId()`
directly instead of using such a nullable copy. Would it make sense to follow
the existing pattern?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -692,13 +693,24 @@ private static StreamsRebalanceData
initStreamsRebalanceData(final UUID processI
final Map<String, StreamsRebalanceData.Subtopology> subtopologies =
initBrokerTopology(config, internalTopologyBuilder);
- return new StreamsRebalanceData(
+ final StreamsRebalanceData streamsRebalanceData = new
StreamsRebalanceData(
processId,
endpoint,
rackId,
subtopologies,
config.getClientTags()
);
+
+ if
(config.getBoolean(StreamsConfig.TOPOLOGY_DESCRIPTION_PUSH_ENABLED_CONFIG)) {
+ final TopologyDescription description =
internalTopologyBuilder.describe();
Review Comment:
`internalTopologyBuilder.describe()` returns source topics from
`SourceNodeFactory.topics` directly — the raw un-decorated names (e.g.
`"foo-repartition"`). `initBrokerTopology` (called right after) goes through
`subtopologyToTopicsInfo()`, which calls `decorateTopic()` for topics in
`internalTopicNamesWithProperties`, yielding `"my-app-foo-repartition"`. So the
wire topology description and the heartbeat subtopologies disagree on internal
topic names for any application with repartition or other internal topics.
--
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]
