@Override
    public Optional<MaterializationRunnable> initMaterialization() throws Exception {
        if (lastConfirmedMaterializationId < materializedId - 1
                && lastFailedMaterializationId < materializedId - 1) {
            // SharedStateRegistry potentially requires that the checkpoint's dependency on the
            // shared file be continuous, it will be broken if we trigger a new materialization
            // before the previous one has either confirmed or failed. See discussion in
            // https://github.com/apache/flink/pull/22669#issuecomment-1593370772 .
            LOG.info(
                    "materialization:{} not confirmed or failed or cancelled, skip trigger new one.",
                    materializedId - 1);
            return Optional.empty();
        }

        SequenceNumber upTo = stateChangelogWriter.nextSequenceNumber();
        SequenceNumber lastMaterializedTo = changelogSnapshotState.lastMaterializedTo();

        LOG.info(
                "Initialize Materialization. Current changelog writers last append to sequence number {}",
                upTo);

        if (upTo.compareTo(lastMaterializedTo) > 0) {

            LOG.info("Starting materialization from {} : {}", lastMaterializedTo, upTo);

            // This ID is not needed for materialization; But since we are re-using the
            // streamFactory that is designed for state backend snapshot, which requires unique
            // checkpoint ID. A faked materialized Id is provided here.
            long materializationID = materializedId++;

            MaterializationRunnable materializationRunnable =
                    new MaterializationRunnable(
                            keyedStateBackend.snapshot(
                                    materializationID,
                                    System.currentTimeMillis(),
                                    // TODO: implement its own streamFactory.
                                    streamFactory,
                                    CHECKPOINT_OPTIONS),
                            materializationID,
                            upTo);

            // log metadata after materialization is triggered
            changelogStateFactory.resetAllWritingMetaFlags();

            return Optional.of(materializationRunnable);
        } else {
            LOG.debug(
                    "Skip materialization, last materialized to {} : last log to {}",
                    lastMaterializedTo,
                    upTo);

            return Optional.empty();
        }
    }
@Test
    public void testInitMaterialization() throws Exception {
        MockKeyedStateBackend<Integer> delegatedBackend = createMock();
        ChangelogKeyedStateBackend<Integer> backend = createChangelog(delegatedBackend);

        try {
            Optional<MaterializationRunnable> runnable;

            appendMockStateChange(backend); // ensure there is non-materialized changelog

            runnable = backend.initMaterialization();
            // 1. should trigger first materialization
            assertTrue("first materialization should be trigger.", runnable.isPresent());

            appendMockStateChange(backend); // ensure there is non-materialized changelog

            // 2. should not trigger new one until the previous one has been confirmed or failed
            assertFalse(backend.initMaterialization().isPresent());

            backend.handleMaterializationFailureOrCancellation(
                    runnable.get().getMaterializationID(),
                    runnable.get().getMaterializedTo(),
                    null);
            runnable = backend.initMaterialization();
            // 3. should trigger new one after previous one failed
            assertTrue(runnable.isPresent());

            appendMockStateChange(backend); // ensure there is non-materialized changelog

            // 4. should not trigger new one until the previous one has been confirmed or failed
            assertFalse(backend.initMaterialization().isPresent());

            backend.handleMaterializationResult(
                    SnapshotResult.empty(),
                    runnable.get().getMaterializationID(),
                    runnable.get().getMaterializedTo());
            checkpoint(backend, checkpointId).get().discardState();
            backend.notifyCheckpointComplete(checkpointId);
            // 5. should trigger new one after previous one has been confirmed
            assertTrue(backend.initMaterialization().isPresent());
        } finally {
            backend.close();
            backend.dispose();
        }
    }