public void asyncWrite(RowData joinKey, byte[] uniqueKeyBytes) throws IOException {
    byte[] joinKeyBytes = serializeKey(joinKey);
    LookupRecord.OpType opType = convertToOpType(joinKey.getRowKind());
    putIntoQueue(LookupRecord.of(opType, joinKeyBytes, uniqueKeyBytes));
  }
Multi-threads would put and delete the joinKeys and Set<ByteArrayWrapper> in the rocksdb.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/lookup/RocksDBSetSpilledState.java