java - 调试自定义 Kafka 连接器的简单有效方法是什么？

Question

我正在使用几个 Kafka 连接器，我在控制台输出中没有看到它们的创建/部署有任何错误，但是我没有得到我正在寻找的结果(没有任何结果) ，期望或其他)。我根据 Kafka 的示例 FileStream 连接器制作了这些连接器，因此我的调试技术基于示例中使用的 SLF4J 记录器的使用。我搜索了我认为会在控制台输出中生成的日志消息，但无济于事。我是不是在错误的地方寻找这些消息？或者是否有更好的方法来调试这些连接器？

我在实现中引用的 SLF4J 记录器的示例用法:

Kafka FileStreamSinkTask

Kafka FileStreamSourceTask

Answer 1

我会尽量从广义上回答你的问题。进行连接器开发的简单方法如下:

• 通过查看众多公开可用的 Kafka 连接器之一来构造和构建连接器源代码(您将在此处找到一个广泛的列表:https://www.confluent.io/product/connectors/)
• 从 https://www.confluent.io/download/ 下载最新的 Confluent 开源版本 (>= 3.3.0)

• 通过以下方式之一使您的连接器包可用于 Kafka Connect:

  1. 将所有连接器 jar 文件(连接器 jar 加上依赖项 jar，不包括 Connect API jar)存储到文件系统中的某个位置，并通过将此位置添加到 plugin.path Connect worker 属性中的属性。例如，如果您的连接器 jar 存储在 /opt/connectors/my-first-connector 中，您将设置 plugin.path=/opt/connectors在您 worker 的属性(property)中(见下文)。
  2. 将所有连接器 jar 文件存储在 ${CONFLUENT_HOME}/share/java 下的文件夹中.例如:${CONFLUENT_HOME}/share/java/kafka-connect-my-first-connector . (需要以 kafka-connect- 前缀开头才能被启动脚本选中)。 $CONFLUENT_HOME 是您安装 Confluent Platform 的位置。

• 可选地，通过在 ${CONFLUENT_HOME}/etc/kafka/connect-log4j.properties 中更改连接的日志级别来增加您的日志记录至 DEBUG甚至 TRACE .

• 使用 Confluent CLI 启动所有服务，包括 Kafka Connect。详情在这里:http://docs.confluent.io/current/connect/quickstart.html

  简要说明:confluent start

Note: The Connect worker's properties file currently loaded by the CLI is ${CONFLUENT_HOME}/etc/schema-registry/connect-avro-distributed.properties. That's the file you should edit if you choose to enable classloading isolation but also if you need to change your Connect worker's properties.

• Connect worker 运行后，通过运行启动连接器:

  confluent load <connector_name> -d <connector_config.properties>

  或

  confluent load <connector_name> -d <connector_config.json>

  连接器配置可以是 java 属性或 JSON 格式。

• 运行 confluent log connect打开 Connect worker 的日志文件，或通过运行直接导航到您的日志和数据的存储位置

  cd "$( confluent current )"

Note: change where your logs and data are stored during a session of the Confluent CLI by setting the environment variable CONFLUENT_CURRENT appropriately. E.g. given that /opt/confluent exists and is where you want to store your data, run:

export CONFLUENT_CURRENT=/opt/confluent
confluent current

• 最后，要以交互方式调试您的连接器，一种可能的方法是在开始使用 Confluent CLI 连接之前应用以下内容:

  confluent stop connect
export CONNECT_DEBUG=y; export DEBUG_SUSPEND_FLAG=y;
confluent start connect

  然后连接您的调试器(例如远程连接到 Connect worker(默认端口:5005)。要停止在 Debug模式下运行连接，只需在完成后运行:unset CONNECT_DEBUG; unset DEBUG_SUSPEND_FLAG;。

我希望以上内容能让您的连接器开发更轻松、更有趣!