mjsax commented on a change in pull request #11447:
URL: https://github.com/apache/kafka/pull/11447#discussion_r741604615
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/GroupedTableOperationRepartitionNode.java
##########
@@ -119,5 +119,10 @@ public void writeToTopology(final InternalTopologyBuilder
topologyBuilder) {
processorParameters
);
}
+
+ @Override
+ public boolean isOptimizable() {
+ return false;
Review comment:
Why is a table repartitioning not optimizable?
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/KStreamRepartitionIntegrationTest.java
##########
@@ -672,6 +672,42 @@ public void shouldGoThroughRebalancingCorrectly() throws
Exception {
assertEquals(2, getNumberOfPartitionsForTopic(repartitionTopicName));
}
+ @Test
+ public void shouldNotFilterOutNullKeysOnRepartition() throws Exception {
+ final String repartitionName = "repartition-test";
+ final long timestamp = System.currentTimeMillis();
+ sendEvents(
+ timestamp,
+ Arrays.asList(
+ new KeyValue<>(1, "A"),
+ new KeyValue<>(2, "B"),
+ new KeyValue<>(3, null)
+ )
+ );
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ final Repartitioned<String, String> repartitioned =
Repartitioned.<String, String>as(repartitionName)
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.String());
+
+ builder.stream(inputTopic, Consumed.with(Serdes.Integer(),
Serdes.String()))
+ .selectKey((key, value) -> value == null ? null : key.toString())
+ .repartition(repartitioned)
+ .mapValues(value -> value != null ? "mapped-" + value :
"default-value")
Review comment:
Why do we need the `mapValues()` step in this test?
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/UnoptimizableRepartitionNode.java
##########
@@ -53,19 +53,13 @@ private UnoptimizableRepartitionNode(final String nodeName,
public void writeToTopology(final InternalTopologyBuilder topologyBuilder)
{
topologyBuilder.addInternalTopic(repartitionTopic,
internalTopicProperties);
- topologyBuilder.addProcessor(
Review comment:
Not sure if I understand this change? Can you elaborate?
--
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]
