如何指定连接器是源还是接收器？

Question

我目前正在配置kafka连接（与debezium/连接docker映像），我成功地使用环境变量将其连接到Kafka：

docker run -it --rm --name AAAAAA-kafka-connect -p 8083:8083 \
    -v aaaaa.jks:aaaaa.jks \
    -v bbbbbb.jks:bbbbbb.jks \
    -e LOG_LEVEL=INFO \
    -e HOST_NAME="AAAAAA-kafka-connect" \
    -e HEAP_OPTS="-Xms256m -Xmx2g" \
    -e BOOTSTRAP_SERVERS="BBBBB:9092" \
    -e CONNECT_CLIENT_ID="xxx-kafka-connect" \
    -e CONNECT_SASL_JAAS_CONFIG="org.apache.kafka.common.security.scram.ScramLoginModule required username=\"...\" password=\"...\";" \
    -e CONNECT_SECURITY_PROTOCOL="SASL_SSL" \
    -e CONNECT_SASL_MECHANISM="PLAIN" \
    -e CONNECT_SSL_TRUSTSTORE_LOCATION="bbbbbb.jks" \
    -e CONNECT_SSL_TRUSTSTORE_PASSWORD="..." \
    -e CONNECT_SSL_KEYSTORE_LOCATION="aaaaa.jks" \
    -e CONNECT_SSL_KEYSTORE_PASSWORD="..." \
    -e GROUP_ID="XXX.grp.kafka.connect" \
    -e CONFIG_STORAGE_TOPIC="XXX.connect.configs.v1" \
    -e OFFSET_STORAGE_TOPIC="XXX.connect.offsets.v1" \
    -e STATUS_STORAGE_TOPIC="XXX.connect.statuses.v1" \
    quay.io/debezium/connect:1.9

现在我必须创建一个源连接器（posgresql-db），我希望kafka connect从源获取的数据在kafka主题中接收。

由于数据库连接器的 json 配置中没有这样的配置，我必须在哪里设置接收器的 kafka 配置？

我必须创造一个连接Kafka主题的接收器吗？如果是，我们在哪里指定这是一个接收器连接器还是一个源连接器？？

PS:我已经创建了kafka主题，我想把数据放在那里

请随意提问

Answer 1

好的，你必须添加CONNECT_PRODUCER_*或CONNECT_CONSUMER_*环境变量来指定源或接收器的配置！！！！！！

像这样:

docker run -it --rm --name AAAAAA-kafka-connect -p 8083:8083 \
    -v aaaaa.jks:aaaaa.jks \
    -v bbbbbb.jks:bbbbbb.jks \
    -e LOG_LEVEL=INFO \
    -e HOST_NAME="AAAAAA-kafka-connect" \
    -e HEAP_OPTS="-Xms256m -Xmx2g" \
    -e BOOTSTRAP_SERVERS="BBBBB:9092" \
    -e CONNECT_CLIENT_ID="xxx-kafka-connect" \
    -e CONNECT_SASL_JAAS_CONFIG="org.apache.kafka.common.security.scram.ScramLoginModule required username=\"...\" password=\"...\";" \
    -e CONNECT_SECURITY_PROTOCOL="SASL_SSL" \
    -e CONNECT_SASL_MECHANISM="PLAIN" \
    -e CONNECT_SSL_TRUSTSTORE_LOCATION="bbbbbb.jks" \
    -e CONNECT_SSL_TRUSTSTORE_PASSWORD="..." \
    -e CONNECT_SSL_KEYSTORE_LOCATION="aaaaa.jks" \
    -e CONNECT_SSL_KEYSTORE_PASSWORD="..." \
    -e GROUP_ID="XXX.grp.kafka.connect" \
    -e CONFIG_STORAGE_TOPIC="XXX.connect.configs.v1" \
    -e OFFSET_STORAGE_TOPIC="XXX.connect.offsets.v1" \
    -e STATUS_STORAGE_TOPIC="XXX.connect.statuses.v1" \
    -e CONNECT_PRODUCER_TOPIC_CREATION_ENABLE=false \
    -e CONNECT_PRODUCER_SASL_JAAS_CONFIG="org.apache.kafka.common.security.scram.ScramLoginModule required username=\"...\" password=\"...\";" \
    -e CONNECT_PRODUCER_SECURITY_PROTOCOL="SASL_SSL" \
    -e CONNECT_PRODUCER_SASL_MECHANISM="PLAIN" \
    -e CONNECT_PRODUCER_SSL_TRUSTSTORE_LOCATION="bbbbbb.jks" \
    -e CONNECT_PRODUCER_SSL_TRUSTSTORE_PASSWORD="..." \
    -e CONNECT_PRODUCER_SSL_KEYSTORE_LOCATION="aaaaa.jks" \
    -e CONNECT_PRODUCER_SSL_KEYSTORE_PASSWORD="..." \
    -e CONNECT_PRODUCER_CLIENT_ID="xxx-kafka-connect" \
    -e CONNECT_PRODUCER_TOPIC_CREATION_ENABLE=false \
    quay.io/debezium/connect:1.9

sink或source属性来自连接器。在连接器的json定义中使用的类。然而，Debeziums CDC连接器只能用作从外部数据库系统捕获实时事件更改记录的源连接器(https://hevodata.com/learn/debezium-vs-kafka-connect/#:~：text=Debezium平台有大量来自外部数据库系统的记录。）

Answer 2

环境变量仅修改客户端参数。

源和汇是在实际创建连接器时确定的。您需要一个JSON配置，它将有一个< code>connector.class。

在Kafka API中有< code>SinkTask和< code>SourceTask。

Debezium始终是一个来源。来源写信给Kafka；这并不意味着Kafka就是一个水槽。你需要安装一个新的连接器插件来为你的数据库获取一个接收器，比如来自Confluent的JDBC连接器，它有源和接收器的类。