Kafka Streams – Почему я не могу агрегировать и суммировать свои длинные данные? ⇐ JAVA

[Anonymous]

Я новичок в Kafka Streams и пытаюсь собрать свое первое приложение.
Я хотел бы сложить суммы моих BankTransactions.

Код: Выделить всё

@Bean
public KStream kStream(StreamsBuilder kStreamBuilder) {
JsonSerde jsonSerde = new JsonSerde(BankTransaction.class);
jsonSerde.configure(Map.of("spring.json.type.mapping", "BankTransaction:nl.sourcelabs.kafkasolo.streamprocessing.BankTransaction"), false);
return kStreamBuilder.stream("transactions", Consumed.with(Serdes.String(), jsonSerde))
.mapValues((readOnlyKey, value) -> value.amount)
.groupByKey(Grouped.with(Serdes.String(), Serdes.Long()))
.aggregate(() -> 0L, (key, value, aggregate) -> aggregate + value,
Materialized.as("transactionAmountCount")
.withKeySerde(Serdes.String())
.withValueSerde(Serdes.Long()))
.toStream();

}

Событие считывается из темы и десериализуется правильно.
Но приложение вылетает со следующей ошибкой:

Код: Выделить всё

org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_0, processor=KSTREAM-SOURCE-0000000000, topic=transactions, partition=0, offset=0, stacktrace=org.apache.kafka.common.errors.SerializationException: Size of data received by IntegerDeserializer is not 4
at org.apache.kafka.common.serialization.IntegerDeserializer.deserialize(IntegerDeserializer.java:30)
at org.apache.kafka.common.serialization.IntegerDeserializer.deserialize(IntegerDeserializer.java:24)
at org.apache.kafka.streams.state.internals.ValueAndTimestampDeserializer.deserialize(ValueAndTimestampDeserializer.java:58)
at org.apache.kafka.streams.state.internals.ValueAndTimestampDeserializer.deserialize(ValueAndTimestampDeserializer.java:31)
at org.apache.kafka.streams.state.StateSerdes.valueFrom(StateSerdes.java:163)
at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.outerValue(MeteredKeyValueStore.java:427)
at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.lambda$get$5(MeteredKeyValueStore.java:318)
at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:887)
at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.get(MeteredKeyValueStore.java:318)
at org.apache.kafka.streams.processor.internals.AbstractReadWriteDecorator$KeyValueStoreReadWriteDecorator.get(AbstractReadWriteDecorator.java:92)
at org.apache.kafka.streams.state.internals.KeyValueStoreWrapper.get(KeyValueStoreWrapper.java:76)
at org.apache.kafka.streams.kstream.internals.KStreamAggregate$KStreamAggregateProcessor.process(KStreamAggregate.java:107)
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:152)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:291)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:270)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:229)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:216)
at org.apache.kafka.streams.kstream.internals.KStreamMapValues$KStreamMapProcessor.process(KStreamMapValues.java:42)
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:154)

Я не понимаю, как LongDeserializer получает однобайтовое хранилище KeyValueStore, когда в него помещены только длинные значения.
Любая помощь будет очень полезна признателен.
Я пробовал отладить код, но все, что я вижу, это то, что журнал изменений транзакцииAmountCount читается, и на этом все рушится.

Подробнее здесь: https://stackoverflow.com/questions/791 ... p-my-longs