用于 postgres 的 Kafka 源连接器 - 第 0 天加载

Posted 2023-03-23

【中文标题】用于 postgres 的 Kafka 源连接器 - 第 0 天加载

【英文标题】：Kafka source connector for postgres - Day0 load

【发布时间】：2021-04-23 11:20:23

【问题描述】：

我正在寻找一个用于从 Postgres 到 Kafka 的 Day0 加载的 Kafka 源连接器。

遇到了 Debezium postgres 连接器。

Docker 镜像，

debezium/connect:1.4

docker run -it --rm --name postgres-connect -p 8083:8083 -e BOOTSTRAP_SERVERS=host1:8080 -e GROUP_ID=test-1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses debezium/connect:1.4

如何传递 postgres 主机详细信息和 kafka sasl 配置？

任何帮助将不胜感激。

【问题讨论】：

SASL 配置：请看答案***.com/a/64355810/7109598

Postgres 主机详细信息应通过连接器配置传递

@IskuskovAlexander 我是 kafka connect 的新手。你能告诉我如何将 SASL 配置文件和 postgress 配置传递给 docker 容器吗？

请在发布的答案中查看一些详细信息

【参考方案1】：

1. SASL 配置

1.1。通常情况下，您需要将以下属性添加到您的connect-distributed.properties：

sasl.mechanism=PLAIN
security.protocol=SASL_PLAINTEXT

sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="connect" \
  password="connect-secret";

producer.sasl.mechanism=PLAIN
producer.security.protocol=SASL_PLAINTEXT
producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="connect" \
  password="connect-secret";

consumer.sasl.mechanism=PLAIN
consumer.security.protocol=SASL_PLAINTEXT
consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="connect" \
  password="connect-secret";

来源：*** answer“Kafka 连接中的 ACL 配置不起作用”

参考：Kafka Connect Security docs

1.2。对于debezium/connect docker 映像，您可以尝试通过环境变量直接传递 SASL 配置（使用these 转换步骤）：

docker run -it --rm --name postgres-connect -p 8083:8083 \
  -e BOOTSTRAP_SERVERS=host1:8080 -e GROUP_ID=test-1 \
  -e CONFIG_STORAGE_TOPIC=my_connect_configs \
  -e OFFSET_STORAGE_TOPIC=my_connect_offsets \
  -e STATUS_STORAGE_TOPIC=my_connect_statuses \
  -e CONNECT_SASL_MECHANISM=PLAIN \
  -e CONNECT_SECURITY_PROTOCOL=SASL_PLAINTEXT \
  -e CONNECT_SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="connect" password="connect-secret"; \
  -e CONNECT_PRODUCER_SASL_MECHANISM=PLAIN \
  -e CONNECT_PRODUCER_SECURITY_PROTOCOL=SASL_PLAINTEXT \
  -e CONNECT_PRODUCER_SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="connect" password="connect-secret"; \
  -e CONNECT_CONSUMER_SASL_MECHANISM=PLAIN \
  -e CONNECT_CONSUMER_SECURITY_PROTOCOL=SASL_PLAINTEXT \
  -e CONNECT_CONSUMER_SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="connect" password="connect-secret"; \
debezium/connect:1.4

2。 PostgreSQL 主机配置

主机详细信息应使用连接器配置通过 Kafka Connect REST API 传递：

curl -i -X PUT -H "Content-Type:application/json" \
    http://localhost:8083/connectors/debezium_postgres_source/config \
    -d '
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "database.hostname": "source-db",
    "database.port": "5432",
    "database.user": "postgresusersource",
    "database.password": "postgrespw",
    "database.dbname" : "sourcedb",
    "database.server.name": "dbserver1"
'    

【讨论】：

为 Postgres 配置调用 REST API --> 我们是否需要在 docker 容器由于某种原因重新启动时调用它？

连接器存储在 Kafka 中的所有元数据。因此，如果您没有丢失此信息，我不必再次调用此方法。您还可以通过 REST API (curl -XGET localhost:8083/connectors/debezium_postgres_source/status) 检查连接器的状态