提示:一个嵌入式的Kafka Connect源连接器的工作机制
文章目录
- 前言
- 一、控制流图
- 二、代码分析
- 1.构造函数
- 2.完成回调
- 3.连接器回调
- 4.RUN
- 总结
前言
工作机制:
* 独立运行:嵌入式连接器在应用程序进程中独立运行,不需要Kafka、Kafka Connect或 Zookeeper进程
* 数据传递:应用程序设置连接器并提供一个消费者函数,连接器将所有包含数据库变更事件的SourceRecord传递给该函数。
* 责任转移:应用程序负责故障恢复、可扩展性和持久性。
* 默认存储:连接器的数据库模式历史和偏移量默认存储在内存中,重启后会丢失。
* 执行与停止:连接器设计为提交给Executor或ExecutorService单线程执行,可以通过调用stop()方法或中断线程来停止。
提示:以下是本篇文章正文内容
一、控制流图
二、代码分析
1.构造函数
private EmbeddedEngine(Configuration config, ClassLoader classLoader, Clock clock, Consumer<SourceRecord> consumer,CompletionCallback completionCallback, ConnectorCallback connectorCallback) {this.config = config;this.consumer = consumer;this.classLoader = classLoader;this.clock = clock;this.completionCallback = completionCallback != null ? completionCallback : (success, msg, error) -> {if (!success) logger.error(msg, error);};this.connectorCallback = connectorCallback;this.completionResult = new CompletionResult();assert this.config != null;assert this.consumer != null;assert this.classLoader != null;assert this.clock != null;keyConverter = config.getInstance(INTERNAL_KEY_CONVERTER_CLASS, Converter.class, () -> this.classLoader);keyConverter.configure(config.subset(INTERNAL_KEY_CONVERTER_CLASS.name() + ".", true).asMap(), true);valueConverter = config.getInstance(INTERNAL_VALUE_CONVERTER_CLASS, Converter.class, () -> this.classLoader);Configuration valueConverterConfig = config;if (valueConverter instanceof JsonConverter) {// Make sure that the JSON converter is configured to NOT enable schemas ...valueConverterConfig = config.edit().with(INTERNAL_VALUE_CONVERTER_CLASS + ".schemas.enable", false).build();}valueConverter.configure(valueConverterConfig.subset(INTERNAL_VALUE_CONVERTER_CLASS.name() + ".", true).asMap(), false);// Create the worker config, adding extra fields that are required for validation of a worker config// but that are not used within the embedded engine (since the source records are never serialized) ...Map<String, String> embeddedConfig = config.asMap(ALL_FIELDS);embeddedConfig.put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, JsonConverter.class.getName());embeddedConfig.put(WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, JsonConverter.class.getName());workerConfig = new EmbeddedConfig(embeddedConfig);}
2.完成回调
/*** A callback function to be notified when the connector completes.*/
public interface CompletionCallback {/*** Handle the completion of the embedded connector engine.* * @param success {@code true} if the connector completed normally, or {@code false} if the connector produced an error* that prevented startup or premature termination.* @param message the completion message; never null* @param error the error, or null if there was no exception*/void handle(boolean success, String message, Throwable error);
}
这段代码定义了一个接口 CompletionCallback,其中包含一个方法 handle。该方法用于处理嵌入式连接器引擎的完成状态。
参数:
success: 布尔值,表示连接器是否正常完成。如果为 true,表示连接器正常完成;如果为 false,表示连接器启动失败或提前终止。
message: 完成消息,不能为空。
error: 异常对象,如果没有异常则为 null。
3.连接器回调
/*** Callback function which informs users about the various stages a connector goes through during startup*/public interface ConnectorCallback {/*** Called after a connector has been successfully started by the engine; i.e. {@link SourceConnector#start(Map)} has* completed successfully*/default void connectorStarted() {// nothing by default}/*** Called after a connector has been successfully stopped by the engine; i.e. {@link SourceConnector#stop()} has* completed successfully*/default void connectorStopped() {// nothing by default}/*** Called after a connector task has been successfully started by the engine; i.e. {@link SourceTask#start(Map)} has* completed successfully*/default void taskStarted() {// nothing by default}/*** Called after a connector task has been successfully stopped by the engine; i.e. {@link SourceTask#stop()} has* completed successfully*/default void taskStopped() {// nothing by default}}
这段代码定义了一个接口 ConnectorCallback,其中包含了四个默认方法:connectorStarted、connectorStopped、taskStarted 和 taskStopped。这些方法用于在连接器和任务的不同生命周期阶段通知用户。
connectorStarted:当连接器成功启动时调用。
connectorStopped:当连接器成功停止时调用。
taskStarted:当连接器任务成功启动时调用。
taskStopped:当连接器任务成功停止时调用。
这些方法的默认实现为空,子类可以根据需要重写这些方法来添加自定义的回调逻辑。
4 RUN方法核心流程
核心流程参考上图
注:debezium-0.6版本
总结
EmbeddedEngine方法和成员变量