用于postgres - Day0加载的Kafka源连接器

Question

我正在寻找一个从Postgres到卡夫卡的Day0加载卡夫卡源连接器。

遇到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配置？

任何帮助都将不胜感激。

Answer 1

1.配置

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";

来源: StackOverflow answer“Kafka connect中的ACL配置不起作用”

参考：Kafka Connect Security docs

1.2。对于SASL镜像，您可以尝试通过环境变量直接传递debezium/connect配置(使用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"
}'