1. Fixed tag type process; 2. Fixed missing data when batch size more than record number
@@ -73,7 +73,6 @@ public class InfluxDBWriterTask
this.writeTimeout = configuration.getInt(Key.WRITE_TIMEOUT_SECONDS, WRITE_TIMEOUT_SECONDS_DEFAULT);
this.batchSize = configuration.getInt(Key.BATCH_SIZE, 1024);
this.postSqls = configuration.getList(Key.POST_SQL, String.class);
-
}
public void init()
@@ -105,7 +104,7 @@ public class InfluxDBWriterTask
public void destroy()
{
- //
+ this.influxDB.close();
}
public void startWrite(RecordReceiver recordReceiver, TaskPluginCollector taskPluginCollector)
@@ -132,6 +131,10 @@ public class InfluxDBWriterTask
String name = this.columns.get(i).getString("name");
String type = this.columns.get(i).getString("type").toUpperCase();
Column column = record.getColumn(i);
+ if ("TAG".equals(type)) {
+ builder.tag(name, column.asString());
+ }
+ else {
switch (type) {
case "INT":
case "LONG":
@@ -149,17 +152,19 @@ public class InfluxDBWriterTask
case "BINARY":
fields.put(name, column.asBytes());
break;
- case "TAG":
- fields.put(name, column.asString());
- break;
default:
fields.put(name, column.asString());
break;
}
}
+ }
builder.fields(fields);
influxDB.write(database, rp, builder.build());
}
+ // flush last batch manual for avoid missing data
+ if (influxDB.isBatchEnabled()) {
+ influxDB.flush();
+ }
}
catch (Exception e) {
taskPluginCollector.collectDirtyRecord(record, e);
