Kafka 中的 Serde String[]

Question

我是卡夫卡的新手。我正在使用商店生成器并希望将两个元素的 String[] 作为与键关联的值。我这样设置商店：

StoreBuilder<KeyValueStore<String, String>> storeBuilder=Stores.keyValueStoreBuilder(
                Stores.persistentKeyValueStore(storeName),
                Serdes.String(),
                Serdes.String());
builder.addStateStore(storeBuilder);

当我调用将数据存入存储的方法时：

String []oldValues = store.get(v.getVessel());

我收到此错误：

Caused by: java.lang.ClassCastException: class java.lang.String cannot be cast to class [Ljava.lang.String; (java.lang.String and [Ljava.lang.String; are in module java.base of loader 'bootstrap')

我应该这样设置商店：

StoreBuilder<KeyValueStore<String, String[]>>

但我不知道用什么代替 Serdes.String()

Answer 1

您需要定义自己的 Serializer、Deserializer 和 Serde 子类来执行此操作。

您可以简单地使用带有 JSON 反序列化的纯字符串来创建字符串数组，或 CSV

Kafka 中的 Serde String[]

Serde String[] in Kafka

java

apache-kafka

apache-kafka-streams