cadonna commented on code in PR #13846:
URL: https://github.com/apache/kafka/pull/13846#discussion_r1238377947
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -755,6 +755,18 @@ public class StreamsConfig extends AbstractConfig {
public static final String DEFAULT_CLIENT_SUPPLIER_CONFIG =
"default.client.supplier";
public static final String DEFAULT_CLIENT_SUPPLIER_DOC = "Client supplier
class that implements the
<code>org.apache.kafka.streams.KafkaClientSupplier</code> interface.";
+ public static final String RACK_AWARE_ASSSIGNMENT_STRATEGY_NONE = "NONE";
+ public static final String RACK_AWARE_ASSSIGNMENT_STRATEGY_MIN_TRAFFIC =
"MIN_TRAFFIC";
+ public static final String
RACK_AWARE_ASSSIGNMENT_STRATEGY_MIN_TRAFFIC_BALANCE_SUBTOPOLOGY =
"MIN_TRAFFIC_BALANCE_SUBTOPOLOGY";
+
+ /** {@code } rack.aware.assignment.strategy */
+ @SuppressWarnings("WeakerAccess")
+ public static final String RACK_AWARE_ASSIGNMENT_STRATEGY =
"rack.aware.assignment.strategy";
+ public static final String RACK_AWARE_ASSIGNMENT_STRATEGY_DOC = "The
strategy we use for rack aware assignment. Rack aware assignment will take
client.rack and racks of TopicPartition into account when assigning"
Review Comment:
```suggestion
public static final String RACK_AWARE_ASSIGNMENT_STRATEGY_DOC = "The
strategy we use for rack aware assignment. Rack aware assignment will take
client.rack and racks of topic partitions into account when assigning"
```
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -1375,6 +1375,18 @@ public void shouldReturnDefaultClientSupplier() {
assertTrue(supplier instanceof DefaultKafkaClientSupplier);
}
+ @Test
+ public void shouldReturnDefaultRackAwareAssignmentConfig() {
+ final String strategy =
streamsConfig.getString(StreamsConfig.RACK_AWARE_ASSIGNMENT_STRATEGY);
+ assertEquals(StreamsConfig.RACK_AWARE_ASSSIGNMENT_STRATEGY_NONE,
strategy);
+ }
+
Review Comment:
I think you also need to test setting the three valid values. In the end
they are part of the public API, and should be a test that verifies that they
are accepted.
The test should be like this:
https://github.com/apache/kafka/blob/474053d2973b8790e50ccfe1bb0699694b0de1c7/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java#L615
Do not use constants `RACK_AWARE_ASSSIGNMENT_STRATEGY_NONE,
RACK_AWARE_ASSSIGNMENT_STRATEGY_MIN_TRAFFIC, or
RACK_AWARE_ASSSIGNMENT_STRATEGY_MIN_TRAFFIC_BALANCE_SUBTOPOLOGY` because you
want to ensure that the test fails if the content of those constants changes.
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -890,6 +902,12 @@ public class StreamsConfig extends AbstractConfig {
in(AT_LEAST_ONCE, EXACTLY_ONCE, EXACTLY_ONCE_BETA,
EXACTLY_ONCE_V2),
Importance.MEDIUM,
PROCESSING_GUARANTEE_DOC)
+ .define(RACK_AWARE_ASSIGNMENT_STRATEGY,
+ Type.STRING,
+ RACK_AWARE_ASSSIGNMENT_STRATEGY_NONE,
+ in(RACK_AWARE_ASSSIGNMENT_STRATEGY_NONE,
RACK_AWARE_ASSSIGNMENT_STRATEGY_MIN_TRAFFIC,
RACK_AWARE_ASSSIGNMENT_STRATEGY_MIN_TRAFFIC_BALANCE_SUBTOPOLOGY),
Review Comment:
nit: I think you could also use an enum like here:
https://github.com/apache/kafka/blob/c958d8719dc2588bd27958b54a65dea514808796/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java#L907
But that is not needed, I just wanted to give you an alternative.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
