@Override
    public void run() {
      LOG.info("{} starting.", name);
      try {
        while (!closed.get() && !initialized.get()) {
          LookupRecord record = lookupRecordsQueue.poll();
          if (record != null) {
            switch (record.opType()) {
              case PUT_BYTES:
                put(record);
                break;
              case DELETE_BYTES:
                delete(record);
                break;
              default:
                throw new IllegalArgumentException(
                    String.format("Not support this OpType %s", record.opType()));
            }
          }
        }
      } catch (Throwable e) {
        LOG.error("writing failed:", e);
        writingThreadException.set(e);
      }
      LOG.info("{} stopping.", name);
    }
This task is running during the initialization phase to write data{@link LookupRecord} to
RocksDB.

<p>During the initialization phase, the Merge-on-Read approach is used to retrieve data, which
will only return INSERT data. When there are multiple entries with the same primary key, only
one entry will be returned.

<p>During the initialization phase, the incremental pull approach is also used to retrieve data
that include four {@link RowKind} rowKinds, -D, +I, -U, and +U.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/lookup/RocksDBCacheState.java