[feat-4789][influxdb] nep fix

Author: [email protected]

flinkx-connectors/flinkx-connector-influxdb/src/main/java/com/dtstack/flinkx/connector/influxdb/conf/InfluxdbSinkConfig.java

@@ -3,6 +3,7 @@
 import com.dtstack.flinkx.sink.WriteMode;
 
 import java.util.List;
+import java.util.Locale;
 
 /** @Author xirang @Company Dtstack @Date: 2022/3/14 6:00 PM */
 public class InfluxdbSinkConfig extends InfluxdbConfig {
@@ -16,7 +17,7 @@ public class InfluxdbSinkConfig extends InfluxdbConfig {
     /** tags of the measurement */
     private List<String> tags;
 
-    private int batchSize = 10000;
+    private int bufferLimit = 10000;
 
     /** flush duration (ms) */
     private int flushDuration = 1000;
@@ -75,8 +76,20 @@ public WriteMode getWriteMode() {
         return writeMode;
     }
 
-    public void setWriteMode(WriteMode writeMode) {
-        this.writeMode = writeMode;
+    public void setWriteMode(String writeMode) {
+        switch (writeMode.toLowerCase(Locale.ENGLISH)) {
+            case "insert":
+                this.writeMode = WriteMode.INSERT;
+                break;
+            case "update":
+                this.writeMode = WriteMode.UPDATE;
+                break;
+            case "upsert":
+                this.writeMode = WriteMode.UPSERT;
+                break;
+            default:
+                this.writeMode = WriteMode.APPEND;
+        }
     }
 
     public List<String> getTags() {
@@ -87,14 +100,12 @@ public void setTags(List<String> tags) {
         this.tags = tags;
     }
 
-    @Override
-    public int getBatchSize() {
-        return batchSize;
+    public int getBufferLimit() {
+        return bufferLimit;
     }
 
-    @Override
-    public void setBatchSize(int batchSize) {
-        this.batchSize = batchSize;
+    public void setBufferLimit(int bufferLimit) {
+        this.bufferLimit = bufferLimit;
     }
 
     public String getPrecision() {

flinkx-connectors/flinkx-connector-influxdb/src/main/java/com/dtstack/flinkx/connector/influxdb/converter/InfluxdbColumnConverter.java

@@ -114,7 +114,7 @@ public InfluxdbColumnConverter(
     protected ISerializationConverter<Point.Builder> wrapIntoNullableExternalConverter(
             ISerializationConverter<Point.Builder> converter, LogicalType type) {
         return (val, index, builder) -> {
-            if (val == null) {
+            if (val == null || val.isNullAt(index)) {
                 return;
             } else {
                 converter.serialize(val, index, builder);

flinkx-connectors/flinkx-connector-influxdb/src/main/java/com/dtstack/flinkx/connector/influxdb/sink/InfluxdbOutputFormat.java

@@ -16,9 +16,9 @@
 import okhttp3.OkHttpClient;
 import org.influxdb.BatchOptions;
 import org.influxdb.InfluxDB;
-import org.influxdb.InfluxDBFactory;
 import org.influxdb.dto.BatchPoints;
 import org.influxdb.dto.Point;
+import org.influxdb.impl.InfluxDBImpl;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -88,13 +88,7 @@ protected void openInternal(int taskNumber, int numTasks) throws IOException {
                         columnNameList, columnTypeList, InfluxdbRawTypeConverter::apply);
         setRowConverter(
                 new InfluxdbColumnConverter(
-                        rowType,
-                        sinkConfig,
-                        measurement,
-                        columnNameList,
-                        tags,
-                        timestamp,
-                        precision));
+                        rowType, sinkConfig, columnNameList, tags, timestamp, precision));
     }
 
     @Override
@@ -111,7 +105,7 @@ private void establishConnnection() {
                         .connectTimeout(15000, TimeUnit.MILLISECONDS)
                         .readTimeout(sinkConfig.getWriteTimeout(), TimeUnit.SECONDS);
         influxDB =
-                InfluxDBFactory.connect(
+                new InfluxDBImpl(
                         sinkConfig.getUrl().get(0),
                         StringUtils.isNullOrWhitespaceOnly(sinkConfig.getUsername())
                                 ? null
@@ -125,7 +119,7 @@ private void establishConnnection() {
         if (enableBatch) {
             BatchOptions options = BatchOptions.DEFAULTS;
             options.precision(precision);
-            options.bufferLimit(sinkConfig.getBatchSize());
+            options.bufferLimit(sinkConfig.getBufferLimit());
             options.flushDuration(sinkConfig.getFlushDuration());
             influxDB.enableBatch(options);
         }

flinkx-connectors/flinkx-connector-influxdb/src/main/java/com/dtstack/flinkx/connector/influxdb/sink/InfluxdbOutputFormatBuilder.java

@@ -4,7 +4,6 @@
 import com.dtstack.flinkx.sink.format.BaseRichOutputFormatBuilder;
 
 import org.apache.flink.util.CollectionUtil;
-import org.apache.flink.util.StringUtils;
 
 /** @Author xirang @Company Dtstack @Date: 2022/3/14 2:57 PM */
 public class InfluxdbOutputFormatBuilder extends BaseRichOutputFormatBuilder {
@@ -25,10 +24,6 @@ protected void checkFormat() {
         InfluxdbSinkConfig sinkConfig = format.getSinkConfig();
         StringBuilder sb = new StringBuilder(256);
         if (CollectionUtil.isNullOrEmpty(sinkConfig.getUrl())) sb.append("No url supplied;\n");
-        if (StringUtils.isNullOrWhitespaceOnly(sinkConfig.getUsername()))
-            sb.append("No username supplied;\n");
-        if (StringUtils.isNullOrWhitespaceOnly(sinkConfig.getPassword()))
-            sb.append("No password supplied;\n");
         if (!"insert".equalsIgnoreCase(sinkConfig.getWriteMode().getMode()))
             sb.append("Only support insert write mode;\n");
         if (sb.length() > 0) {

flinkx-connectors/flinkx-connector-influxdb/src/main/java/com/dtstack/flinkx/connector/influxdb/sink/InfluxdbSinkFactory.java

@@ -27,6 +27,8 @@ public InfluxdbSinkFactory(SyncConf syncConf) {
         Gson gson = new GsonBuilder().create();
         GsonUtil.setTypeAdapter(gson);
         this.influxdbConfig = gson.fromJson(gson.toJson(parameter), InfluxdbSinkConfig.class);
+        Object writeMode = parameter.get("writeMode");
+        influxdbConfig.setWriteMode(writeMode == null ? null : writeMode.toString());
         influxdbConfig.setColumn(syncConf.getWriter().getFieldList());
         super.initFlinkxCommonConf(influxdbConfig);
     }