[feat][stream] Add Stream Connector, print rowData RowKind in first Column.

Author: chaozwn

flinkx-connectors/flinkx-connector-stream/src/main/java/com/dtstack/flinkx/connector/stream/sink/StreamOutputFormat.java

@@ -55,7 +55,8 @@ protected void writeSingleRecordInternal(RowData rowData) throws WriteRecordExce
             RowData row =
                     (RowData)
                             rowConverter.toExternal(
-                                    rowData, new GenericRowData(rowData.getArity()));
+                                    rowData,
+                                    new GenericRowData(rowData.getRowKind(), rowData.getArity()));
             if (streamConf.getPrint()) {
                 TablePrintUtil.printTable(row, getFieldNames(rowData));
             }

flinkx-connectors/flinkx-connector-stream/src/main/java/com/dtstack/flinkx/connector/stream/util/TablePrintUtil.java

@@ -167,18 +167,25 @@ public static void printTable(RowData row, String[] fieldNames) {
         }
 
         List<String[]> data = new ArrayList<>(2);
+        String[] recordStr = new String[genericRowData.getArity() + 1];
+        recordStr[0] = row.getRowKind().toString();
         boolean emptyFieldNames = false;
         if (fieldNames == null) {
-            fieldNames = new String[genericRowData.getArity()];
+            fieldNames = new String[genericRowData.getArity() + 1];
             emptyFieldNames = true;
+        } else {
+            String[] newFieldNames = new String[genericRowData.getArity() + 1];
+            System.arraycopy(fieldNames, 0, newFieldNames, 1, genericRowData.getArity());
+            fieldNames = newFieldNames;
         }
-        String[] recordStr = new String[genericRowData.getArity()];
+        fieldNames[0] = "rowKind";
         for (int i = 0; i < genericRowData.getArity(); i++) {
             if (emptyFieldNames) {
-                fieldNames[i] = "col" + i;
+                fieldNames[i + 1] = "col" + i;
             }
-            recordStr[i] = String.valueOf(genericRowData.getField(i));
+            recordStr[i + 1] = String.valueOf(genericRowData.getField(i));
         }
+
         data.add(fieldNames);
         data.add(recordStr);
         TablePrintUtil.build(data).print();