App.java

        StreamsBuilder builder = new StreamsBuilder();

        builder
          .stream("inJlinkTopic", Consumed.with(Serdes.String(), Serdes.String()))
          .peek((key, value) -> log.info("Received message: {}", value))
          .filter((key, value) -> "PASS".equals(value))
          .to("outJlinkTopic", Produced.with(Serdes.String(), Serdes.String()));

        KafkaStreams streams = new KafkaStreams(builder.build(), config);
        streams.start();