kafka流中的处理器节点

我正在努力 processor node 在卡夫卡溪流中。对于一个简单的代码,我编写如下代码只是为了过滤 UserID ,这是正确的做法吗 处理器节点 在卡夫卡河?

但是,以下代码未编译,引发错误: The method filter(Predicate<? super Object,? super Object>) in the type KStream<Object,Object> is not applicable for the arguments (new Predicate<String,String>(){})

KStreamBuilder builder = new KStreamBuilder();

builder.stream(topic)
    .filter(new Predicate <String, String>() {
        //@Override
        public boolean test(String key, String value) {
            Hashtable<Object, Object> message;
            // put you processor logic here
            return message.get("UserID").equals("1");
        }
    })
    .to(streamouttopic);

    final KafkaStreams streams = new KafkaStreams(builder, props);
    final CountDownLatch latch = new CountDownLatch(1);

    // attach shutdown handler to catch control-c
    Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
    @Override
    public void run() {
       streams.close();
       latch.countDown();
    }
});

try {
    streams.start();
    latch.await();
} catch (Throwable e) {
    System.exit(1);
}
System.exit(0);

有人能指引我吗?