如何配置带SSL证书认证的Kafka Connect JDBC Sink PostgreSQL连接器

现有一个基于用户名密码认证的PostgreSQL JDBC Sink连接器可用配置：

{
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "connection.url": "jdbc:postgresql://localhost:5432/postgres",
    "connection.user": "postgres-user",
    "connection.password": "postgres-pw",
    "topics.regex": "my-topic",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",    
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "auto.create": "true",
    "insert.mode": "upsert",
    "auto.evolve": "true",
    "table.name.format" : "${topic}",
    "pk.mode": "record_key",
    "pk.fields": "id"
}

需要将该连接器连接到要求用户名、.cert客户端证书文件和.key客户端密钥文件的SSL认证PostgreSQL服务器，但仅在文档中看到ssl.rootcertfile配置选项，想确认是否可以在连接器配置中设置.cert和.key文件？

可以实现，但不是通过连接器的顶层配置项，而是通过PostgreSQL JDBC URL的参数来传递客户端证书和密钥的配置——这些参数属于PostgreSQL JDBC驱动的SSL配置范畴，而非Kafka Connect JDBC连接器的专属配置。

你需要修改connection.url，添加以下关键参数：

• ssl=true：启用SSL连接
• sslcert=/path/to/your/client.cert：指定客户端证书文件的绝对路径
• sslkey=/path/to/your/client.key：指定客户端密钥文件的绝对路径
• sslrootcert=/path/to/your/root.crt：指定CA根证书文件的绝对路径（用于验证服务器证书）
• 如果你的客户端密钥文件设置了密码，还需要添加sslpassword=your-key-password

修改后的完整配置示例如下：

{
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "connection.url": "jdbc:postgresql://your-postgres-host:5432/postgres?ssl=true&sslcert=/path/to/client.cert&sslkey=/path/to/client.key&sslrootcert=/path/to/root.crt",
    "connection.user": "postgres-user",
    "topics.regex": "my-topic",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",    
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "auto.create": "true",
    "insert.mode": "upsert",
    "auto.evolve": "true",
    "table.name.format" : "${topic}",
    "pk.mode": "record_key",
    "pk.fields": "id"
}

注意事项：

• 确保Kafka Connect进程对指定的证书、密钥文件有读取权限
• 路径必须是绝对路径，避免因工作目录变化导致文件找不到
• 如果密钥文件是PKCS#12格式，需要改用sslkeyfactory参数指定对应的密钥工厂，同时调整证书和密钥的参数格式

内容的提问来源于stack exchange，提问作者Noy Gafni