allow custom row-data to passthrough

padmejin · Jan 26, 2022 · 7e7e3ce · 7e7e3ce
1 parent 88f3447
commit 7e7e3ce
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 12 deletions.
diff --git a/src/main/java/com/starrocks/connector/flink/manager/StarRocksStreamLoadVisitor.java b/src/main/java/com/starrocks/connector/flink/manager/StarRocksStreamLoadVisitor.java
@@ -93,7 +93,7 @@ public Map<String, Object> doStreamLoad(Tuple3<String, Long, ArrayList<byte[]>>
         Map<String, Object> loadResult = doHttpPut(loadUrl, labeledRows.f0, joinRows(labeledRows.f2, labeledRows.f1.intValue()));
         final String keyStatus = "Status";
         if (null == loadResult || !loadResult.containsKey(keyStatus)) {
-            throw new IOException("Unable to flush data to StarRocks: unknown result status, usually caused by: 1.authentication or permission related problems. 2.Wrong column_separator or row_delimiter. 3.Column count exceeded the limitation.");
+            throw new IOException("Unable to flush data to StarRocks: unknown result status, usually caused by: 1.authorization or permission related problems. 2.Wrong column_separator or row_delimiter. 3.Column count exceeded the limitation.");
         }
         if (LOG.isDebugEnabled()) {
             LOG.debug(String.format("Stream Load response: \n%s\n", JSON.toJSONString(loadResult)));

diff --git a/src/main/java/com/starrocks/connector/flink/table/StarRocksDynamicSinkFunction.java b/src/main/java/com/starrocks/connector/flink/table/StarRocksDynamicSinkFunction.java
@@ -131,17 +131,15 @@ public synchronized void invoke(T value, Context context) throws Exception {
                 Alter alter = (Alter) stmt;
             }
         }
-        if (!(value instanceof RowData)) {
-            LOG.warn("Unsupported row data invoked: [%s]", value.getClass().getName());
-            return;
-        }
-        if (RowKind.UPDATE_BEFORE.equals(((RowData)value).getRowKind())) {
-            // do not need update_before, cauz an update action happened on the primary keys will be separated into `delete` and `create`
-            return;
-        }
-        if (!sinkOptions.supportUpsertDelete() && RowKind.DELETE.equals(((RowData)value).getRowKind())) {
-            // let go the UPDATE_AFTER and INSERT rows for tables who have a group of `unique` or `duplicate` keys.
-            return;
+        if (value instanceof RowData) {
+            if (RowKind.UPDATE_BEFORE.equals(((RowData)value).getRowKind())) {
+                // do not need update_before, cauz an update action happened on the primary keys will be separated into `delete` and `create`
+                return;
+            }
+            if (!sinkOptions.supportUpsertDelete() && RowKind.DELETE.equals(((RowData)value).getRowKind())) {
+                // let go the UPDATE_AFTER and INSERT rows for tables who have a group of `unique` or `duplicate` keys.
+                return;
+            }
         }
         sinkManager.writeRecord(
             serializer.serialize(rowTransformer.transform(value, sinkOptions.supportUpsertDelete()))