@Override
    public void stop() {
        if (executor != null) {
            executor.shutdown();
            // Best effort wait for any get() and set() tasks (and caller's callbacks) to complete.
            try {
                executor.awaitTermination(30, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            if (!executor.shutdownNow().isEmpty()) {
                throw new ConnectException(
                        "Failed to stop FlinkOffsetBackingStore. Exiting without cleanly "
                                + "shutting down pending tasks and/or callbacks.");
            }
            executor = null;
        }
    }
A implementation of {@link OffsetBackingStore} backed on Flink's state mechanism.

<p>The {@link #OFFSET_STATE_VALUE} in the {@link WorkerConfig} is the raw position and offset
data in JSON format. It is set into the config when recovery from failover by {@link
DebeziumSourceFunction} before startup the {@link DebeziumEngine}. If it is not a restoration,
the {@link #OFFSET_STATE_VALUE} is empty. {@link DebeziumEngine} relies on the {@link
OffsetBackingStore} for failover recovery.</p>

@see DebeziumSourceFunction
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/FlinkOffsetBackingStore.java