我有一个Kafka Streams处理器，它使用三个主题并尝试合并（连接操作）它们的键。连接成功后，它会进行一些聚合，然后将结果推送到目标主题。应用程序第一次运行后，它会尝试使用这些主题中的所有数据。其中两个主题使用类似查找表的数据，这意味着我需要从一开始就使用所有的数据。但是其中一个主题是我的主主题。所以我需要从最新的主题开始使用。但是我的应用程序从一开始就使用所有Kafka主题。所以我想从一开始使用两个主题，从最新的主题使用一个主题。我用的是Spring的云流，Kafka的流绑定器。这里是我的配置和一些代码片段;
应用程序名称：

spring.cloud.stream.function.definition: processName;
spring.cloud.stream.kafka.streams.binder.functions.processName.applicationId: myappId
spring.cloud.stream.bindings.processName-in-0.destination: mainTopic
spring.cloud.stream.bindings.processName-in-0.consumer.group: mainTopic-cg
spring.cloud.stream.bindings.processName-in-0.consumer.startOffset: latest
spring.cloud.stream.bindings.processName-in-1.destination: secondTopic
spring.cloud.stream.bindings.processName-in-1.consumer.group: secondTopic-cg
spring.cloud.stream.bindings.processName-in-1.consumer.startOffset: earliest
spring.cloud.stream.bindings.processName-in-2.destination: thirdTopic
spring.cloud.stream.bindings.processName-in-2.consumer.group: thirdTopic-cg
spring.cloud.stream.bindings.processName-in-2.consumer.startOffset: earliest
spring.cloud.stream.bindings.processName-out-0.destination: otputTopics

spring.cloud.stream.kafka.streams.binder.replication-factor: 1
spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms: 10000
spring.cloud.stream.kafka.streams.binder.configuration.state.dir: state-store

流处理器：

public Function<KStream<String, MainTopic>,
        Function<KTable<String, SecondTopic>,
                Function<KTable<String, ThirdTopic>,
                        KStream<String, OutputTopic>>>> processName(){
    return mainTopicKStream -> (
            secondTopicTable -> (
                    thirdTopicKTable -> (
                            aggregateOperations.AggregateByAmount(
                                    joinOperations.JoinSecondThirdTopic(mainTopicKStream ,secondTopicTable ,thirdTopicKTable )
                                            .filter((k,v) -> v.IsOk() != 4) 
                                            .groupByKey(Grouped.with(AppSerdes.String(), AppSerdes.OutputTopic()))
                                    , TimeWindows.of(Duration.ofMinutes(1)).advanceBy(Duration.ofMinutes(1))
                            ).toStream()
                    )
            ));
}