@VisibleForTesting
    static CompletedCheckpointStore createCompletedCheckpointStore(
            Configuration jobManagerConfig,
            CheckpointRecoveryFactory recoveryFactory,
            Executor ioExecutor,
            Logger log,
            JobID jobId,
            RestoreMode restoreMode)
            throws Exception {
        return recoveryFactory.createRecoveredCompletedCheckpointStore(
                jobId,
                DefaultCompletedCheckpointStoreUtils.getMaximumNumberOfRetainedCheckpoints(
                        jobManagerConfig, log),
                SharedStateRegistry.DEFAULT_FACTORY,
                ioExecutor,
                restoreMode);
    }
@Test
    void testSharedStateRegistration() throws Exception {
        UUID backendId = UUID.randomUUID();
        String localPath = "k0";
        StreamStateHandle handle = new ByteStreamStateHandle("h0", new byte[] {1, 2, 3});
        CheckpointRecoveryFactory recoveryFactory =
                buildRecoveryFactory(
                        buildCheckpoint(buildIncrementalHandle(localPath, handle, backendId)));

        CompletedCheckpointStore checkpointStore =
                SchedulerUtils.createCompletedCheckpointStore(
                        new Configuration(),
                        recoveryFactory,
                        Executors.directExecutor(),
                        log,
                        new JobID(),
                        RestoreMode.CLAIM);

        SharedStateRegistry sharedStateRegistry = checkpointStore.getSharedStateRegistry();

        IncrementalRemoteKeyedStateHandle newHandle =
                buildIncrementalHandle(
                        localPath,
                        new PlaceholderStreamStateHandle(
                                handle.getStreamStateHandleID(), handle.getStateSize(), false),
                        backendId);
        newHandle.registerSharedStates(sharedStateRegistry, 1L);

        assertThat(
                        newHandle.getSharedState().stream()
                                .filter(e -> e.getLocalPath().equals(localPath))
                                .findFirst()
                                .get()
                                .getHandle())
                .isEqualTo(handle);
    }