@Override
  public void processElement(StreamRecord<MergeOnReadInputSplit> element) {
    splits.add(element.getValue());
    enqueueProcessSplits();
  }
@Test
  public void testCheckpoint() throws Exception {
    // Received emitted splits: split1, split2, split3, split4, checkpoint request is triggered
    // when reading records from split1.
    TestData.writeData(TestData.DATA_SET_INSERT, conf);
    long timestamp = 0;
    try (OneInputStreamOperatorTestHarness<MergeOnReadInputSplit, RowData> harness = createReader()) {
      harness.setup();
      harness.open();

      SteppingMailboxProcessor processor = createLocalMailbox(harness);
      StreamReadMonitoringFunction func = TestUtils.getMonitorFunc(conf);

      List<MergeOnReadInputSplit> splits = generateSplits(func);
      assertThat("Should have 4 splits", splits.size(), is(4));

      for (MergeOnReadInputSplit split : splits) {
        harness.processElement(split, ++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");

      assertTrue(processor.runMailboxStep(), "Should have processed the split0");
      assertTrue(processor.runMailboxStep(), "Should have processed the snapshot state action");

      assertThat(TestData.rowDataToString(harness.extractOutputValues()),
          is(getSplitExpected(Collections.singletonList(splits.get(0)), EXPECTED)));

      // Read records from split1.
      assertTrue(processor.runMailboxStep(), "Should have processed the split1");

      // Read records from split2.
      assertTrue(processor.runMailboxStep(), "Should have processed the split2");

      // Read records from split3.
      assertTrue(processor.runMailboxStep(), "Should have processed the split3");

      // Assert the output has expected elements.
      TestData.assertRowDataEquals(harness.extractOutputValues(), TestData.DATA_SET_INSERT);
    }
  }