fix possible over size exception when batch udp

README.md

@@ -80,12 +80,13 @@ influxdb-java client doesn't enable gzip compress for http request body by defau
 
 UDP's support:
 
-influxdb-java client support udp protocol now. you can call followed methods directly to wirte through UDP.
+influxdb-java client support udp protocol now. you can call followed methods directly to write through UDP.
 ```
 public void write(final int udpPort, final String records);
 public void write(final int udpPort, final List<String> records);
 public void write(final int udpPort, final Point point);
 ```
+note: make sure write content's total size should not > UDP protocol's limit(64K), or you should use http instead of udp.
 
 ### Changes in 2.4
 influxdb-java now uses okhttp3 and retrofit2.  As a result, you can now pass an ``OkHttpClient.Builder``

src/main/java/org/influxdb/impl/BatchProcessor.java

@@ -221,7 +221,9 @@ void write() {
           BatchProcessor.this.influxDB.write(batchPoints);
       }
       for (Entry<Integer, List<String>> entry : udpPortToBatchPoints.entrySet()) {
-          BatchProcessor.this.influxDB.write(entry.getKey(), entry.getValue());
+          for (String lineprotocolStr : entry.getValue()) {
+              BatchProcessor.this.influxDB.write(entry.getKey(), lineprotocolStr);
+          }
       }
     } catch (Throwable t) {
       // any exception wouldn't stop the scheduler

src/test/java/org/influxdb/InfluxDBTest.java

@@ -8,6 +8,7 @@
 import org.junit.Test;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Set;
@@ -252,6 +253,31 @@ public void testWriteMultipleStringDataLinesThroughUDP() {
         Assert.assertEquals(result.getResults().get(0).getSeries().get(2).getTags().get("atag"), "test3"); 
     }
 
+    /**
+     * When batch of points' size is over UDP limit, the expected exception
+     * is java.lang.RuntimeException: java.net.SocketException: 
+     * The message is larger than the maximum supported by the underlying transport: Datagram send failed
+     * @throws Exception
+     */
+    @Test(expected = RuntimeException.class)
+    public void writeMultipleStringDataLinesOverUDPLimit() throws Exception {
+        //prepare data
+        List<String> lineProtocols = new ArrayList<String>();
+        int i = 0;
+        int length = 0;
+        while ( true ) {
+            Point point = Point.measurement("udp_single_poit").addField("v", i).build();
+            String lineProtocol = point.lineProtocol();
+            length += (lineProtocol.getBytes("utf-8")).length;
+            lineProtocols.add(lineProtocol);
+            if( length > 65535 ){
+                break;
+            }
+        }
+        //write batch of string which size is over 64K
+        this.influxDB.write(UDP_PORT, lineProtocols);
+    }
+
     /**
      * Test writing multiple records to the database using string protocol.
      */

src/test/java/org/influxdb/PerformanceTests.java

@@ -4,22 +4,38 @@
 import org.influxdb.InfluxDB.LogLevel;
 import org.influxdb.dto.BatchPoints;
 import org.influxdb.dto.Point;
+import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 public class PerformanceTests {
 	private InfluxDB influxDB;
 	private final static int COUNT = 1;
 	private final static int POINT_COUNT = 100000;
 	private final static int SINGLE_POINT_COUNT = 10000;
+
+	private final static int UDP_PORT = 8089;
+	private final static String UDP_DATABASE = "udp";
 
 	@Before
 	public void setUp() {
 		this.influxDB = InfluxDBFactory.connect("http://" + TestUtils.getInfluxIP() + ":" + TestUtils.getInfluxPORT(true), "root", "root");
 		this.influxDB.setLogLevel(LogLevel.NONE);
+		this.influxDB.createDatabase(UDP_DATABASE);
+	}
+
+	/**
+	 * delete UDP database after all tests end.
+	 */
+	@After
+	public void clearup(){
+		this.influxDB.deleteDatabase(UDP_DATABASE);
 	}
 
 	@Test
@@ -87,4 +103,32 @@ public void maxWritePointsPerformance() {
 		System.out.println("5Mio points:" + watch);
 		this.influxDB.deleteDatabase(dbName);
 	}
+
+	@Test
+	public void writeCompareUDPPerformanceForBatchWithSinglePoints() {
+		//prepare data
+		List<String> lineProtocols = new ArrayList<String>();
+		for (int i = 0; i < 1000; i++) {
+		    Point point = Point.measurement("udp_single_poit").addField("v", i).build();
+		    lineProtocols.add(point.lineProtocol());
+		}
+
+		//write batch of 1000 single string.
+		Stopwatch watch = Stopwatch.createStarted();
+		this.influxDB.write(UDP_PORT, lineProtocols);
+		long elapsedForBatchWrite = watch.elapsed(TimeUnit.MILLISECONDS);
+		System.out.println("performance(ms):write udp with batch of 1000 string:" + elapsedForBatchWrite);
+
+		//write 1000 single string by udp.
+		watch = Stopwatch.createStarted();
+		for (String lineProtocol: lineProtocols){
+		    this.influxDB.write(UDP_PORT, lineProtocol);
+		}
+
+		long elapsedForSingleWrite = watch.elapsed(TimeUnit.MILLISECONDS);
+		System.out.println("performance(ms):write udp with 1000 single strings:" + elapsedForSingleWrite);
+
+		Assert.assertTrue(elapsedForSingleWrite - elapsedForBatchWrite > 0);
+	}
+
 }