Kafka streams :如何处理过滤器中的动态条件？

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;

public class FilterExample {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();

        // 从输入主题中读取数据
        KStream<String, String> input = builder.stream("input-topic");

        // 定义过滤器函数
        Predicate<String, String> filterFunction = (key, value) -> {
            // 根据动态条件过滤数据
            // 这里可以根据需要自定义过滤逻辑
            // 返回true表示保留数据，返回false表示过滤数据
            return value.contains("condition");
        };

        // 应用过滤器操作
        KStream<String, String> filteredStream = input.filter(filterFunction);

        // 将过滤后的数据写回到输出主题
        filteredStream.to("output-topic");

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