Zystkmt

I want to use the spring cloud stream api to aggreate events from a topic.
Therefore i use as input a KStream.

KStream<Object, LoggerCreatedMessage>

Now i want to use an aggregator to store my new Object in a KeyValue Store, so i use following code:

input

  .map((key, value) -> {

    return new KeyValue<>(value.logger_id,value);

  })

  /*.groupBy(

    (s, loggerEvent) -> loggerEvent.logger_id,

    Serialized.with(null, loggerEventSerde))*/

  .groupByKey()

  .aggregate(

    String::new,

    (s, loggerEvent, vr) -> {

      return vr;

    },

    Materialized.<String, String, KeyValueStore<Bytes, byte>>as(STORE_NAME).withKeySerde(Serdes.String()).

      withValueSerde(Serdes.String())

  );

Why can i only use a String as an Initializer is it not possible to use any Object?

Instead of String::new i wanted to use LoggerDomain::new, but i only get this error message:

Bad return type in method reference: cannot convert LoggerDomain to VR

Do i miss something?

You define <key,value> as <String, String> via Materialized.<String, String, KeyValueStore<Bytes, byte>> -- if you value type should be LoggerDomain, it should be Materialized.<KeyType, LoggerDomain, KeyValueStore<Bytes, byte>>().

Note that you need to provide a custom Serde for LoggerDomain for this case to Materialized, too.