MINOR: add test for repartition/source-topic/changelog optimization
If topology optimization is enabled, KafkaStreams does not create store changelog topics but re-uses source input topics if possible. However, this optimization should not be applied to internal repartition topics, because those are actively purged.
Reviewers: A. Sophie Blee-Goldman
*/
package org.apache.kafka.streams;
-import java.util.regex.Pattern;
import org.apache.kafka.common.serialization.LongSerializer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringSerializer;
@@ -62,6 +61,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
+import java.util.regex.Pattern;
import static java.util.Arrays.asList;
import static org.hamcrest.CoreMatchers.equalTo;
@@ -462,6 +462,35 @@ public class StreamsBuilderTest {
equalTo(true));
}
+ @Test
+ public void shouldNotReuseRepartitionTopicAsChangelogs() {
+ final String topic = "topic";
+ builder.<Long, String>stream(topic).repartition().toTable(Materialized.as("store"));
+ final Properties props = StreamsTestUtils.getStreamsConfig("appId");
+ props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
+ final Topology topology = builder.build(props);
+
+ final InternalTopologyBuilder internalTopologyBuilder = TopologyWrapper.getInternalTopologyBuilder(topology);
+ internalTopologyBuilder.rewriteTopology(new StreamsConfig(props));
+
+ assertThat(
+ internalTopologyBuilder.buildTopology().storeToChangelogTopic(),
+ equalTo(Collections.singletonMap("store", "appId-store-changelog"))
+ );
+ assertThat(
+ internalTopologyBuilder.stateStores().keySet(),
+ equalTo(Collections.singleton("store"))
+ );
+ assertThat(
+ internalTopologyBuilder.stateStores().get("store").loggingEnabled(),
+ equalTo(true)
+ );
+ assertThat(
+ internalTopologyBuilder.topicGroups().get(1).stateChangelogTopics.keySet(),
+ equalTo(Collections.singleton("appId-store-changelog"))
+ );
+ }
+
@Test
public void shouldNotReuseSourceTopicAsChangelogsByDefault() {
final String topic = "topic";
