将新的接收器配置添加到 kafka 连接答案

【问题标题】：Adding new sink config to kafka connect将新的接收器配置添加到 kafka 连接
【发布时间】：2020-06-30 18:47:05
【问题描述】：

我正忙于学习 kafka，特别是使用 sink 连接器 (jdbc) 将数据从 kafka 主题发送到 mysql 的下游元素。

我已经使用 jdbc 连接器和 mysql jdbc 驱动程序从 confluent kafka-connect-base 映像构建了一个映像，在这样的 Dockerfile 中：

FROM confluentinc/cp-kafka-connect-base

RUN confluent-hub install --no-prompt confluentinc/kafka-connect-jdbc:5.4.1

ENV MYSQL_DRIVER_VERSION 5.1.39

RUN curl -k -SL "https://dev.mysql.com/get/Downloads/Connector-J/mysql-connector-java-${MYSQL_DRIVER_VERSION}.tar.gz" \
     | tar -xzf - -C /usr/share/java/kafka/ --strip-components=1 mysql-connector-java-5.1.39/mysql-connector-java-${MYSQL_DRIVER_VERSION}-bin.jar

老实说，我在文档中有点迷失了，我已经到了想要包含接收器配置的地步，但我不知道如何包含它或在哪里包含将其复制到。我已经创建了接收器配置文件，但不知道放在哪里。作为映像构建的一部分，还是在运行 kafka-connect 容器时？

最终的结果是为此连接器创建 helm 部署，但我还没有。

在我全面了解 Kafka 和 Kafka Connect 时，我们将不胜感激。

【问题讨论】：

标签： apache-kafka apache-kafka-connect confluent-platform

【解决方案1】：

由于您在分布式模式下使用 Kafka Connect（即generally the best option），因此您将连接器配置作为 REST 调用传递。

这是一个示例 JDBC 接收器配置 - 您需要修改它以适合您的源主题 serialisation 等：

curl -X PUT http://localhost:8083/connectors/sink_postgres_foo_00/config -H "Content-Type: application/json" -d '{
      "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
      "connection.url": "jdbc:mysql://mysql-host:3306/",
      "connection.user": "user",
      "connection.password": "pw",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "value.converter.schemas.enable": "true",
      "tasks.max": "1",
      "topics": "foo",
      "auto.create": "true",
      "auto.evolve":"true",
      "pk.mode":"none"          
    }'

如果您想在容器实例化时传递您的连接器配置，您可以将其嵌入到 start 命令中，该命令应该启动 worker，等待它可用，然后传递配置。 Here's an example。另见https://rmoff.net/2018/12/15/docker-tips-and-tricks-with-ksql-and-kafka/。

【讨论】：