如何使用KStreams -kafka和kafka-stream在spring Bean中记录偏移量

@Configuration
public class KafkaStreamsConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Bean
    public StreamsBuilderFactoryBean streamsBuilder() {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-streams-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        StreamsBuilderFactoryBean streamsBuilder = new StreamsBuilderFactoryBean();
        streamsBuilder.setStreamsConfiguration(props);

        return streamsBuilder;
    }
}

@Component
public class MyKafkaStreamsProcessor {

    @Autowired
    private StreamsBuilder streamsBuilder;

    @Bean
    public KStream<String, String> process() {
        KStream<String, String> input = streamsBuilder.stream("my-input-topic");

        // 在这里进行数据处理和转换

        input.to("my-output-topic");

        return input;
    }
}

@SpringBootApplication
@EnableKafkaStreams
public class MyApplication {

    public static void main(String[] args) {
        SpringApplication.run(MyApplication.class, args);
    }

    @Bean
    public MyKafkaStreamsProcessor myKafkaStreamsProcessor() {
        return new MyKafkaStreamsProcessor();
    }
}