public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
        String basePath = "<some path>";
        String savepointPath = basePath;

        SavepointReader sr = SavepointReader.read(see, savepointPath + "/<savepointpath>");

        // Create Field object
        Field privateField
                = SavepointReader.class.getDeclaredField("metadata");
        privateField.setAccessible(true);
        SavepointMetadataV2 metadata = (SavepointMetadataV2) privateField.get(sr);

        for (OperatorState state : metadata.getExistingOperators()) {
            if (state.getCoordinatorState() != null) {
                saveState(metadata, OperatorIdentifier.forUidHash(state.getOperatorID().toString()), basePath);
            }
        }

    }
    public static void saveState(SavepointMetadataV2 metadata, OperatorIdentifier operatorIdentifier, String basePath) throws Exception {
        String uid = operatorIdentifier.toString();

        DynamoDbStreamsShardSplitSerializer splitSerializer = new DynamoDbStreamsShardSplitSerializer();
        DynamoDbStreamsSourceEnumeratorStateSerializer enumeratorStateSerializer = new DynamoDbStreamsSourceEnumeratorStateSerializer(splitSerializer);

        OperatorState state = metadata.getOperatorState(operatorIdentifier);
        byte[] coordinatorState = state.getCoordinatorState().getData();
        byte[] enumeratorBytes = Arrays.copyOfRange(coordinatorState, 12, coordinatorState.length);
        Map<String, DynamoDBStreamsShardSplitWithAssignmentStatus> knownSplits = new HashMap<>();
        Map<String, DynamoDBStreamsShardSplitWithAssignmentStatus> allKnownSplits = new HashMap<>();
        FileOutputStream out_known = new FileOutputStream(basePath + "/known_splits_" + uid + ".txt");

        DynamoDbStreamsSourceEnumeratorState enumerator = enumeratorStateSerializer.deserialize(0, enumeratorBytes);
        for (DynamoDBStreamsShardSplitWithAssignmentStatus s : enumerator.getKnownSplits()) {
            if(s.assignmentStatus() == SplitAssignmentStatus.ASSIGNED) {
                knownSplits.put(s.split().getShardId(), s);
            }
            allKnownSplits.put(s.split().getShardId(), s);
            out_known.write((s + "\n").getBytes());
        }
        out_known.close();


        Map<String, DynamoDbStreamsShardSplit> assignedSplits = new HashMap<>();
        FileOutputStream out = new FileOutputStream(basePath + "/splits_" + uid + ".txt");
        for (OperatorSubtaskState os : state.getSubtaskStates().values()) {
            for (OperatorStateHandle osh : os.getManagedOperatorState()) {
                byte[] stateBytes;
                stateBytes = osh.asBytesIfInMemory().orElseGet(() -> {
                    Path path = Paths.get(basePath, ((RelativeFileStateHandle) osh.getDelegateStateHandle()).getRelativePath());
                    try {
                        return Files.readAllBytes(path);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                });
//                byte[] stateBytes = osh.asBytesIfInMemory().get();
                for (OperatorStateHandle.StateMetaInfo smi : osh.getStateNameToPartitionOffsets().values()) {
                    long[] offsets = smi.getOffsets();
                    for (int i = 0; i < offsets.length; i++) {
                        int start = (int) offsets[i] + 12;
                        int end = 0;
                        if (i < offsets.length - 1) {
                            end = (int) offsets[i + 1];
                        } else {
                            end = stateBytes.length;
                        }
                        byte[] bytes = Arrays.copyOfRange(stateBytes, start, end);
                        DynamoDbStreamsShardSplit split = splitSerializer.deserialize(0, bytes);
                        assignedSplits.put(split.getShardId(), split);
                        if (split.isFinished()) {
                            System.out.println("Finished split: " + split + " uid: " + uid);
                        }
                        out.write((split + "\n").getBytes());
                    }
                }
            }
        }
        out.flush();
        out.close();

        knownSplits.keySet().removeAll(assignedSplits.keySet());
        System.out.println("Missing splits in " + operatorIdentifier + ": " + knownSplits.keySet());
    }