@Override
  public void open() throws Exception {
    super.open();

    final String operatorID = getRuntimeContext().getOperatorUniqueID();
    this.workerPool =
        ThreadPools.newWorkerPool("iceberg-worker-pool-" + operatorID, workerPoolSize);
  }
@TestTemplate
  public void testCommitTxnWithoutDataFiles() throws Exception {
    long checkpointId = 0;
    long timestamp = 0;
    JobID jobId = new JobID();
    OperatorID operatorId;
    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
      harness.setup();
      harness.open();
      operatorId = harness.getOperator().getOperatorID();

      SimpleDataUtil.assertTableRows(table, Lists.newArrayList(), branch);
      assertSnapshotSize(0);
      assertMaxCommittedCheckpointId(jobId, operatorId, -1L);

      // It's better to advance the max-committed-checkpoint-id in iceberg snapshot, so that the
      // future flink job
      // failover won't fail.
      for (int i = 1; i <= 3; i++) {
        harness.snapshot(++checkpointId, ++timestamp);
        assertFlinkManifests(0);

        harness.notifyOfCompletedCheckpoint(checkpointId);
        assertFlinkManifests(0);

        assertSnapshotSize(i);
        assertMaxCommittedCheckpointId(jobId, operatorId, checkpointId);
      }
    }
  }