Как настроить десериализатор для приложения Kafka Streams, когда сообщение было создано в .NET и имеет тип Confluent.Kaf ⇐ C#

[Anonymous]

Я пишу сообщение в тему «ITEM_PRICES», используя библиотеку dontnet(Confluent.kafka). Это сообщение должно быть парой «ключ-значение», где ключ — строка, а значение — число с плавающей запятой. Я настраиваю сериализаторы производителей следующим образом:

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

_producer = new ProducerBuilder(producerConfig)
.SetKeySerializer(Serializers.Utf8)
.SetValueSerializer(Serializers.Single)
.Build();

А потом создадим тему вот так:

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

_producer.Produce(ITEM_PRICES, new Confluent.Kafka.Message { Key= "ITEMNAME", Value = Convert.ToSingle(itemPrice)});

Мне нужно использовать эти сообщения в моем приложении потоков Kafka (Java) и попытаться сделать это следующим образом:

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

KStream itemPriceStream = streamsBuilder
.stream("ITEM_PRICES", Consumed.with(Serdes.String(), Serdes.Float()));
itemPriceStream .peek((key, value) -> System.out.println("Key: " + key + ", Value: " + value));

Но я получаю эту ошибку: размер данных, полученных десериализатором, не равен 4.
Мой вопрос: нужно ли мне создать собственный десериализатор в чтобы прочитать эти сообщения в моем потоке? Как мне это сделать?
Я попробовал десериализацию с помощью встроенных десериализаторов String и Float, но это не сработало.

Подробнее здесь: https://stackoverflow.com/questions/788 ... was-produc