@Test
    public void shouldCompactAndDeleteTopicsForWindowStoreChangelogs() throws Exception {
        final String appID = APP_ID + "-compact-delete";
        streamsProp.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);

        //
        // Step 1: Configure and start a simple word count topology
        //
        final StreamsBuilder builder = new StreamsBuilder();
        final KStream<String, String> textLines = builder.stream(DEFAULT_INPUT_TOPIC);

        final int durationMs = 2000;

        textLines.flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.getDefault()).split("\\W+")))
            .groupBy(MockMapper.selectValueMapper())
            .windowedBy(TimeWindows.of(ofSeconds(1L)).grace(ofMillis(0L)))
            .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("CountWindows").withRetention(ofSeconds(2L)));

        try (final KafkaStreams streams = new KafkaStreams(builder.build(), streamsProp)) {
            startApplicationAndWaitUntilRunning(streams);

            //
            // Step 2: Produce some input data to the input topic.
            //
            produceData(Arrays.asList("hello", "world", "world", "hello world"));

            //
            // Step 3: Verify the state changelog topics are compact
            //
            waitForCompletion(streams, 2, 30000L);
        }

        final Properties properties = getTopicProperties(ProcessorStateManager.storeChangelogTopic(appID, "CountWindows", null));
        final List<String> policies = Arrays.asList(properties.getProperty(TopicConfig.CLEANUP_POLICY_CONFIG).split(","));
        assertEquals(2, policies.size());
        assertTrue(policies.contains(TopicConfig.CLEANUP_POLICY_COMPACT));
        assertTrue(policies.contains(TopicConfig.CLEANUP_POLICY_DELETE));
        // retention should be 1 day + the window duration
        final long retention = TimeUnit.MILLISECONDS.convert(1, TimeUnit.DAYS) + durationMs;
        assertEquals(retention, Long.parseLong(properties.getProperty(TopicConfig.RETENTION_MS_CONFIG)));

        final Properties repartitionProps = getTopicProperties(appID + "-CountWindows-repartition");
        assertEquals(TopicConfig.CLEANUP_POLICY_DELETE, repartitionProps.getProperty(TopicConfig.CLEANUP_POLICY_CONFIG));
        assertEquals(4, repartitionProps.size());
    }
Tests related to internal topics in streams
streams/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.java