Kafka Streams to topic

Question

I need to calculate an average with a kafka streams. The producer produce with Avro and so i need to deserialize with it and i receive a GenericRecord with a json String that i have to elaborate.

I use a user defined type as support.

 private class Tuple {

    public int occ;
    public int sum;


    public Tuple (int occ, int sum) {
        this.occ = occ;
        this.sum = sum;
    }

    public void sum (int toAdd) {
        this.sum += toAdd;
        this.occ ++;
    }

    public Double getAverage () {
        return new Double (this.sum / this.occ);
    }

    public String toString() {
        return "occorrenze: " + this.occ + ", somma: " + sum + ", media -> " + getAverage();
    }

}

Now the elaboration:

 StreamsBuilder builder = new StreamsBuilder();
    KStream<GenericRecord, GenericRecord> source =
          builder.stream(topic);

    KStream<GenericRecord, GenericRecord>[] branches = source.branch(
            (key,value) -> partition(value.toString()),
            (key, value) -> true
    );

    KGroupedStream <String, String> groupedStream = branches[0]
            .mapValues(value -> createJson(value.toString()))
            .map((key, value) -> KeyValue.pair(new String("T_DUR_CICLO"), value.getNumberValue("payload", "T_DUR_CICLO")))
            .groupByKey( Serialized.with(
                    Serdes.String(), /* key (note: type was modified) */
                    Serdes.String()));  /* value */

    branches[0].foreach((key, value) -> System.out.println(key + " " + value));

    KTable<String, Tuple> aggregatedStream = groupedStream.aggregate(
            () -> new Tuple(0, 0), // initializer 
            (aggKey, newValue, aggValue) ->  new Tuple (aggValue.occ + 1, aggValue.sum + Integer.parseInt(newValue)),
            Materialized.as("aggregate-state-store").with(Serdes.String(), new MySerde()));

    aggregatedStream
            .toStream()
            .foreach((key, value) -> System.out.println(key + ": " + value));

    KStream<String, Double> average = aggregatedStream
            .mapValues(v -> v.getAverage())
            .toStream();

The problem is when i go to store the stream in a topic with:

 average.to("average");

Here the exception:

  Exception in thread "streamtest-6d743b83-ce22-435e-aee5-76a745ce3571-StreamThread-1" org.apache.kafka.streams.errors.ProcessorStateException: task [1_0] Failed to flush state store KSTREAM-AGGREGATE-STATE-STORE-0000000007
    at org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:242)
    at org.apache.kafka.streams.processor.internals.AbstractTask.flushState(AbstractTask.java:202)
    at org.apache.kafka.streams.processor.internals.StreamTask.flushState(StreamTask.java:420)
    at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:394)
    at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:382)
    at org.apache.kafka.streams.processor.internals.AssignedTasks$1.apply(AssignedTasks.java:67)
    at org.apache.kafka.streams.processor.internals.AssignedTasks.applyToRunningTasks(AssignedTasks.java:362)
    at org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:352)
    at org.apache.kafka.streams.processor.internals.TaskManager.commitAll(TaskManager.java:401)
    at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1042)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:845)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736)
Caused by: org.apache.kafka.streams.errors.StreamsException: A serializer (key: io.confluent.kafka.streams.serdes.avro.GenericAvroSerializer / value: io.confluent.kafka.streams.serdes.avro.GenericAvroSerializer) is not compatible to the actual key or value type (key type: java.lang.String / value type: java.lang.Double). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters.
    at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:94)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:143)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:126)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:90)
    at org.apache.kafka.streams.kstream.internals.KStreamMapValues$KStreamMapProcessor.process(KStreamMapValues.java:41)
    at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:50)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.runAndMeasureLatency(ProcessorNode.java:244)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:133)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:143)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:126)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:90)
    at org.apache.kafka.streams.kstream.internals.KTableMapValues$KTableMapValuesProcessor.process(KTableMapValues.java:106)
    at org.apache.kafka.streams.kstream.internals.KTableMapValues$KTableMapValuesProcessor.process(KTableMapValues.java:83)
    at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:50)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.runAndMeasureLatency(ProcessorNode.java:244)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:133)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:143)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:129)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:90)
    at org.apache.kafka.streams.kstream.internals.ForwardingCacheFlushListener.apply(ForwardingCacheFlushListener.java:42)
    at org.apache.kafka.streams.state.internals.CachingKeyValueStore.putAndMaybeForward(CachingKeyValueStore.java:101)
    at org.apache.kafka.streams.state.internals.CachingKeyValueStore.access$000(CachingKeyValueStore.java:38)
    at org.apache.kafka.streams.state.internals.CachingKeyValueStore$1.apply(CachingKeyValueStore.java:83)
    at org.apache.kafka.streams.state.internals.NamedCache.flush(NamedCache.java:141)
    at org.apache.kafka.streams.state.internals.NamedCache.flush(NamedCache.java:99)
    at org.apache.kafka.streams.state.internals.ThreadCache.flush(ThreadCache.java:125)
    at org.apache.kafka.streams.state.internals.CachingKeyValueStore.flush(CachingKeyValueStore.java:123)
    at org.apache.kafka.streams.state.internals.InnerMeteredKeyValueStore.flush(InnerMeteredKeyValueStore.java:284)
    at org.apache.kafka.streams.state.internals.MeteredKeyValueBytesStore.flush(MeteredKeyValueBytesStore.java:149)
    at org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:239)
    ... 12 more
Caused by: java.lang.ClassCastException: java.lang.String cannot be cast to org.apache.avro.generic.GenericRecord
    at io.confluent.kafka.streams.serdes.avro.GenericAvroSerializer.serialize(GenericAvroSerializer.java:39)
    at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:156)
    at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:101)
    at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:89)
    ... 41 more

----- EDIT ------ i add tha class for serialize and deserialize

serializer:

  private class TupleSerializer implements Serializer<Tuple> {

    @Override
    public void configure(Map<String, ?> map, boolean bln) {
    }

    @Override
    public byte[] serialize(String string, Tuple t) {
         ByteBuffer buffer = ByteBuffer.allocate(4 + 4);
         buffer.putInt(t.occ);
         buffer.putInt(t.sum);
         return buffer.array();
    }

    @Override
    public void close() {
    }

}

deserializer:

 private class TupleDeserializer implements Deserializer<Tuple> {

    @Override
    public void configure(Map<String, ?> map, boolean bln) {
    }

    @Override
    public void close() {
    }

    @Override
    public Tuple deserialize(String string, byte[] bytes) {
         ByteBuffer buffer = ByteBuffer.wrap(bytes);
         int occ = buffer.getInt();
         int sum = buffer.getInt();
         return new Tuple (occ, sum);
    }

}

MySerde:

private class MySerde implements Serde<Tuple> {

    @Override
    public void configure(Map<String, ?> map, boolean bln) {
    }

    @Override
    public void close() {
    }

    @Override
    public Serializer<Tuple> serializer() {
        return new TupleSerializer ();
    }

    @Override
    public Deserializer<Tuple> deserializer() {
        return new TupleDeserializer ();
    }

}

I don't think you need to convert to JSON here... That is just going to slow the process down — OneCricketeer

Nishu Tayal Nishu Tayal · Accepted Answer · 2018-11-20T23:28:15

You have to define the Serdes with .to() method to override the default serde type.

average.to("average",Produced.with(Serdes.String(),Serdes.Double());

Please refer more details here :

https://docs.confluent.io/current/streams/developer-guide/dsl-api.html#writing-streams-back-to-kafka

Kafka Streams to topic

1 Answers