@Override
  public void processElement(StreamRecord<T> element) throws Exception {
    writer.write(element.getValue());
  }
@TestTemplate
  public void testTableWithoutSnapshot() throws Exception {
    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
        createIcebergStreamWriter()) {
      assertThat(testHarness.extractOutputValues()).isEmpty();
    }
    // Even if we closed the iceberg stream writer, there's no orphan data file.
    assertThat(scanDataFiles()).isEmpty();

    try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
        createIcebergStreamWriter()) {
      testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
      // Still not emit the data file yet, because there is no checkpoint.
      assertThat(testHarness.extractOutputValues()).isEmpty();
    }
    // Once we closed the iceberg stream writer, there will left an orphan data file.
    assertThat(scanDataFiles()).hasSize(1);
  }