【发布时间】：2019-11-15 13:41:22
【问题描述】：
我想将发送到主题的数据发送到 postgresql 数据库。所以我关注this guide 并像这样配置了属性文件：
name=transaction-sink
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
topics=transactions
connection.url=jdbc:postgresql://localhost:5432/db
connection.user=db-user
connection.password=
auto.create=true
insert.mode=insert
table.name.format=transaction
pk.mode=none
我用
开始连接器./bin/connect-standalone etc/schema-registry/connect-avro-standalone.properties etc/kafka-connect-jdbc/sink-quickstart-postgresql.properties
sink-connector 已创建，但由于此错误而无法启动：
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!
架构采用 avro 格式并已注册，我可以向主题发送（生成）消息并从中读取（使用）。但我似乎无法将其发送到数据库。
这是我的./etc/schema-registry/connect-avro-standalone.properties
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081
这是使用 java-api 提供主题的生产者：
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
properties.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
try (KafkaProducer<String, Transaction> producer = new KafkaProducer<>(properties)) {
Transaction transaction = new Transaction();
transaction.setFoo("foo");
transaction.setBar("bar");
UUID uuid = UUID.randomUUID();
final ProducerRecord<String, Transaction> record = new ProducerRecord<>(TOPIC, uuid.toString(), transaction);
producer.send(record);
}
我正在验证数据是否正确序列化和反序列化使用
./bin/kafka-avro-console-consumer --bootstrap-server localhost:9092 \
--property schema.registry.url=http://localhost:8081 \
--topic transactions \
--from-beginning --max-messages 1
数据库已启动并正在运行。
【问题讨论】：
-
您为 Producer 设置的
KEY_SERIALIZER_CLASS_CONFIG是什么？ -
更新了我的问题。 StringSerializer.class.
-
Bingo :) 我已经更新了我的答案。
-
你太棒了。 ：-） 有用。我必须转换消息，因为我使用 BigDecimal，它在 avro 中是字符串类型，java-class java.math.BigDecimal 来规避 STRUCT 类型没有映射到 SQL 错误。我会看一些链接。
-
更改 key.converter 后，我尝试将主题下沉数据放到 postgresql 表中。一个看似简单的任务，因为我没有嵌套数据。但它会因错误“不支持的源数据类型：STRUCT”而停止。我看过docs.confluent.io/3.1.1/connect/connect-jdbc/docs/… 和stackoverflow.com/questions/44385722/…，但后者似乎无关。属性位于顶部。我可以编辑错误吗？
标签： apache-kafka apache-kafka-connect