@Override
    public void start() {
        if (executor == null) {
            executor =
                    Executors.newFixedThreadPool(
                            1,
                            ThreadUtils.createThreadFactory(
                                    this.getClass().getSimpleName() + "-%d", false));
        }
    }
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