@Override
  public byte[] serializeKey(RowData key) throws IOException {
    if (joinKeySerializerThreadLocal.get() == null) {
      joinKeySerializerThreadLocal.set(joinKeySerializer.clone());
    }
    return serializeKey(joinKeySerializerThreadLocal.get(), key);
  }
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