我需要在模式注册表下注册kafka流更改日志主题的模式吗

Question

我正在使用处理器API和kafka StreamDSL实现kafka流项目。我在处理器中的处理功能是

@Override
public void process(final String key, final T event) {
keyValueStore.put(key, event);
}

我的拓扑是

protected Topology buildTopology() {
final StreamsBuilder builder = new StreamsBuilder();
KeyValueBytesStoreSupplier storeSupplier = Stores.persistentKeyValueStore(stateStoreName);
StoreBuilder<KeyValueStore<String, T>> storeBuilder =
Stores.keyValueStoreBuilder(storeSupplier, Serdes.String(), Serdes.serdeFrom(
new EventSerializer(streamProperties()),
new EventDeserializer(streamProperties())));
builder.addStateStore(storeBuilder);
final KStream<String, T> stream = builder.stream(inputTopic);
stream.process(() -> new Processor<>(stateStoreName), stateStoreName);
stream.to(outputTopic);
return builder.build();
}

最后，这是我的自定义EventSerializer类：

public class EventSerializer<T extends SpecificRecordBase & SpecificRecord>
implements Serializer<T> {
private final KafkaAvroSerializer inner;
public EventSerializer(Map<String, ?> properties) {
inner = new KafkaAvroSerializer();
configure(properties, false);
}
@Override
public void configure(Map<String, ?> configs, boolean isKey) {
inner.configure(EventSerdeConfig.withProducerConfig(configs), isKey);
}
@Override
public byte[] serialize(final String topic, final T record) {
return inner.serialize(topic, record);
}
}

当处理器将事件放入keyValueStore时，我遇到错误io.confluent.rest.exceptions.RestNotFoundException：未找到主题经过一段时间的调试，我意识到这是因为序列化程序在序列化事件时遇到了问题。函数public byte[] serialize(final String topic, final T record)中的主题是application id-store-changelog。这是卡夫卡内心的行为，尽管我不知道为什么。Serialzer找不到此组合主题的架构，因此引发错误。我需要为这个组合主题注册模式吗？或者有没有任何方法可以将真正的使用者主题传递给已经注册了模式的serializer？

Answer 1

当您有new KafkaAvroSerializer();时，它默认指向架构注册表的localhost:8081。

您不需要注册(尽管可以(，因为Producer将其作为inner.serialize序列化逻辑的一部分进行注册

注意：扩展KafkaAvroSerializer可能更有意义