@Override
    public void destroy() {
        synchronized (this) {
            if (!destroyed) {
                executor.shutdownNow();
                binlogSnapshot.close();
                destroyed = true;
            }
        }
    }
pair.left: table name
pair.right: actual data
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/BinlogReader.java