当前位置：首页 > article >正文

Debezium-EmbeddedEngine

article 2024/11/19 23:49:36

提示：一个嵌入式的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：当连接器任务成功停止时调用。
这些方法的默认实现为空，子类可以根据需要重写这些方法来添加自定义的回调逻辑。