Kafka Connect MongoDB Sink自定义写入策略集成指南-java教程-PHP中文网

Kafka Connect MongoDB Sink自定义写入策略集成指南

本文深入探讨了在Kafka Connect MongoDB Sink连接器中集成自定义写入策略时常见的“类未找到”错误及其解决方案。核心在于理解Kafka Connect的插件加载机制，并正确配置plugin.path以确保自定义JAR包及其依赖能够被正确识别，避免不必要的JAR文件复制操作，从而实现灵活的数据写入逻辑。

1. 理解Kafka Connect MongoDB Sink的自定义写入策略

kafka connect mongodb sink连接器提供了高度的灵活性，允许用户通过实现自定义的writemodelstrategy接口来控制数据如何从kafka主题写入到mongodb。这对于需要复杂插入、更新或删除逻辑的场景至关重要，例如基于特定业务规则进行幂等更新、数据合并或条件写入。

WriteModelStrategy接口的核心是createWriteModel方法，它接收一个SinkDocument对象（包含Kafka消息的键和值），并返回一个MongoDB的WriteModel实例（如InsertOneModel、UpdateOneModel、DeleteOneModel等）。

1.1 自定义写入策略示例

以下是一个简单的自定义写入策略CustomWriteModelStrategy示例，它演示了如何根据传入的Kafka消息值构建一个MongoDB的UpdateOneModel，实现消息计数和消息列表的更新：

package com.fu.connect.sink;

import org.bson.*;
import com.mongodb.client.model.UpdateOneModel;
import com.mongodb.client.model.UpdateOptions;
import com.mongodb.client.model.WriteModel;
import com.mongodb.kafka.connect.sink.converter.SinkDocument;
import com.mongodb.kafka.connect.sink.writemodel.strategy.WriteModelStrategy;
import org.apache.kafka.connect.errors.DataException;

public class CustomWriteModelStrategy implements WriteModelStrategy {

    private static final UpdateOptions UPDATE_OPTIONS = new UpdateOptions().upsert(true);

    @Override
    public WriteModel<BsonDocument> createWriteModel(SinkDocument document) {
        // 尝试获取SinkDocument的值部分，如果缺失则抛出异常
        BsonDocument valueDoc = document.getValueDoc().orElseThrow(
                () -> new DataException("Error: cannot build the WriteModel since the value document was missing unexpectedly"));

        // 从传入文档中提取“message”字段
        BsonString message = new BsonString("");
        if (valueDoc.containsKey("message")) {
            message = valueDoc.get("message").asString();
        }

        // 定义更新操作的过滤条件：查找counter小于10的文档
        BsonDocument filters = new BsonDocument("counter", new BsonDocument("$lt", new BsonInt32(10)));

        // 定义更新操作：递增counter字段，并将新消息推入messages数组
        BsonDocument updateStatement = new BsonDocument();
        updateStatement.append("$inc", new BsonDocument("counter", new BsonInt32(1)));
        updateStatement.append("$push", new BsonDocument("messages", new BsonDocument("message", message)));

        // 返回一个UpdateOneModel，执行单次更新操作
        return new UpdateOneModel<BsonDocument>(
                filters,
                updateStatement,
                UPDATE_OPTIONS
        );
    }
}

登录后复制

这个策略会查找counter字段小于10的文档，将其counter递增1，并将传入的message添加到messages数组中。如果找不到匹配文档，则根据upsert: true创建新文档。

2. Kafka Connect插件管理与部署挑战

将自定义的WriteModelStrategy集成到Kafka Connect环境中，需要正确地打包、部署并配置。通常，这涉及将自定义Java类编译成JAR包，然后将其放置在Kafka Connect的plugin.path所指定的目录中。

2.1 常见的部署方式（问题根源）

在Docker环境中，一个常见的部署模式是构建一个包含自定义JAR包的镜像。以下是一个可能导致问题的Dockerfile片段，它尝试将各种JAR包复制到不同的目录，并手动构建plugin.path：

# ... (构建自定义JAR的部分) ...

FROM confluentinc/cp-kafka-connect:7.2.2

# ... (用户和目录创建) ...

USER root

