哪个版本的spring cloud stream binder kstream与Kafka 1.0.0兼容

Question

当尝试运行稍微修改过的word count示例版本时，我遇到了一个错误，即“没有符合条件的'org.apache.kafka.streams.kstream.KStreamBuilder'类型的bean”。在我的POM中，我使用了Spring-Cloud-Stream依赖项：Elmhurst。M3导入依赖项，其中导入了spring cloud stream绑定器kstream:2.0.0。立方米。

然而，我不认为我做了什么特别的事情：

@SpringBootApplication
@EnableBinding(KStreamProcessor.class)
public class KafkaExampleSpringcloud1Application {

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

    }

    @Autowired
    private TimeWindows timeWindows;

    @StreamListener("input")
    @SendTo("output")
    public KStream<?, WordCount> process(KStream<Object, String> input) {

        return input
                .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
                .map((key, value) -> new KeyValue<>(value, value))
                .groupByKey(Serdes.String(), Serdes.String())
                .count(timeWindows, "WordCounts")
                .toStream()
                .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end()))));
    }



    static class WordCount {

        private String word;

        private long count;

        private Date start;

        private Date end;

        WordCount(String word, long count, Date start, Date end) {
            this.word = word;
            this.count = count;
            this.start = start;
            this.end = end;
        }

        public String getWord() {
            return word;
        }

        public void setWord(String word) {
            this.word = word;
        }

        public long getCount() {
            return count;
        }

        public void setCount(long count) {
            this.count = count;
        }

        public Date getStart() {
            return start;
        }

        public void setStart(Date start) {
            this.start = start;
        }

        public Date getEnd() {
            return end;
        }

        public void setEnd(Date end) {
            this.end = end;
        }
    }

这是我的申请表。yml：

application.name: kafka-example-01
spring.cloud.stream:
  kstream:
    configuration:
      commit.interval.ms: 1000
      key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
      value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
    timeWindow.length: 5000
    binder:
      brokers: localhost
      zkNodes: localhost
  kafka:
    binder:
      autoCreateTopics: false
  bindings:
    output.destination: word-count-output
    input.destnation : word-count-input

我尝试使用旧版本，但我一直有ClassNotFoundException错误（例如StreamsBuilder类）

Answer 1

让它与这些一起工作：

<dependencies>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream-binder-kstream</artifactId>
        <version>2.0.0.BUILD-SNAPSHOT</version>
    </dependency>

    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream</artifactId>
        <version>2.0.0.BUILD-SNAPSHOT</version>
    </dependency>

    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
        <version>2.0.0.BUILD-SNAPSHOT</version>
    </dependency>

    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-lang3</artifactId>
    </dependency>

    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <version>1.0.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>1.0.0</version>
    </dependency>

    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
        <version>2.1.1.RELEASE</version>
    </dependency>

    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.11</artifactId>
        <version>1.0.0</version>
    </dependency>
</dependencies>

Answer 2

它目前并不是一个里程碑，只有2.0.0。构建快照。它将在M4中。

在这里promise。