StreamOperatorStateContext build(Logger logger) throws IOException {
        final Environment environment =
                new SavepointEnvironment.Builder(ctx, executionConfig, maxParallelism)
                        .setConfiguration(configuration)
                        .setSubtaskIndex(split.getSplitNumber())
                        .setPrioritizedOperatorSubtaskState(
                                split.getPrioritizedOperatorSubtaskState())
                        .build();

        StateBackend stateBackend;
        try {
            Configuration jobConfig = environment.getJobConfiguration();
            jobConfig.set(StateChangelogOptions.ENABLE_STATE_CHANGE_LOG, false);
            Configuration clusterConfig = new Configuration(configuration);
            stateBackend =
                    StateBackendLoader.fromApplicationOrConfigOrDefault(
                            applicationStateBackend,
                            jobConfig,
                            clusterConfig,
                            ctx.getUserCodeClassLoader(),
                            logger);
        } catch (DynamicCodeLoadingException e) {
            throw new IOException("Failed to load state backend", e);
        }

        StreamTaskStateInitializer initializer =
                new StreamTaskStateInitializerImpl(environment, stateBackend);

        try {
            return initializer.streamOperatorStateContext(
                    operatorState.getOperatorID(),
                    operatorState.getOperatorID().toString(),
                    new NeverFireProcessingTimeService(),
                    keyContext,
                    keySerializer,
                    registry,
                    ctx.getMetricGroup(),
                    1.0,
                    false);
        } catch (Exception e) {
            throw new IOException("Failed to restore state backend", e);
        }
    }
@Test(expected = CustomStateBackendFactory.ExpectedException.class)
    public void testStateBackendLoading() throws Exception {
        Configuration configuration = new Configuration();
        configuration.set(
                StateBackendOptions.STATE_BACKEND,
                CustomStateBackendFactory.class.getCanonicalName());

        MockStreamingRuntimeContext context = new MockStreamingRuntimeContext(true, 1, 0);
        StreamOperatorContextBuilder builder =
                new StreamOperatorContextBuilder(
                        context,
                        configuration,
                        new OperatorState(new OperatorID(), 1, 128),
                        new PrioritizedOperatorSubtaskStateInputSplit() {
                            @Override
                            public PrioritizedOperatorSubtaskState
                                    getPrioritizedOperatorSubtaskState() {
                                return PrioritizedOperatorSubtaskState.emptyNotRestored();
                            }

                            @Override
                            public int getSplitNumber() {
                                return 0;
                            }
                        },
                        new CloseableRegistry(),
                        null,
                        context.getExecutionConfig());

        builder.build(LOG);
    }