@Override
    public Future<Map<ByteBuffer, ByteBuffer>> get(final Collection<ByteBuffer> keys) {
        return executor.submit(
                () -> {
                    Map<ByteBuffer, ByteBuffer> result = new HashMap<>();
                    for (ByteBuffer key : keys) {
                        result.put(key, data.get(key));
                    }
                    return result;
                });
    }
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