如何覆盖 Aiven JdbcSinkConnector 上的默认退避期和最大重试次数

Question

我正在使用 Aiven JdbcSinkConnector (io.aiven.connect.jdbc.JdbcSinkConnector) 连接到 Postgres DB。

我想设置自定义退避时间和最大重试次数，但这似乎不可能，它总是无法达到默认值（10000 毫秒和 3）。

有人知道如何以及在哪里可以设置用于覆盖此属性的属性吗？

这是我的数据库不可用时收到的异常：

Unable to connect to database on attempt 1/3. Will retry in 10000 ms. (io.aiven.connect.jdbc.util.CachedConnectionProvider:97)

我正在使用以下 POST 命令创建接收器连接器，但这不起作用：

curl -X POST -H "Content-Type: application/json" --data '{
    "name": "jdbc-sink-my-topic", "config": {
    "connector.class": "io.aiven.connect.jdbc.JdbcSinkConnector",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://localhost:8081",
    "tasks.max": 1,
    "connection.url": "jdbc:postgresql://<private_ip>/<db_name>?currentSchema=<schema_name>",
    "connection.user": <username>,
    "connection.password": <password>,
    "connection.attempts": 5,
    "connection.backoff.ms": 20000,
    "topics": <topic_name>,
    "table.name.format": <db_table_name>,
    "insert.mode": "multi",
    "pk.mode": "none",
    "auto.create": "false",
    "errors.tolerance": "all",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true"}}' http://localhost:8083/connectors | jq

Answer 1

不支持。使用退避的构造函数仅用于 JDBCSource Task，而不用于 Sink。

https://github.com/Aiven-Open/jdbc-connector-for-apache-kafka/blob/master/src/main/java/io/aiven/connect/jdbc/util/CachedConnectionProvider.java#L43- L59