@Override
    public void run(SourceContext<RowData> ctx) throws Exception {
      if (isFailoverScenario && getRuntimeContext().getAttemptNumber() == 0) {
        // In the first execution, we first send a part of record...
        sendRecordsUntil((int) (numberOfRecords * FAILOVER_RATIO * 0.5), ctx);

        // Wait till the first part of data is committed.
        while (!hasCompletedCheckpoint) {
          Thread.sleep(50);
        }

        // Then we write the second part of data...
        sendRecordsUntil((int) (numberOfRecords * FAILOVER_RATIO), ctx);

        // And then trigger the failover.
        if (getRuntimeContext().getIndexOfThisSubtask() == 0) {
          throw new RuntimeException("Designated Exception");
        } else {
          while (true) {
            Thread.sleep(50);
          }
        }
      } else {
        // If we are not going to trigger failover or we have already triggered failover,
        // run until finished.
        sendRecordsUntil(numberOfRecords, ctx);

        // Wait the last checkpoint to commit all the pending records.
        isWaitingCheckpointComplete = true;
        CountDownLatch latch = LATCH_MAP.get(latchId);
        latch.await();
      }
    }
Whether the test is executing in a scenario that induces a failover. This doesn't mean that
this source induces the failover.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/write/MixedFormatFileWriterITCase.java