remove redundant switch statement

Author: polyzos

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/sink/serializer/RowSerializationSchema.java

@@ -20,9 +20,7 @@
 import com.alibaba.fluss.flink.row.RowWithOp;
 import com.alibaba.fluss.row.InternalRow;
 
-import org.apache.flink.table.api.TableException;
 import org.apache.flink.table.data.RowData;
-import org.apache.flink.types.RowKind;
 
 /** Default implementation of RowDataConverter for RowData. */
 public class RowSerializationSchema implements FlussSerializationSchema<RowData> {
@@ -40,22 +38,6 @@ public RowWithOp<RowData> serialize(RowData value) throws Exception {
         }
 
         InternalRow row = converter.replace(value);
-        RowWithOp<RowData> rowWithOp = new RowWithOp<>(row);
-        switch (value.getRowKind()) {
-            case INSERT:
-                rowWithOp.setRowKind(RowKind.INSERT);
-                return rowWithOp;
-            case UPDATE_BEFORE:
-                rowWithOp.setRowKind(RowKind.UPDATE_BEFORE);
-                return rowWithOp;
-            case UPDATE_AFTER:
-                rowWithOp.setRowKind(RowKind.UPDATE_AFTER);
-                return rowWithOp;
-            case DELETE:
-                rowWithOp.setRowKind(RowKind.DELETE);
-                return rowWithOp;
-            default:
-                throw new TableException("Unsupported message kind: " + value.getRowKind());
-        }
+        return new RowWithOp<>(row, value.getRowKind());
     }
 }

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/sink/writer/FlinkSinkWriter.java

@@ -164,7 +164,7 @@ public void write(InputT inputValue, Context context) throws IOException, Interr
         checkAsyncException();
 
         try {
-            RowWithOp rowWithOp = serializationSchema.serialize(inputValue);
+            RowWithOp<InputT> rowWithOp = serializationSchema.serialize(inputValue);
 
             InternalRow internalRow = rowWithOp.getInternalRow();
             RowKind rowKind = rowWithOp.getRowKind();