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

Code before:
            System.out.println(
                    "Starting " + getClass().getCanonicalName() + "#" + name.getMethodName() + ".");
            final CompletableFuture<JobSubmissionResult> result =
                    miniCluster.getMiniCluster().submitJob(streamGraph.getJobGraph());

            checkCounters(
                    miniCluster
                            .getMiniCluster()
                            .requestJobResult(result.get().getJobID())
                            .get()
                            .toJobExecutionResult(getClass().getClassLoader()));
            System.out.println(
                    "Finished " + getClass().getCanonicalName() + "#" + name.getMethodName() + ".");
        } catch (Exception e) {
            if (!ExceptionUtils.findThrowable(e, TestException.class).isPresent()) {
                throw e;
            }
        } finally {
            miniCluster.after();
        }
        if (settings.generateCheckpoint) {
            return TestUtils.getMostRecentCompletedCheckpoint(checkpointDir);
        }
        return null;
    }

    private StreamGraph getStreamGraph(UnalignedSettings settings, Configuration conf) {
        // a dummy environment used to retrieve the DAG, mini cluster will be used later
        final StreamExecutionEnvironment setupEnv =
                StreamExecutionEnvironment.createLocalEnvironment(conf);
        settings.configure(setupEnv);

        settings.dagCreator.create(
                setupEnv,
                settings.minCheckpoints,
                settings.channelType.slotSharing,
                settings.expectedFailures - settings.failuresAfterSourceFinishes,
                settings.sourceSleepMs);

        return setupEnv.getStreamGraph();
    }

    protected abstract void checkCounters(JobExecutionResult result);

    /** A source that generates longs in a fixed number of splits. */
    protected static class LongSource
            implements Source<Long, LongSource.LongSplit, LongSource.EnumeratorState> {
        private final int minCheckpoints;
        private final int numSplits;
        private final int expectedRestarts;
        private final long checkpointingInterval;

Review comment (suggestion):
Find latest checkpoint should be placed in catch block as well, right?

The `TestException` always happen in the test, it caused empty file is returned.

Code after:
        try {
            // print the test parameters to help debugging when the case is stuck
            System.out.println(
                    "Starting " + getClass().getCanonicalName() + "#" + name.getMethodName() + ".");
            final CompletableFuture<JobSubmissionResult> result =
                    miniCluster.getMiniCluster().submitJob(streamGraph.getJobGraph());

            final JobID jobID = result.get().getJobID();
            checkCounters(
                    miniCluster
                            .getMiniCluster()
                            .requestJobResult(jobID)
                            .get()
                            .toJobExecutionResult(getClass().getClassLoader()));
            System.out.println(
                    "Finished " + getClass().getCanonicalName() + "#" + name.getMethodName() + ".");
            if (settings.generateCheckpoint) {
                return CommonTestUtils.getLatestCompletedCheckpointPath(
                                jobID, miniCluster.getMiniCluster())
                        .map(File::new)
                        .orElseThrow(() -> new AssertionError("Could not generate checkpoint"));
            }
        } catch (Exception e) {
            if (ExceptionUtils.findThrowable(e, TestException.class).isEmpty()) {
                throw e;
            }
        } finally {
            miniCluster.after();
        }
        return null;
    }

    private StreamGraph getStreamGraph(UnalignedSettings settings, Configuration conf) {
        // a dummy environment used to retrieve the DAG, mini cluster will be used later
        final StreamExecutionEnvironment setupEnv =
                StreamExecutionEnvironment.createLocalEnvironment(conf);
        settings.configure(setupEnv);

        settings.dagCreator.create(
                setupEnv,
                settings.minCheckpoints,
                settings.channelType.slotSharing,
                settings.expectedFailures - settings.failuresAfterSourceFinishes,
                settings.sourceSleepMs);

        return setupEnv.getStreamGraph();
    }

    protected abstract void checkCounters(JobExecutionResult result);

    /** A source that generates longs in a fixed number of splits. */