public void waitWriteRocksDBDone() {
    long every5SecondsPrint = Long.MIN_VALUE;

    while (true) {
      if (lookupRecordsQueue.isEmpty()) {
        initialized.set(true);
        break;
      } else if (every5SecondsPrint < System.currentTimeMillis()) {
        LOG.info("Currently rocksDB queue size is {}.", lookupRecordsQueue.size());
        every5SecondsPrint = System.currentTimeMillis() + 5000;
      }
    }
    // Wait for all threads to finish
    for (Future<?> future : writeRocksDBThreadFutures) {
      try {
        // wait for the task to complete, with a timeout of 5 seconds
        future.get(5, TimeUnit.SECONDS);
      } catch (TimeoutException e) {
        // task took too long, interrupt the thread and terminate the task
        future.cancel(true);
      } catch (InterruptedException | ExecutionException e) {
        // handle other exceptions
        throw new FlinkRuntimeException(e);
      }
    }
  }
Waiting for the writing threads completed.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/lookup/RocksDBCacheState.java