Copilot commented on code in PR #22639:
URL: https://github.com/apache/kafka/pull/22639#discussion_r3457703093
##########
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();
+ if (description != null) {
+ streamsRebalanceData.setWireTopologyDescription(
+ TopologyDescriptionConverter.toWire(description)
+ );
+ }
+ }
Review Comment:
`InternalTopologyBuilder.describe()` always returns a non-null
`TopologyDescription` (it constructs and returns a new instance), so the `if
(description != null)` branch is dead code and can be removed to simplify the
startup path.
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -902,6 +902,11 @@ public class StreamsConfig extends AbstractConfig {
"Whether to use the configured <code>" +
PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG + "</code> during global store/KTable
processing. " +
"Disabled by default. This config will be removed in Kafka
Streams 5.0, where global exception handling will be enabled by default";
+ public static final String TOPOLOGY_DESCRIPTION_PUSH_ENABLED_CONFIG =
"topology.description.push.enabled";
+ private static final String TOPOLOGY_DESCRIPTION_PUSH_ENABLED_DOC =
"Controls whether the Kafka Streams client sends topology descriptions to the
broker when requested. " +
+ "When set to false, the client will not prepare or push topology
descriptions. " +
+ "Enabled by default.";
Review Comment:
New config constant `TOPOLOGY_DESCRIPTION_PUSH_ENABLED_CONFIG` is missing
the small javadoc marker (`/** {@code ...} */`) that StreamsConfig consistently
uses to label config keys, which makes the config list harder to scan and keep
consistent.
##########
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)) {
Review Comment:
This adds new startup behavior that stashes a wire topology description onto
`StreamsRebalanceData` when the config is enabled, but there is no unit test
exercising the enabled/disabled paths (StreamThread has comprehensive unit
tests already). Adding a focused test in `StreamThreadTest` would help prevent
regressions (eg, accidentally computing/publishing the topology when disabled,
or failing to publish it when enabled).
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1342,7 +1347,13 @@ public class StreamsConfig extends AbstractConfig {
Type.LONG,
null,
Importance.LOW,
- WINDOW_SIZE_MS_DOC);
+ WINDOW_SIZE_MS_DOC)
+ .define(TOPOLOGY_DESCRIPTION_PUSH_ENABLED_CONFIG,
+ Type.BOOLEAN,
+ true,
+ Importance.MEDIUM,
+ TOPOLOGY_DESCRIPTION_PUSH_ENABLED_DOC
+ );
Review Comment:
The new `.define(TOPOLOGY_DESCRIPTION_PUSH_ENABLED_CONFIG, ...)` call ends
with a standalone `);` which is inconsistent with the surrounding `ConfigDef`
chaining style (other entries end with `DOC)` and the final entry ends with
`DOC);`). This makes the config block harder to read and easy to mis-edit.
--
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]
