support abort the job when sink failed with config (#111)

Author: Nicole00

connector/src/main/java/org.apache.flink/connector/nebula/options/ExecutionOptions.java

@@ -18,21 +18,24 @@ public class ExecutionOptions implements Serializable {
     private final int           batchSize;
     private final int           retryTimes;
     private final long          intervalMs;
+    private final boolean       errorWhenFailed;
 
     protected ExecutionOptions(String graphName,
                                List<String> nebulaFields,
                                List<String> flinkFields,
                                WriteModeEnum writeMode,
                                int batchSize,
                                int retryTimes,
-                               long intervalMs) {
+                               long intervalMs,
+                               boolean errorWhenFailed) {
         this.graphName = graphName;
         this.nebulaFields = nebulaFields;
         this.flinkFields = flinkFields;
         this.writeMode = writeMode;
         this.batchSize = batchSize;
         this.retryTimes = retryTimes;
         this.intervalMs = intervalMs;
+        this.errorWhenFailed = errorWhenFailed;
     }
 
     public String getGraphName() {
@@ -62,4 +65,8 @@ public int getRetryTimes() {
     public long getIntervalMs() {
         return intervalMs;
     }
+
+    public boolean throwErrorWhenFailed() {
+        return errorWhenFailed;
+    }
 }

connector/src/main/java/org.apache.flink/connector/nebula/options/SinkEdgeOptions.java

@@ -6,6 +6,7 @@
 
 package org.apache.flink.connector.nebula.options;
 
+import static org.apache.flink.connector.nebula.utils.NebulaConstant.DEFAULT_ERROR_WHEN_FAILED;
 import static org.apache.flink.connector.nebula.utils.NebulaConstant.DEFAULT_INTERVAL_MILLIS;
 import static org.apache.flink.connector.nebula.utils.NebulaConstant.DEFAULT_RETRY_TIMES;
 import static org.apache.flink.connector.nebula.utils.NebulaConstant.DEFAULT_WRITE_BATCH_SIZE;
@@ -32,7 +33,8 @@ private SinkEdgeOptions(Builder builder) {
               builder.writeMode,
               builder.batchSize,
               builder.retryTimes,
-              builder.intervalMs);
+              builder.intervalMs,
+              builder.errorWhenFailed);
         this.builder = builder;
         this.edgeType = builder.edgeType;
         this.nebulaSrcPks = builder.nebulaSrcPks;
@@ -84,6 +86,7 @@ public static final class Builder implements Serializable {
         private int           batchSize        = DEFAULT_WRITE_BATCH_SIZE;
         private int           retryTimes       = DEFAULT_RETRY_TIMES;
         private long          intervalMs       = DEFAULT_INTERVAL_MILLIS;
+        private boolean       errorWhenFailed  = DEFAULT_ERROR_WHEN_FAILED;
 
 
         public Builder withGraphName(String graphName) {
@@ -154,6 +157,10 @@ public Builder withIntervalMs(long intervalMs) {
             return this;
         }
 
+        public Builder withErrorWhenFailed(boolean errorWhenFailed) {
+            this.errorWhenFailed = errorWhenFailed;
+            return this;
+        }
 
         private void check() {
             if (nebulaFields.size() != flinkFields.size()) {

connector/src/main/java/org.apache.flink/connector/nebula/options/SinkNodeOptions.java

@@ -6,6 +6,7 @@
 
 package org.apache.flink.connector.nebula.options;
 
+import static org.apache.flink.connector.nebula.utils.NebulaConstant.DEFAULT_ERROR_WHEN_FAILED;
 import static org.apache.flink.connector.nebula.utils.NebulaConstant.DEFAULT_INTERVAL_MILLIS;
 import static org.apache.flink.connector.nebula.utils.NebulaConstant.DEFAULT_RETRY_TIMES;
 import static org.apache.flink.connector.nebula.utils.NebulaConstant.DEFAULT_WRITE_BATCH_SIZE;
@@ -22,7 +23,8 @@ public class SinkNodeOptions extends ExecutionOptions {
 
     private SinkNodeOptions(Builder builder) {
         super(builder.graphName, builder.nebulaFields, builder.flinkFields,
-              builder.writeMode, builder.batchSize, builder.retryTimes, builder.intervalMs);
+              builder.writeMode, builder.batchSize, builder.retryTimes, builder.intervalMs,
+              builder.errorWhenFailed);
         this.builder = builder;
         this.nodeType = builder.nodeType;
     }
@@ -49,7 +51,8 @@ public static final class Builder implements Serializable {
         private WriteModeEnum writeMode    = WriteModeEnum.INSERTREPLACE;
         private int           batchSize    = DEFAULT_WRITE_BATCH_SIZE;
 
-        private int retryTimes = DEFAULT_RETRY_TIMES;
+        private int     retryTimes      = DEFAULT_RETRY_TIMES;
+        private boolean errorWhenFailed = DEFAULT_ERROR_WHEN_FAILED;
 
         public Builder withGraphName(String graphName) {
             this.graphName = graphName;
@@ -91,6 +94,11 @@ public Builder withIntervalMs(long intervalMs) {
             return this;
         }
 
+        public Builder withErrorWhenFailed(boolean errorWhenFailed) {
+            this.errorWhenFailed = errorWhenFailed;
+            return this;
+        }
+
         private void check() {
             if (nebulaFields.size() != flinkFields.size()) {
                 throw new IllegalArgumentException(

connector/src/main/java/org.apache.flink/connector/nebula/sink/NebulaEdgeBatchExecutor.java

@@ -7,8 +7,6 @@
 package org.apache.flink.connector.nebula.sink;
 
 
-
-
 import com.vesoft.nebula.driver.graph.data.ResultSet;
 import java.util.ArrayList;
 import java.util.List;
@@ -104,6 +102,9 @@ public String executeBatch(GraphProvider graphProvider) {
             end = System.currentTimeMillis();
         } catch (Exception e) {
             LOG.error(">>>>>> write data error, ", e);
+            if (executionOptions.throwErrorWhenFailed()) {
+                throw new RuntimeException("write edge failed", e);
+            }
             nebulaEdgeList.clear();
             return statement;
         }
@@ -116,6 +117,9 @@ public String executeBatch(GraphProvider graphProvider) {
         } else {
             LOG.error(">>>>> write edge failed: {}", execResult.getErrorMessage());
             LOG.error(">>>>> failed gql: {}", statement);
+            if (executionOptions.throwErrorWhenFailed()) {
+                throw new RuntimeException("write edge failed:" + execResult.getErrorMessage());
+            }
             nebulaEdgeList.clear();
             return statement;
         }

connector/src/main/java/org.apache.flink/connector/nebula/sink/NebulaNodeBatchExecutor.java

@@ -93,6 +93,9 @@ public String executeBatch(GraphProvider graphProvider) {
             end = System.currentTimeMillis();
         } catch (Exception e) {
             LOG.error("write data error, ", e);
+            if (executionOptions.throwErrorWhenFailed()) {
+                throw new RuntimeException("write node failed", e);
+            }
             nebulaVertexList.clear();
             return statement;
         }
@@ -105,6 +108,9 @@ public String executeBatch(GraphProvider graphProvider) {
         } else {
             LOG.error(">>>>> write data failed: {}", execResult.getErrorMessage());
             LOG.error(">>>>> failed gql: {}", statement);
+            if (executionOptions.throwErrorWhenFailed()) {
+                throw new RuntimeException("write node failed:" + execResult.getErrorMessage());
+            }
             nebulaVertexList.clear();
             return statement;
         }

connector/src/main/java/org.apache.flink/connector/nebula/utils/NebulaConstant.java

@@ -26,6 +26,7 @@ public class NebulaConstant {
     public static final long DEFAULT_INTERVAL_MILLIS = 0;
 
     public static final int DEFAULT_RETRY_TIMES = 0;
+    public static final boolean DEFAULT_ERROR_WHEN_FAILED = false;
 
     // default value for connection
     public static final int DEFAULT_CONNECTION_TIMEOUT_MS = 3000;