Repository: apache/flink
File: flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java (Java)

Code before:
        }

        public Configuration getConfiguration(File checkpointDir) {
            Configuration conf = new Configuration();

            conf.set(TaskManagerOptions.NETWORK_MEMORY_FRACTION, 0.9f);
            conf.set(TaskManagerOptions.MEMORY_SEGMENT_SIZE, MemorySize.parse("4kb"));
            conf.set(StateBackendOptions.STATE_BACKEND, "hashmap");
            conf.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, checkpointDir.toURI().toString());
            if (restoreCheckpoint != null) {
                conf.set(StateRecoveryOptions.SAVEPOINT_PATH, restoreCheckpoint);
            }

            conf.set(
                    ShuffleServiceOptions.SHUFFLE_SERVICE_FACTORY_CLASS,
                    SharedPoolNettyShuffleServiceFactory.class.getName());
            conf.set(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX, 60000);
            // half memory consumption of network buffers (default is 64mb), as some tests spawn a
            // large number of task managers (25)
            // 12mb was sufficient to run the tests, so 32mb should put us above the recommended
            // amount of buffers
            conf.set(TaskManagerOptions.NETWORK_MEMORY_MIN, MemorySize.ofMebiBytes(32));
            conf.set(TaskManagerOptions.NETWORK_MEMORY_MAX, MemorySize.ofMebiBytes(32));
            conf.set(RpcOptions.ASK_TIMEOUT_DURATION, Duration.ofMinutes(1));
            return conf;
        }

        @Override
        public String toString() {
            return "UnalignedSettings{"
                    + "parallelism="
                    + parallelism
                    + ", minCheckpoints="
                    + minCheckpoints
                    + ", restoreCheckpoint="
                    + restoreCheckpoint
                    + ", generateCheckpoint="
                    + generateCheckpoint
                    + ", expectedFailures="
                    + expectedFailures
                    + ", dagCreator="
                    + dagCreator
                    + ", alignmentTimeout="
                    + alignmentTimeout
                    + ", failuresAfterSourceFinishes="
                    + failuresAfterSourceFinishes
                    + ", channelType="
                    + channelType
                    + ", sourceSleepMs="
                    + sourceSleepMs
                    + '}';

Review comment (bug):
It adds the wrong prefix, so updated `restoreCheckpoint` to String.

<img width="1308" height="866" alt="Image" src="https://github.com/user-attachments/assets/5c2b5242-9a58-4111-bc1c-bbdd035d9cdc" />

Code after:
                    .setAlignedCheckpointTimeout(Duration.ofMillis(alignmentTimeout));
            env.getCheckpointConfig().setCheckpointTimeout(checkpointTimeout.toMillis());
            env.getCheckpointConfig()
                    .setTolerableCheckpointFailureNumber(tolerableCheckpointFailures);
            env.setParallelism(parallelism);
            RestartStrategyUtils.configureFixedDelayRestartStrategy(
                    env, generateCheckpoint ? expectedFailures / 2 : expectedFailures, 100L);
            env.getCheckpointConfig().enableUnalignedCheckpoints(true);
            // for custom partitioner
            env.getCheckpointConfig().setForceUnalignedCheckpoints(true);
            if (generateCheckpoint) {
                env.getCheckpointConfig()
                        .setExternalizedCheckpointRetention(
                                ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION);
            }
        }

        public Configuration getConfiguration(File checkpointDir) {
            Configuration conf = new Configuration();

            conf.set(TaskManagerOptions.NETWORK_MEMORY_FRACTION, 0.9f);
            conf.set(TaskManagerOptions.MEMORY_SEGMENT_SIZE, MemorySize.parse("4kb"));
            conf.set(StateBackendOptions.STATE_BACKEND, "hashmap");
            conf.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, checkpointDir.toURI().toString());
            if (restoreCheckpoint != null) {
                conf.set(StateRecoveryOptions.SAVEPOINT_PATH, restoreCheckpoint);
            }

            conf.set(
                    ShuffleServiceOptions.SHUFFLE_SERVICE_FACTORY_CLASS,
                    SharedPoolNettyShuffleServiceFactory.class.getName());
            conf.set(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX, 60000);
            // half memory consumption of network buffers (default is 64mb), as some tests spawn a
            // large number of task managers (25)
            // 12mb was sufficient to run the tests, so 32mb should put us above the recommended
            // amount of buffers
            conf.set(TaskManagerOptions.NETWORK_MEMORY_MIN, MemorySize.ofMebiBytes(32));
            conf.set(TaskManagerOptions.NETWORK_MEMORY_MAX, MemorySize.ofMebiBytes(32));
            conf.set(RpcOptions.ASK_TIMEOUT_DURATION, Duration.ofMinutes(1));
            return conf;
        }

        @Override
        public String toString() {
            return "UnalignedSettings{"
                    + "parallelism="
                    + parallelism
                    + ", minCheckpoints="
                    + minCheckpoints
                    + ", restoreCheckpoint="
                    + restoreCheckpoint