@Override
  public void processElement(StreamRecord<FlinkInputSplit> element) {
    splits.add(element.getValue());
    enqueueProcessSplits();
  }
@TestTemplate
  public void testTriggerCheckpoint() throws Exception {
    // Received emitted splits: split1, split2, split3, checkpoint request is triggered when reading
    // records from
    // split1.
    List<List<Record>> expectedRecords = generateRecordsAndCommitTxn(3);

    List<FlinkInputSplit> splits = generateSplits();
    assertThat(splits).hasSize(3);

    long timestamp = 0;
    try (OneInputStreamOperatorTestHarness<FlinkInputSplit, RowData> harness = createReader()) {
      harness.setup();
      harness.open();

      SteppingMailboxProcessor processor = createLocalMailbox(harness);

      harness.processElement(splits.get(0), ++timestamp);
      harness.processElement(splits.get(1), ++timestamp);
      harness.processElement(splits.get(2), ++timestamp);

      // Trigger snapshot state, it will start to work once all records from split0 are read.
      processor.getMainMailboxExecutor().execute(() -> harness.snapshot(1, 3), "Trigger snapshot");

      assertThat(processor.runMailboxStep()).as("Should have processed the split0").isTrue();
      assertThat(processor.runMailboxStep())
          .as("Should have processed the snapshot state action")
          .isTrue();

      TestHelpers.assertRecords(readOutputValues(harness), expectedRecords.get(0), SCHEMA);

      // Read records from split1.
      assertThat(processor.runMailboxStep()).as("Should have processed the split1").isTrue();

      // Read records from split2.
      assertThat(processor.runMailboxStep()).as("Should have processed the split2").isTrue();

      TestHelpers.assertRecords(
          readOutputValues(harness), Lists.newArrayList(Iterables.concat(expectedRecords)), SCHEMA);
    }
  }