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环境中集成自定义逻辑时遇到的常见类加载问题，确保连接器能够稳定、高效地运行。
