如何在Spring云kafka活页夹中将偏移量重置为开始

Question

我正在使用spring-cloud-kafka绑定器将数据读取到KStream。在阅读其中一个主题的数据时，我需要从头开始阅读。

我尝试设置kafka偏移重置和启动偏移属性。但是，找不到任何参考。

你能帮我提供任何示例application.yaml来重置偏移量吗？这样我就可以从一开始就使用主题中的消息。

添加我使用过的应用程序：

spring.cloud.stream.bindings.input:
  destination: input-topic1
  consumer:
    useNativeDecoding: true
    headerMode: raw
spring.cloud.stream.bindings.output:
  destination: output-topic
  producer:
    useNativeDecoding: true
    headerMode: raw
spring.cloud.stream.bindings.beginningInput:
  destination: beginning-topic
  consumer:
    useNativeDecoding: true
    headerMode: raw
spring.cloud.stream.kafka.streams.bindings.input:
  consumer:
    keySerde: org.apache.kafka.common.serialization.Serdes$StringSerde
    valueSerde: org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.bindings.output:
  producer:
    keySerde: org.apache.kafka.common.serialization.Serdes$StringSerde
    valueSerde: org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.bindings.beginningInput:
  consumer:
    keySerde: org.apache.kafka.common.serialization.Serdes$StringSerde
    valueSerde: org.apache.kafka.common.serialization.Serdes$StringSerde
    resetOffsets: true
    startOffset: earliest
spring.cloud.stream.kafka.streams.binder:
  brokers: 127.0.0.1
  zkNodes: 127.0.0.1
  configuration:
    default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
    default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
    commit.interval.ms: 1000

Answer 1

resetOffset已损坏。它在2.0.0.RELEASE.恢复。

PR在这里。