[feat-4789][influxdb]

Fix data precision issues and influxdb java cli swallow exceptions issues

Author: [email protected]

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

@@ -210,11 +210,11 @@ protected ISerializationConverter<Point.Builder> createExternalConverter(Logical
                 };
             case FLOAT:
                 return (val, index, builder) -> {
-                    builder.addField(fieldNameList.get(index), val.getFloat(index));
+                    builder.addField(fieldNameList.get(index), val.getDouble(index));
                 };
             case INTEGER:
                 return (val, index, builder) -> {
-                    builder.addField(fieldNameList.get(index), val.getInt(index));
+                    builder.addField(fieldNameList.get(index), val.getLong(index));
                 };
             case BOOLEAN:
                 return (val, index, builder) -> {
@@ -236,7 +236,8 @@ private boolean specicalField(
                 builder.time(value.getLong(index), precision);
                 return true;
             } else if (CollectionUtils.isNotEmpty(tags) && tags.contains(fieldName)) {
-                builder.tag(fieldName, ((ColumnRowData) value).getField(index).asString());
+                if (!value.isNullAt(index))
+                    builder.tag(fieldName, ((ColumnRowData) value).getField(index).asString());
                 return true;
             }
         }

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

@@ -7,6 +7,7 @@
 import com.dtstack.flinkx.connector.influxdb.enums.TimePrecisionEnums;
 import com.dtstack.flinkx.sink.format.BaseRichOutputFormat;
 import com.dtstack.flinkx.throwable.WriteRecordException;
+import com.dtstack.flinkx.util.ExceptionUtil;
 import com.dtstack.flinkx.util.TableUtil;
 
 import org.apache.flink.table.data.RowData;
@@ -24,6 +25,7 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -70,7 +72,7 @@ protected void writeMultipleRecordsInternal() throws Exception {
             pointList.add(builder.build());
         }
         batchPoints.points(pointList);
-        influxDB.writeWithRetry(batchPoints.build());
+        influxDB.write(batchPoints.build());
     }
 
     @Override
@@ -118,9 +120,22 @@ private void establishConnnection() {
         if (!StringUtils.isNullOrWhitespaceOnly(rp)) influxDB.setRetentionPolicy(rp);
         if (enableBatch) {
             BatchOptions options = BatchOptions.DEFAULTS;
-            options.precision(precision);
-            options.bufferLimit(sinkConfig.getBufferLimit());
-            options.flushDuration(sinkConfig.getFlushDuration());
+            options =
+                    options.exceptionHandler(
+                                    (iterable, e) -> {
+                                        Iterator<Point> iterator = iterable.iterator();
+                                        while (iterator.hasNext()) {
+                                            dirtyManager.collect(iterator.next(), e, null);
+                                        }
+                                        if (LOG.isTraceEnabled()) {
+                                            LOG.trace(
+                                                    "write data error, e = {}",
+                                                    ExceptionUtil.getErrorMessage(e));
+                                        }
+                                    })
+                            .precision(precision)
+                            .bufferLimit(sinkConfig.getBufferLimit())
+                            .flushDuration(sinkConfig.getFlushDuration());
             influxDB.enableBatch(options);
         }
     }