### Question:
FindDistinctEvents { public Topology buildTopology(Properties envProps, final SpecificAvroSerde<Click> clicksSerde) { final StreamsBuilder builder = new StreamsBuilder(); final String inputTopic = envProps.getProperty("input.topic.name"); final String outputTopic = envProps.getProperty("output.topic.name"); final Duration windowSize = Duration.ofMinutes(2); final Duration retentionPeriod = windowSize; final StoreBuilder<WindowStore<String, Long>> dedupStoreBuilder = Stores.windowStoreBuilder( Stores.persistentWindowStore(storeName, retentionPeriod, windowSize, false ), Serdes.String(), Serdes.Long()); builder.addStateStore(dedupStoreBuilder); builder .stream(inputTopic, Consumed.with(Serdes.String(), clicksSerde)) .transformValues(() -> new DeduplicationTransformer<>(windowSize.toMillis(), (key, value) -> value.getIp()), storeName) .filter((k, v) -> v != null) .to(outputTopic, Produced.with(Serdes.String(), clicksSerde)); return builder.build(); }  Topology buildTopology(Properties envProps,
                                  final SpecificAvroSerde<Click> clicksSerde); void createTopics(Properties envProps); static Properties buildStreamsProperties(Properties envProps); static Properties loadEnvProperties(String fileName); static void main(String[] args);  }### Answer:
@Test public void shouldFilterDistinctEvents() { final FindDistinctEvents distinctifier = new FindDistinctEvents(); String inputTopic = envProps.getProperty("input.topic.name"); String outputTopic = envProps.getProperty("output.topic.name"); final SpecificAvroSerde<Click> clickSerde = makeSerializer(envProps); Topology topology = distinctifier.buildTopology(envProps, clickSerde); final List<Click> expectedOutput; List<Click> actualOutput; try (TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProps)) { Serializer<String> keySerializer = Serdes.String().serializer(); final List<Click> clicks = asList( new Click("10.0.0.1", "https: "2019-09-16T14:53:43+00:00"), new Click("10.0.0.2", "https: "2019-09-16T14:53:43+00:01"), new Click("10.0.0.3", "https: "2019-09-16T14:53:43+00:03"), new Click("10.0.0.1", "https: "2019-09-16T14:53:43+00:00"), new Click("10.0.0.2", "https: "2019-09-16T14:53:43+00:01"), new Click("10.0.0.3", "https: "2019-09-16T14:53:43+00:03")); final TestInputTopic<String, Click> testDriverInputTopic = testDriver.createInputTopic(inputTopic, keySerializer, clickSerde.serializer()); clicks.forEach(clk -> testDriverInputTopic.pipeInput(clk.getIp(), clk)); expectedOutput = asList(clicks.get(0), clicks.get(1), clicks.get(2)); Deserializer<String> keyDeserializer = Serdes.String().deserializer(); actualOutput = testDriver.createOutputTopic(outputTopic, keyDeserializer, clickSerde.deserializer()).readValuesToList() .stream().filter( Objects::nonNull).collect(Collectors.toList()); } Assert.assertEquals(expectedOutput, actualOutput); }