Flink: fix the deadlock of waiting for extra checkpoints in BoundedTestSource with TestFlinkIcebergSink.testTwoSinksInDisjointedDAG

[stevenzwu]

@kbendick and @nastra found out that this test caused build to stuck in the infinite loop of waiting for required checkpoint to complete

      synchronized (ctx.getCheckpointLock()) {
        while (running && numCheckpointsComplete.get() < checkpointToAwait) {
          ctx.getCheckpointLock().wait(1);
        }
      }

The problem is that TestFlinkIcebergSink.testTwoSinksInDisjointedDAG has two disjointed pipelines. The extra +2 checkpoints waiting broke in this situation where part of the DAG has finished.

        // Let's say checkpointToAwait = numCheckpointsComplete.get() + delta, in fact the value of delta should not
        // affect the final table records because we only need to make sure that there will be exactly
        // elementsPerCheckpoint.size() checkpoints to emit each records buffer from the original elementsPerCheckpoint.
        // Even if the checkpoints that emitted results are not continuous, the correctness of the data should not be
        // affected in the end. Setting the delta to be 2 is introducing the variable that produce un-continuous
        // checkpoints that emit the records buffer from elementsPerCheckpoints.
        checkpointToAwait = numCheckpointsComplete.get() + 2;

While FLIP-147 can make this test work again, we can add a boolean flag to BoundedTestSource that can disable the +2 behavior. what do you think? @kbendick @openinx

[kbendick]

I think placing the +2 is a decent idea, if that works.

If we disable the +2 from the checks, will the test then fail on situations where it was currently stuck in an infinite loop?

I also noticed that there is a callback for checkpoint aborted events. Could we possibly hook into that and on failure check if the union of checkpoint events and events yet to be checkpoint is complete for only 1 of the 2 topologies? Seems really flakey, but maybe you can think of something better than I can for the checkpoint aborted exceptions.

If disabling the +2 wouldn't leave the test flakey, I 100% support that decision.

[kbendick]

Actually

TLDR Since FLIP-147 will remove this issue, I think adding the flag is a good idea. If it leads to flakey tests, we'll then ignore the test again. But there's not really much way to know if it will lead to a flakey test or not without running it in full CI (as I've never been able to replicate this bug outside of CI).

[stevenzwu]

Yeah. I tried to run the TestFlinkIcebergSink single test class in a loop locally and couldn't reproduce the problem. This make it difficult to have confidence to verify the fix.

[stevenzwu]

@kbendick I thought more about your proposal of notifyCheckpointAborted. I think it might be the more reliable solution.

removing the extra checkpoint wait may still suffer the same race condition. When one source is running and emitting record, the other source may already finished.

      final int checkpointToAwait;
      synchronized (ctx.getCheckpointLock()) {
        checkpointToAwait = numCheckpointsComplete.get() + 1 + idleCheckpointCyclesPerStep;
        for (T element : elementsPerCheckpoint.get(checkpoint)) {
          ctx.collect(element);
        }
      }

      synchronized (ctx.getCheckpointLock()) {
        while (running && numCheckpointsComplete.get() < checkpointToAwait) {
          ctx.getCheckpointLock().wait(1);
        }
      }

[openinx]

The problem is that TestFlinkIcebergSink.testTwoSinksInDisjointedDAG has two disjointed pipelines. The extra +2 checkpoints waiting broke in this situation where part of the DAG has finished.

If only one of the two iceberg pipelines has finished the checkpoint state process (while the another haven't), shouldn't it wait another pipeline to finish its checkpoint ? and once all of them finished then trigger the operator's notifyCheckpointComplete. Finally, the two pipelines should increment its numCheckpointsComplete to the target value in my thought.

So did I miss something or misunderstand the case ? @stevenzwu , could you explain more details about the root cause that make it flakey ?

[stevenzwu]

@openinx I think the two parts have calculated different values of checkpointToAwait. E.g., one part computed the value as 3 after one checkpoint already completed, while the other part computed it as 2 before checkpoint completed.

checkpointToAwait = numCheckpointsComplete.get() + 2;

Then we can get into the infinite wait loop because checkpoint #3 will never complete as the other part already finished and no checkpoint can ever succeed.