如何使 Instaclustr Kafka Sink Connector 与 Avro 序列化值一起使用到 postgres？答案

【问题标题】：How to make Instaclustr Kafka Sink Connector work with Avro serialized value to postgres?如何使 Instaclustr Kafka Sink Connector 与 Avro 序列化值一起使用到 postgres？
【发布时间】：2020-05-14 08:33:04
【问题描述】：

我有一个 Avro 序列化值的 Kafka 主题。

我正在尝试设置 JDBC(postgres) 接收器连接器以将这些消息转储到 postgres 表中。

但是，我遇到了错误。

"org.apache.kafka.common.config.ConfigException: Invalid value io.confluent.connect.avro.AvroConverter for configuration value.converter: Class io.confluent.connect.avro.AvroConverter could not be found."

我的 Sink.json 是

{"name": "postgres-sink",
  "config": {
    "connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max":"1",
    "topics": "<topic_name>",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "instaclustr_schema_registry_host:8085",
    "connection.url": "jdbc:postgresql://postgres:5432/postgres?currentSchema=local",
    "connection.user": "postgres",
    "connection.password": "postgres",
    "auto.create": "true",
    "auto.evolve":"true",
    "pk.mode":"none",
    "table.name.format": "<table_name>"
  }
}

另外，我在 connect-distributed.properties（引导服务器）中进行了更改。

我正在运行的命令是 -

curl -X POST -H "Content-Type: application/json" --data @postgres-sink.json https://<instaclustr_schema_registry_host>:8083/connectors

【问题讨论】：

标签： postgresql apache-kafka avro apache-kafka-connect

【解决方案1】：

io.confluent.connect.avro.AvroConverter 不是 Apache Kafka 发行版的一部分。您可以将 Apache Kafka 作为 Confluent Platform 的一部分运行（它随转换器一起提供并且更容易），或者您可以download it separately 并自己安装它。

【讨论】：