如何使 Kafka Sink 连接器与 Avro 序列化键和值一起使用到 postgres [关闭]

Question

我有一个 Kafka 主题，其中包含带有 Avro 序列化键和 Avro 序列化值的消息。

我正在尝试设置一个接收器连接器以将这些值放入 postgres 数据库（在本例中为 AWS RDS）中的表中。

我已经尝试了许多关于主题、消息和接收器配置本身的变体，但是看看下面的例子，如果有人可以提供关于我哪里出错的指导，那就太好了！ :)

我的主题具有以下架构（在架构注册表中）...

密钥架构

{
    "type": "record",
    "name": "TestTopicKey",
    "namespace": "test.messaging.avro",
    "doc": "Test key schema.",
    "fields": [
        {
            "name": "unitId",
            "type": "int"
        }
    ]
}

值模式

{
    "type": "record",
    "name": "TestTopicValues",
    "namespace": "test.messaging.avro",
    "doc": "Test value schema.",
    "fields": [
        {
            "name": "unitPrice",
            "type": "int",
            "doc": "Price in AUD excluding GST."
        },
        {
            "name": "unitDescription",
            "type": "string"
        }
    ]
}

我正在使用“kafka-avro-console-producer”手动生成主题记录，如下所示：

/bin/kafka-avro-console-producer --broker-list kafka-box-one:9092 --topic test.units --property parse.key=true --property "key.separator=|" --property "schema.registry.url=http://kafka-box-one:8081" --property key.schema='{"type":"record","name":"TestTopicKey","namespace":"test.messaging.avro","doc":"Test key schema.","fields":[{"name":"unitId","type":"int"}]}' --property value.schema='{"type":"record","name":"TestTopicValues","namespace":"test.messaging.avro","doc":"Test value schema.","fields":[{"name":"unitPrice","type":"int","doc":"Price in AUD excluding GST."},{"name":"unitDescription","type":"string"}]}'

一旦 Producer 启动，我就可以成功地将记录添加到主题，如下所示：

{"unitId":111}|{"unitPrice":15600,"unitDescription":"A large widget thingy."}

注意：我也可以按预期成功使用 kafka-avro-console-consumer 消费。

我试图沉入的 postgres 表如下所示：

CREATE TABLE test_area.unit_prices (
    unitId int4 NOT NULL,
    unitPrice int4 NULL,
    unitDescription text NULL,
    CONSTRAINT unit_prices_unitid_pk PRIMARY KEY (unitId)
);

我的接收器连接器如下所示：

{
  "name": "test.area.unit.prices.v01",
  "config": {
      "connector.class": "JdbcSinkConnector",
      "topics": "test.units",
      "group.id": "test.area.unit.prices.v01",
      "key.converter": "io.confluent.connect.avro.AvroConverter",
      "key.converter.schema.registry.url": "http://kafka-box-one:8081",
      "value.converter": "io.confluent.connect.avro.AvroConverter",
      "value.converter.schema.registry.url": "http://kafka-box-one:8081",
      "connection.user": "KafkaSinkUser",
      "connection.password": "KafkaSinkPassword",
      "connection.url": "jdbc:postgresql://unit-catalogue.abcdefghij.my-region-1.rds.amazonaws.com:5432/unit_sales?currentSchema=test_area",
      "table.name.format": "unit_prices",
      "auto.create": false,
      "auto.evole": "false"
  }
}

我的期望是记录会在 Sink 显示为正在运行后不久出现在 postgres 表中。然而，什么都没有下沉。

补充说明：

• 我可以通过使用 usql 的 Sink 连接器使用凭据发布此 Sink 连接器的 Kafka Connect 框连接和写入 postgres RDS 实例。
• 接收器连接器状态为“正在运行”，这表明接收器语法中没有错误。

Answer 1

对于迟到的回复对此表示歉意。在最终使日志正常工作后，这是一个代理问题。谢谢大家的帮助。