如何在Spring Boot应用关闭时优雅关闭Debezium Connector？

DebeziumEngine本身没有暴露直接的stop()方法，但它实现了AutoCloseable接口，可通过close()方法触发优雅关闭。结合Spring Boot的生命周期回调，有两种可靠实现方式：

方式一：使用@PreDestroy注解 #

在你的Listener组件中添加销毁回调方法，调用DebeziumEngine的close()并处理线程池关闭：

@Component
public class Listener {
    private final DebeziumEngine<RecordChangeEvent<SourceRecord>> debeziumEngine;
    private final ExecutorService executorService;

    public Listener(Configuration eventsConnectorConfig) {
        this.debeziumEngine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
                .using(eventsConnectorConfig.asProperties())
                .notifying(this::handleChangeEvent)
                .build();
        // 初始化单线程池运行DebeziumEngine，避免阻塞Spring启动
        this.executorService = Executors.newSingleThreadExecutor();
        executorService.execute(debeziumEngine);
    }

    private void handleChangeEvent(RecordChangeEvent<SourceRecord> event) {
        // 你的事件处理逻辑
    }

    @PreDestroy
    public void shutdownDebezium() {
        try {
            // 触发Debezium优雅关闭：提交偏移量、停止监听、释放资源
            debeziumEngine.close();
            // 关闭线程池，等待任务终止
            executorService.shutdown();
            if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
            }
        } catch (IOException | InterruptedException e) {
            Thread.currentThread().interrupt();
            // 此处可添加日志记录异常信息
        }
    }
}

方式二：实现SmartLifecycle接口 #

如果需要更精细的生命周期控制，让Listener实现Spring的SmartLifecycle接口，在stop()方法中处理关闭逻辑：

@Component
public class Listener implements SmartLifecycle {
    private final DebeziumEngine<RecordChangeEvent<SourceRecord>> debeziumEngine;
    private final ExecutorService executorService;
    private boolean running = false;

    public Listener(Configuration eventsConnectorConfig) {
        this.debeziumEngine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
                .using(eventsConnectorConfig.asProperties())
                .notifying(this::handleChangeEvent)
                .build();
        this.executorService = Executors.newSingleThreadExecutor();
    }

    private void handleChangeEvent(RecordChangeEvent<SourceRecord> event) {
        // 你的事件处理逻辑
    }

    @Override
    public void start() {
        executorService.execute(debeziumEngine);
        running = true;
    }

    @Override
    public void stop() {
        try {
            debeziumEngine.close();
            executorService.shutdown();
            if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
            }
            running = false;
        } catch (IOException | InterruptedException e) {
            Thread.currentThread().interrupt();
            // 此处可添加日志记录异常信息
        }
    }

    @Override
    public boolean isRunning() {
        return running;
    }
}

核心注意点 #

• 必须将DebeziumEngine放入独立线程池运行，否则会阻塞Spring应用的启动流程。
• 调用close()后，Debezium会自动完成偏移量提交、数据源监听停止、资源释放等优雅关闭步骤。
• 线程池的终止等待需设置合理超时，避免应用退出被无限阻塞。

内容的提问来源于stack exchange，提问作者user674669