[iotdb] branch master updated: [IoTDB-1449] path already exist error in IoTDBSink (#3569)

[chaow]

This is an automated email from the ASF dual-hosted git repository.

chaow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new abfa695  [IoTDB-1449] path already exist error in IoTDBSink (#3569)
abfa695 is described below

commit abfa6954cd228401e5907b34eafd32c6861fcfa8
Author: Ring-k <36235611+rin...@users.noreply.github.com>
AuthorDate: Wed Jul 14 17:58:36 2021 +0800

    [IoTDB-1449] path already exist error in IoTDBSink (#3569)
---
 .../src/main/java/org/apache/iotdb/flink/IoTDBSink.java       | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git 
a/flink-iotdb-connector/src/main/java/org/apache/iotdb/flink/IoTDBSink.java 
b/flink-iotdb-connector/src/main/java/org/apache/iotdb/flink/IoTDBSink.java
index eaba61d..95e40a2 100644
--- a/flink-iotdb-connector/src/main/java/org/apache/iotdb/flink/IoTDBSink.java
+++ b/flink-iotdb-connector/src/main/java/org/apache/iotdb/flink/IoTDBSink.java
@@ -97,8 +97,15 @@ public class IoTDBSink<IN> extends RichSinkFunction<IN> {
 
     for (IoTDBSinkOptions.TimeseriesOption option : 
options.getTimeseriesOptionList()) {
       if (!pool.checkTimeseriesExists(option.getPath())) {
-        pool.createTimeseries(
-            option.getPath(), option.getDataType(), option.getEncoding(), 
option.getCompressor());
+        try {
+          pool.createTimeseries(
+              option.getPath(), option.getDataType(), option.getEncoding(), 
option.getCompressor());
+        } catch (StatementExecutionException e) {
+          // path could have been created by the other process here
+          if (e.getStatusCode() != 
TSStatusCode.PATH_ALREADY_EXIST_ERROR.getStatusCode()) {
+            throw e;
+          }
+        }
       }
     }
   }