Kafka Connect，JdbcSinkConnector - Getting "检索id为1的Avro架构时出错，找不到主题。；错误代码:40401 "

Question

我创建了一个NiFi流，该流最终将json记录发布为具有Avro编码值和字符串键的记录，使用了值模式的融合注册表中的模式。这是NiFi中AvroRecordSetWriter的配置。

我现在正在尝试使用Kafka Connect（Connect-独立）使用JdbcSinkConnector将消息移动到PostgreSQL数据库，但收到以下错误：检索id 1的Avro架构时出错

我已经确认我的Confluent注册表中有一个架构，ID为1。以下是我对Connect任务的配置

工人配置：

bootstrap.servers=localhost:29092
key.converter.schemas.enable=false
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
offset.storage.file.filename=/tmp/connect.offsets
rest.host.name=localhost
rest.port=8083
plugin.path=share/java

连接器配置：

name=pg-sink
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
topics=rds
connection.url=jdbc:postgresql://localhost:5432/test
connection.user=postgres
connection.password=xxxxxxxx
insert.mode=upsert
table.name.format=test_data
auto.create=true

我在NiFi中创建了一个正确使用消息的流，并且通过指定＜code＞--property schema.registry.url，我还成功地使用kafka avro console consumer使用了消息（在输出中格式化为JSON）=http://schema-registry:8081。请注意，我在Docker容器中运行消费者，这就是为什么url不是localhost。

我不确定我错过了什么。我唯一的想法是我对键转换器使用了错误的类，但是这对于给定的错误没有意义。有人能看出我做错了什么吗？

Answer 1

我对Nifi了解不多，但我看到模式的名称是“rds”，并且在错误日志中说它没有在模式注册表中找到主题。

Kafka使用KafkaAvroSerializer序列化avro记录，同时在模式注册表中注册相关的avro模式。它使用KafkaAvroDeserializer反序列化avro记录并从模式注册表中检索相关的模式。

Schema registry将模式存储到称为“主题”的类别中，为记录命名主题的默认行为是:对于值记录为< code>topic_name-value，对于键为< code>topic_name-key。

在你的情况下，你没有向Kafka注册架构，而是向Nifi注册了架构，所以我的猜测是名称“rds”出现在架构注册表中，或者是架构注册表上的使用者名称。

您如何验证您的架构是否被严格存储？

通常情况下，正确的主题是＜code＞rds-value＜/code＞，因为您只对值记录使用模式注册表。