@Override
  public void processElement(StreamRecord<MergeOnReadInputSplit> element) {
    splits.add(element.getValue());
    enqueueProcessSplits();
  }
@Test
  void testWriteRecords() throws Exception {
    TestData.writeData(TestData.DATA_SET_INSERT, conf);
    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) {
        // Process this element to enqueue to mail-box.
        harness.processElement(split, -1);

        // Run the mail-box once to read all records from the given split.
        assertThat("Should process 1 split", processor.runMailboxStep());
      }
      // Assert the output has expected elements.
      TestData.assertRowDataEquals(harness.extractOutputValues(), TestData.DATA_SET_INSERT);

      TestData.writeData(TestData.DATA_SET_UPDATE_INSERT, conf);
      final List<MergeOnReadInputSplit> splits2 = generateSplits(func);
      assertThat("Should have 4 splits", splits2.size(), is(4));
      for (MergeOnReadInputSplit split : splits2) {
        // Process this element to enqueue to mail-box.
        harness.processElement(split, -1);

        // Run the mail-box once to read all records from the given split.
        assertThat("Should processed 1 split", processor.runMailboxStep());
      }
      // The result sets behaves like append only: DATA_SET_ONE + DATA_SET_TWO
      List<RowData> expected = new ArrayList<>(TestData.DATA_SET_INSERT);
      expected.addAll(TestData.DATA_SET_UPDATE_INSERT);
      TestData.assertRowDataEquals(harness.extractOutputValues(), expected);
    }
  }