Hi folks, I am trying to use this API to read my checkpoints that should only have kafka consumer offsets in them. My checkpoint directories only have

_metadata

files, and I suppose that's because the state is really small? Does anyone have any pointers to how I can inspect the checkpointed offsets for

KafkaSource

operator? Thank you!

Hi, I did somethign liek that to read State of Delta Source connector, so instead using Delta Deserializers you need to use Kafka connectors one. Maybe this will help you. Please see commennts in code

package org.example.streamprocessor;

import io.delta.flink.source.internal.state.DeltaEnumeratorStateCheckpoint;
import io.delta.flink.source.internal.state.DeltaPendingSplitsCheckpointSerializer;
import io.delta.flink.source.internal.state.DeltaSourceSplit;
import io.delta.flink.source.internal.state.DeltaSourceSplitSerializer;
import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.File;
import java.io.IOException;
import java.util.Comparator;
import org.apache.flink.core.memory.DataInputViewStreamWrapper;
import org.apache.flink.runtime.checkpoint.OperatorState;
import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata;
import org.apache.flink.runtime.source.coordinator.SourceCoordinatorSerdeUtils;
import org.apache.flink.state.api.runtime.SavepointLoader;
import org.apache.flink.state.api.runtime.metadata.SavepointMetadataV2;
import org.junit.jupiter.api.Test;

public class CheckpointReadTest {

    @Test
    public void testDeserializeEnumState() throws IOException {
        File resourcesDirectory = new File("src/test/resources/e920a0830ddb1308aa5f2b03b94a3a72/chk-165");

        CheckpointMetadata metadata = SavepointLoader.loadSavepointMetadata( resourcesDirectory.getAbsolutePath());

        int maxParallelism =
            metadata.getOperatorStates().stream()
                .map(OperatorState::getMaxParallelism)
                .max(Comparator.naturalOrder())
                .orElseThrow(
                    () ->
                        new RuntimeException(
                            "Savepoint must contain at least one operator state."));

        SavepointMetadataV2 savepointMetadata =
            new SavepointMetadataV2(
                maxParallelism, metadata.getMasterStates(), metadata.getOperatorStates());

        // change uid here
        byte[] data =
            savepointMetadata.getOperatorState("myStreamingDeltaSourceUid").getCoordinatorState()
                .getData();


        try (ByteArrayInputStream bais = new ByteArrayInputStream(data);
            DataInputStream in = new DataInputViewStreamWrapper(bais)) {
            final int coordinatorSerdeVersion = readAndVerifyCoordinatorSerdeVersion(in);
            int enumSerializerVersion = in.readInt();
            int serializedEnumChkptSize = in.readInt();
            byte[] serializedEnumChkpt = readBytes(in, serializedEnumChkptSize);

            if (coordinatorSerdeVersion != SourceCoordinatorSerdeUtils.VERSION_0
                && bais.available() > 0) {
                throw new IOException("Unexpected trailing bytes in enumerator checkpoint data");
            }

            DeltaPendingSplitsCheckpointSerializer<DeltaSourceSplit> des = new DeltaPendingSplitsCheckpointSerializer<>(
                DeltaSourceSplitSerializer.INSTANCE);

            DeltaEnumeratorStateCheckpoint<DeltaSourceSplit> deserializeEnumeratorState =
                des.deserialize(enumSerializerVersion, serializedEnumChkpt);

            System.out.println("Delta Snapshot Version: " + deserializeEnumeratorState.getSnapshotVersion());

            // Splits that were not yet assigned to readers.
            for (DeltaSourceSplit split : deserializeEnumeratorState.getSplits()) {
                System.out.println(split.path());
            }
        }
    }

    static byte[] readBytes(DataInputStream in, int size) throws IOException {
        byte[] bytes = new byte[size];
        in.readFully(bytes);
        return bytes;
    }

    static int readAndVerifyCoordinatorSerdeVersion(DataInputStream in) throws IOException {
        int version = in.readInt();
        if (version > 1) {
            throw new IOException("Unsupported source coordinator serde version " + version);
        }
        return version;
    }

}