protected LookupRecord.OpType convertToOpType(RowKind rowKind) {
    switch (rowKind) {
      case INSERT:
      case UPDATE_AFTER:
        return LookupRecord.OpType.PUT_BYTES;
      case DELETE:
      case UPDATE_BEFORE:
        return LookupRecord.OpType.DELETE_BYTES;
      default:
        throw new IllegalArgumentException(String.format("Not support this rowKind %s", rowKind));
    }
  }
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