@Override
  public void processElement(StreamRecord<FlinkInputSplit> element) {
    splits.add(element.getValue());
    enqueueProcessSplits();
  }
@TestTemplate
  public void testProcessAllRecords() throws Exception {
    List<List<Record>> expectedRecords = generateRecordsAndCommitTxn(10);

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

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

      SteppingMailboxProcessor processor = createLocalMailbox(harness);

      List<Record> expected = Lists.newArrayList();
      for (int i = 0; i < splits.size(); i++) {
        // Process this element to enqueue to mail-box.
        harness.processElement(splits.get(i), -1);

        // Run the mail-box once to read all records from the given split.
        assertThat(processor.runMailboxStep()).as("Should processed 1 split").isTrue();

        // Assert the output has expected elements.
        expected.addAll(expectedRecords.get(i));
        TestHelpers.assertRecords(readOutputValues(harness), expected, SCHEMA);
      }
    }
  }