Kafka Streams Materialized View with Kotlin

Question

To create a kafka streams state store in Java I can do this:

final KGroupedStream<String, String> wordCounts = textLines
            .flatMapValues(value -> Arrays.asList(pattern.split(value.toLowerCase())))
            .groupBy((key, word) -> word);

wordCounts.count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as(WORD_COUNT_STORE));

I am attempting to convert this to Kotlin, like this:

val wordCounts: KGroupedStream<String, String> = textLines
        .flatMapValues({value -> value.split("\\W+") })
        .groupBy({ _, word -> word})

wordCounts.count(Materialized.<String, Long, KeyValueStore<Bytes, Array<Byte>>>as(WORD_COUNT_STORE))

However, I get the following compiler error:

Interface KeyValueStore does not have constructors

What do I need to do?

Possible duplicate of [Why does this Kotlin method have enclosing backticks?](https://stackoverflow.com/questions/44149474/why-does-this-kotlin-method-have-enclosing-backticks) — LF00, Nov 23 '19 at 08:57

score 4 · Answer 1 · answered Sep 14 '18 at 12:30

In case it's of use to anyone else, as well as the backticks suggested by Raman, I had to make couple of other changes:

First, the generic types need to be specified after the as method, not straight after the Materialized class.
Secondly, rather than using Array<Byte> I had to use ByteArray.

So the full line of code that worked for me was:

wordCounts.count(Materialized.`as`<String, Long, KeyValueStore<Bytes, ByteArray>>(WORD_COUNT_STORE))

score 2 · Accepted Answer · answered May 24 '18 at 22:04

2

Since as is a reserved word in Kotlin, try surrounding the as with backticks i.e.

`as`

answered May 24 '18 at 22:04

Raman

17,606
5
95
112

Kafka Streams Materialized View with Kotlin

2 Answers2