public void bulkIntoRocksDB() {
    LOG.info("Total size={} in the tmp map, try to bulk into rocksdb", tmpInitializationMap.size());
    int[] count = {0};
    long start = System.currentTimeMillis();

    tmpInitializationMap.forEach(
        (byteArrayWrapper, set) -> {
          rocksDB.put(
              columnFamilyHandle, byteArrayWrapper.bytes, ByteArraySetSerializer.serialize(set));
          set = null;
          count[0] = count[0] + 1;
          if (count[0] % 100000 == 0) {
            LOG.info("Ingested {} into rocksdb.", count[0]);
          }
        });
    tmpInitializationMap.clear();

    LOG.info("Ingested {} completely, cost:{} ms.", count, System.currentTimeMillis() - start);
  }
Retrieve the elements of the key.

<p>Fetch the Collection from guava cache, if not present, fetch the result from RocksDB. if
present, just return the result.

@return not null, but may be empty.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/lookup/RocksDBSetSpilledState.java