Reading Flink Kafka Consumer Offsets from a Savepoint

ReadKafkaSourceState.java

package dev.irontools.flink.state;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.state.api.OperatorIdentifier;
import org.apache.flink.state.api.SavepointReader;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;

/**
 * Reads KafkaSource operator state (assigned splits) from a savepoint.
 *
 * Each split is stored in ListState named "SourceReaderState" as byte[] elements
 * wrapped with SimpleVersionedSerialization: [version(4)][length(4)][splitData(N)].
 *
 * The split data follows KafkaPartitionSplitSerializer format:
 * [topic(UTF)][partition(int)][startingOffset(long)][stoppingOffset(long)]
 */
public class ReadKafkaSourceState {

    private static final String SOURCE_READER_STATE_NAME = "SourceReaderState";
    private static final long NO_STOPPING_OFFSET = Long.MIN_VALUE;

    public static class KafkaPartitionSplitState {
        public String topic;
        public int partition;
        public long startingOffset;
        public long stoppingOffset;

        public KafkaPartitionSplitState() {}

        public KafkaPartitionSplitState(String topic, int partition, long startingOffset, long stoppingOffset) {
            this.topic = topic;
            this.partition = partition;
            this.startingOffset = startingOffset;
            this.stoppingOffset = stoppingOffset;
        }

        @Override
        public String toString() {
            String stoppingOffsetStr = stoppingOffset == NO_STOPPING_OFFSET ? "NONE" : String.valueOf(stoppingOffset);
            return "KafkaPartitionSplitState{topic='" + topic
                + "', partition=" + partition
                + ", startingOffset=" + startingOffset
                + ", stoppingOffset=" + stoppingOffsetStr + "}";
        }
    }

    static KafkaPartitionSplitState deserializeSplit(byte[] raw) throws IOException {
        try (DataInputStream in = new DataInputStream(new ByteArrayInputStream(raw))) {
            // SimpleVersionedSerialization wrapper
            int serializerVersion = in.readInt();
            int dataLength = in.readInt();

            // KafkaPartitionSplitSerializer format (version 0)
            String topic = in.readUTF();
            int partition = in.readInt();
            long startingOffset = in.readLong();
            long stoppingOffset = in.readLong();

            return new KafkaPartitionSplitState(topic, partition, startingOffset, stoppingOffset);
        }
    }

    public static void main(String[] args) throws Exception {
        String savepointPath = "/tmp/flink-savepoints/savepoint-e5a859-482480384cdc";
        // SQL-generated operator UID hash — find via savepoint_metadata() or savepoint files
        String operatorUidHash = "cbc357ccb763df2852fee8c4fc7d55f2";

        Configuration conf = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        SavepointReader savepoint = SavepointReader.read(
            env,
            savepointPath,
            new HashMapStateBackend()
        );

        savepoint.readListState(
                OperatorIdentifier.forUidHash(operatorUidHash),
                SOURCE_READER_STATE_NAME,
                TypeInformation.of(byte[].class)
            )
            .map(ReadKafkaSourceState::deserializeSplit)
            .print();

        env.execute("Read Kafka Source State");
    }
}

ReadSavepoint.java

package dev.irontools.flink.state;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class ReadSavepoint {
  public static void main(String[] args) {
      EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
      settings.getConfiguration().setString("table.display.max-column-width", "1000");
      TableEnvironment tEnv = TableEnvironment.create(settings);

      tEnv.executeSql("LOAD MODULE state");
      tEnv.executeSql("SELECT `operator-uid-hash` FROM savepoint_metadata('/tmp/flink-savepoints/savepoint-e5a859-482480384cdc') WHERE `operator-name` = 'Source: LoadTest[1]'").print();
    }
}

Complete tutorial https://streamacademy.io/tutorial/flink-kafka-consumer-offsets-with-the-state-processor-api/