RUN confluent-hub install --no-prompt confluentinc/kafka-connect-avro-converter:5.5.3 && \
    confluent-hub install --no-prompt mongodb/kafka-connect-mongodb:1.8.0 && \
    mkdir /usr/share/confluent-hub-components/plugins && \
    mkdir /usr/share/confluent-hub-components/mongo_plugins && \
    cp /usr/share/confluent-hub-components/mongodb-kafka-connect-mongodb/lib/*.jar /usr/share/confluent-hub-components/mongo_plugins && \
    cp /usr/share/confluent-hub-components/confluentinc-kafka-connect-avro-converter/lib/*.jar /usr/share/confluent-hub-components/plugins && \
    cp /usr/share/filestream-connectors/*.jar /usr/share/confluent-hub-components/plugins

USER appuser

ENV ARTIFACT_ID=CustomPlugins-1.0-SNAPSHOT.jar
COPY --from=build /app/resources/custom_plugins/target/$ARTIFACT_ID /usr/share/confluent-hub-components/mongo_plugins/$ARTIFACT_ID

# ... (其他配置) ...

登录后复制

随后，Kafka Connect的配置文件中plugin.path可能被设置为：

plugin.path=/usr/share/confluent-hub-components/plugins,/usr/share/confluent-hub-components/mongo_plugins/CustomPlugins-1.0-SNAPSHOT.jar,/usr/share/confluent-hub-components/mongo_plugins/mongo-kafka-connect-1.8.0-confluent.jar,

登录后复制

并在连接器配置中指定自定义策略：

writemodel.strategy=com.fu.connect.sink.CustomWriteModelStrategy

登录后复制

2.2 “类未找到”错误分析

尽管自定义的JAR包似乎被放置在plugin.path中的某个位置，并且自定义的Kafka Connect SMTs（Single Message Transforms）可能正常工作，但尝试启动连接器时，仍会遇到以下错误：

java.util.concurrent.ExecutionException: org.apache.kafka.connect.runtime.rest.errors.BadRequestException: Connector configuration is invalid and contains the following 1 error(s):
Invalid value com.fu.connect.sink.CustomWriteModelStrategy for configuration writemodel.strategy: Class not found: com.fu.connect.sink.CustomWriteModelStrategy

登录后复制

这个错误表明Kafka Connect的类加载器无法找到com.fu.connect.sink.CustomWriteModelStrategy这个类。虽然自定义SMTs可以正常加载，但自定义WriteModelStrategy的加载机制可能有所不同，或者对JAR包的物理位置和结构有更严格的要求。

根本原因在于Kafka Connect的插件加载器对JAR文件的位置和结构非常“挑剔”，特别是对于像WriteModelStrategy这类由连接器内部动态加载的组件。当使用confluent-hub install安装连接器时，它会将所有必要的JAR文件放置在一个特定的目录结构中（例如/usr/share/confluent-hub-components/mongodb-kafka-connect-mongodb/lib/）。手动将这些JAR文件从其原始安装位置复制到其他目录，并试图通过plugin.path直接指向这些散落的JAR文件，会破坏Kafka Connect预期的类加载路径和依赖解析。

集简云

软件集成平台，快速建立企业自动化与智能化

查看详情

3. 解决“类未找到”错误：正确的插件管理实践

解决这个问题的关键在于尊重Kafka Connect的插件加载机制，并避免不必要的JAR文件复制。

3.1 简化JAR管理与plugin.path配置

最有效的解决方案是：

避免手动复制confluent-hub install安装的JAR文件。 让Confluent Hub插件管理器负责管理其自身的JAR文件。
将自定义JAR包放置在一个独立的、清晰的目录中。 这个目录应该只包含你的自定义代码及其必要的依赖，而不是Confluent Hub安装的连接器JAR。
在plugin.path中明确指出所有插件的根目录。 Kafka Connect会递归扫描这些根目录下的所有JAR文件。

根据上述原则，对Dockerfile进行以下修改：

# ... (构建自定义JAR的部分，保持不变) ...

FROM confluentinc/cp-kafka-connect:7.2.2

# ... (用户和目录创建) ...

USER root

# 1. 正常安装Confluent Hub连接器，不手动复制其JAR文件
RUN confluent-hub install --no-prompt confluentinc/kafka-connect-avro-converter:5.5.3 && \
    confluent-hub install --no-prompt mongodb/kafka-connect-mongodb:1.8.0

# 2. 为自定义插件创建一个专用的目录
RUN mkdir -p /usr/share/confluent-hub-components/custom_plugins

USER appuser

ENV ARTIFACT_ID=CustomPlugins-1.0-SNAPSHOT.jar
# 3. 将自定义JAR复制到这个专用目录
COPY --from=build /app/resources/custom_plugins/target/$ARTIFACT_ID /usr/share/confluent-hub-components/custom_plugins/$ARTIFACT_ID

# ... (其他配置) ...

登录后复制

然后，更新Kafka Connect的配置文件中的plugin.path，使其指向Confluent Hub安装的插件目录和你的自定义插件目录：

# plugin.path指向Confluent Hub安装插件的根目录，以及你自定义插件的根目录
plugin.path=/usr/share/confluent-hub-components,/usr/share/confluent-hub-components/custom_plugins

登录后复制

在这种配置下：

/usr/share/confluent-hub-components将包含所有由confluent-hub install安装的连接器，每个连接器都有自己的子目录（例如mongodb-kafka-connect-mongodb/lib）。Kafka Connect的类加载器能够正确地找到这些连接器及其依赖。
/usr/share/confluent-hub-components/custom_plugins将包含你的自定义CustomPlugins-1.0-SNAPSHOT.jar。

3.2 连接器配置

连接器配置保持不变，继续指定自定义写入策略的完整类名：

name=my-mongodb-sink-connector
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
tasks.max=1
topics=your-topic
connection.uri=mongodb://localhost:27017
database=mydatabase
collection=mycollection
writemodel.strategy=com.fu.connect.sink.CustomWriteModelStrategy 
# ... 其他配置 ...

登录后复制

通过以上调整，Kafka Connect的类加载器将能够正确地解析com.fu.connect.sink.CustomWriteModelStrategy类，从而成功加载并执行自定义写入策略。

4. 总结与注意事项

在Kafka Connect中集成自定义组件（如WriteModelStrategy、SMTs或自定义Converter）时，对plugin.path的理解和JAR包的管理至关重要。

保持插件结构清晰： 尽量让每个插件（包括自定义插件）都有一个独立的根目录，并在plugin.path中列出这些根目录。
避免手动移动已安装的JAR： confluent-hub install会建立一个Kafka Connect期望的插件目录结构。手动复制这些JAR文件会破坏这种结构，导致类加载问题。
依赖管理： 确保自定义JAR包包含了所有必要的运行时依赖，或者这些依赖也通过plugin.path可被Kafka Connect找到。
排查ClassPath问题： 当遇到ClassNotFoundException时，首先检查plugin.path配置是否正确，然后确认自定义JAR包是否确实存在于指定路径下，并且其内部结构（包名、类名）与配置中指定的完全一致。

遵循这些最佳实践，可以有效避免在Kafka Connect环境中集成自定义逻辑时遇到的常见类加载问题，确保连接器能够稳定、高效地运行。

以上就是Kafka Connect MongoDB Sink自定义写入策略集成指南的详细内容，更多请关注php中文网其它相关文章！