@Override
    public void configure(WorkerConfig config) {
        // eagerly initialize the executor, because OffsetStorageWriter will use it later
        start();

        Map<String, ?> conf = config.originals();
        if (!conf.containsKey(OFFSET_STATE_VALUE)) {
            // a normal startup from clean state, not need to initialize the offset
            return;
        }

        String stateJson = (String) conf.get(OFFSET_STATE_VALUE);
        DebeziumOffsetSerializer serializer = new DebeziumOffsetSerializer();
        DebeziumOffset debeziumOffset;
        try {
            debeziumOffset = serializer.deserialize(stateJson.getBytes(StandardCharsets.UTF_8));
        } catch (IOException e) {
            LOG.error("Can't deserialize debezium offset state from JSON: " + stateJson, e);
            throw new RuntimeException(e);
        }

        final String engineName = (String) conf.get(EmbeddedEngine.ENGINE_NAME.name());
        Converter keyConverter = new JsonConverter();
        Converter valueConverter = new JsonConverter();
        keyConverter.configure(config.originals(), true);
        Map<String, Object> valueConfigs = new HashMap<>(conf);
        valueConfigs.put("schemas.enable", false);
        valueConverter.configure(valueConfigs, true);
        OffsetStorageWriter offsetWriter =
                new OffsetStorageWriter(
                        this,
                        // must use engineName as namespace to align with Debezium Engine
                        // implementation
                        engineName,
                        keyConverter,
                        valueConverter);

        offsetWriter.offset(debeziumOffset.sourcePartition, debeziumOffset.sourceOffset);

        // flush immediately
        if (!offsetWriter.beginFlush()) {
            // if nothing is needed to be flushed, there must be something wrong with the
            // initialization
            LOG.warn(
                    "Initialize FlinkOffsetBackingStore from empty offset state, this shouldn't happen.");
            return;
        }

        // trigger flushing
        Future<Void> flushFuture =
                offsetWriter.doFlush(
                        (error, result) -> {
                            if (error != null) {
                                LOG.error("Failed to flush initial offset.", error);
                            } else {
                                LOG.debug("Successfully flush initial offset.");
                            }
                        });

        // wait until flushing finished
        try {
            flushFuture.get(FLUSH_TIMEOUT_SECONDS, TimeUnit.SECONDS);
            LOG.info(
                    "Flush offsets successfully, partition: {}, offsets: {}",
                    debeziumOffset.sourcePartition,
                    debeziumOffset.sourceOffset);
        } catch (InterruptedException e) {
            LOG.warn("Flush offsets interrupted, cancelling.", e);
            offsetWriter.cancelFlush();
        } catch (ExecutionException e) {
            LOG.error("Flush offsets threw an unexpected exception.", e);
            offsetWriter.cancelFlush();
        } catch (TimeoutException e) {
            LOG.error("Timed out waiting to flush offsets to storage.", e);
            offsetWriter.cancelFlush();
        }
    }
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/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/debezium/internal/FlinkOffsetBackingStore.java