public RowData turnRowKind(RowData rowData) {
    switch (rowData.getRowKind()) {
      case INSERT:
        rowData.setRowKind(RowKind.DELETE);
        break;
      case DELETE:
        rowData.setRowKind(RowKind.INSERT);
        break;
      case UPDATE_AFTER:
        rowData.setRowKind(RowKind.UPDATE_BEFORE);
        break;
      case UPDATE_BEFORE:
        rowData.setRowKind(RowKind.UPDATE_AFTER);
        break;
      default:
        throw new FlinkRuntimeException("unKnown ChangeAction=" + rowData.getRowKind());
    }
    LOG.debug("after retract a row, ChangeAction={}", rowData.getRowKind());
    return rowData;
  }
Turn row kind of a row.

<pre>
+I -> -D
-D -> +I
-U -> +U
+U -> -U
</pre>

@param rowData Before reset row
@return After reset row kind.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/source/log/LogSourceHelper.java