@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 testMaxContinuousEmptyCommits() throws Exception {
    table.updateProperties().set(MAX_CONTINUOUS_EMPTY_COMMITS, "3").commit();

    JobID jobId = new JobID();
    long checkpointId = 0;
    long timestamp = 0;
    try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness = createStreamSink(jobId)) {
      harness.setup();
      harness.open();

      assertSnapshotSize(0);

      for (int i = 1; i <= 9; i++) {
        harness.snapshot(++checkpointId, ++timestamp);
        harness.notifyOfCompletedCheckpoint(checkpointId);

        assertSnapshotSize(i / 3);
      }
    }
  }