代码之家 › 专栏 › 技术社区 › user123

PostgreSQL和Kafka Connect集成问题

apache-kafka-connect apache-kafka postgresql

user123 · 技术社区 · 8 年前

我正在测试JDBC接收器连接器,以便将记录从Kafka转储到PostgreSQL。以下是连接器配置:

{
    "name": "jdbc-sink-postgresql-1",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "tasks.max": "1",
        "topics": "role",
        "connection.url": "jdbc:postgresql://localhost:5432/postgres?user=&password=",
        "auto.create": "false",                                                   
        "insert.mode": "upsert",
        "mode":"incrementing",
        "table.name.format":"role",
        "pk.mode":"record_value",
        "pk.fields":"role_id"
    }
}

当我运行连接器时,我得到以下异常:

java.sql.BatchUpdateException: Batch entry 1 INSERT INTO "role" ("role_id","role_name") VALUES (123,'admin') ON CONFLICT ("role_id") DO UPDATE SET "role_name"=EXCLUDED."role_name" was aborted.  
   Call getNextException to see the cause.
   at org.postgresql.jdbc2.AbstractJdbc2Statement$BatchResultHandler.handleError(AbstractJdbc2Statement.java:2778))

有没有关于我在这里遗漏了什么的建议?如果需要更多信息,请告诉我。

1 回复 | 直到 8 年前

user123 8 年前

所以,问题出在桌子上。这就是我最初创建表的方式:

CREATE TABLE role(
 role_id int PRIMARY KEY,
 role_name VARCHAR (255) UNIQUE NOT NULL
);

主题中的测试数据如下所示:

./bin/kafka-avro-console-producer --broker-list localhost:9092 --topic role --property schema.registry.url=http://localhost:8081/  --property value.schema='{"type":"record","name":"myRecord","fields": [{"name": "role_id","type": "int"},{"name": "role_name","type": "string"}]}' --key-serializer org.apache.kafka.common.serialization.StringSerializer --value-serializer io.confluent.kafka.serializers.KafkaAvroSerializer --property print.key=true
{"role_id":122, "role_name":"admin"}
{"role_id":123, "role_name":"admin"}
{"role_id":124, "role_name":"admin"}
{"role_id":125, "role_name":"admin"}
{"role_id":126, "role_name":"admin"}

因此,当我的测试数据一次又一次地为role_name字段提供相同的值时,它违反了惟一约束,因此出现了错误。

我都做了什么?

我把桌子掉了。

创建了一个没有唯一键约束的新表,上述数据被毫无问题地推送到PostgreSQL。

推荐文章

Narut Promsuparoj · Kafka连接器时间戳转换器与SMT一起使用时不工作

2 年前

shyaka gen · 连接器类无效。检查连接器配置文件(Kafka Connect Jdbc)

3 年前

Muhammad Affan · Kafka Connect中面临“由于序列化错误,将byte[]转换为Kafka Connect数据失败”

3 年前

Hongbo Miao · java.lang.ClassNotFoundException:Kafka Connect中的io.apicurio.registry.serde.avro.AvroKafkaSerializer

3 年前

Vaibhav Sharma · 使用Servlet连接KafkaProducer

8 年前

Evaldas Buinauskas · Kafka Connect将JSON字符串转换为实际JSON

9 年前

Nandish Kotadia · 我可以编写自定义kafka connect转换来将JSON转换为AVRO吗?

9 年前

Cody Ferguson · PubSub-Kafka连接节点连接文件结束异常

9 年前

Jaime Caffarel Daniel Koverman · kafka SourceTask类的poll()方法是否线程安全?

9 年前

Warsame · confluent-3.2.0 S3连接器默认分区设置

9 年前