BatchProcessor exception handling so that scheduler is not cancelled after connection error

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

@@ -8,6 +8,8 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 import org.influxdb.InfluxDB;
 import org.influxdb.dto.BatchPoints;
@@ -24,6 +26,8 @@
  *
  */
 public class BatchProcessor {
+
+	private static final Logger logger = Logger.getLogger(BatchProcessor.class.getName());
 	protected final BlockingQueue<BatchEntry> queue = new LinkedBlockingQueue<>();
 	private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
 	final InfluxDBImpl influxDB;
@@ -144,26 +148,31 @@ public void run() {
 	}
 
 	void write() {
-		if (this.queue.isEmpty()) {
-			return;
-		}
-
-		Map<String, BatchPoints> databaseToBatchPoints = Maps.newHashMap();
-		List<BatchEntry> batchEntries = new ArrayList<>(this.queue.size());
-		this.queue.drainTo(batchEntries);
+		try {
+			if (this.queue.isEmpty()) {
+				return;
+			}
 
-		for (BatchEntry batchEntry : batchEntries) {
-			String dbName = batchEntry.getDb();
-			if (!databaseToBatchPoints.containsKey(dbName)) {
-				BatchPoints batchPoints = BatchPoints.database(dbName).retentionPolicy(batchEntry.getRp()).build();
-				databaseToBatchPoints.put(dbName, batchPoints);
+			Map<String, BatchPoints> databaseToBatchPoints = Maps.newHashMap();
+			List<BatchEntry> batchEntries = new ArrayList<>(this.queue.size());
+			this.queue.drainTo(batchEntries);
+
+			for (BatchEntry batchEntry : batchEntries) {
+				String dbName = batchEntry.getDb();
+				if (!databaseToBatchPoints.containsKey(dbName)) {
+					BatchPoints batchPoints = BatchPoints.database(dbName).retentionPolicy(batchEntry.getRp()).build();
+					databaseToBatchPoints.put(dbName, batchPoints);
+				}
+				Point point = batchEntry.getPoint();
+				databaseToBatchPoints.get(dbName).point(point);
 			}
-			Point point = batchEntry.getPoint();
-			databaseToBatchPoints.get(dbName).point(point);
-		}
 
-		for (BatchPoints batchPoints : databaseToBatchPoints.values()) {
-			BatchProcessor.this.influxDB.write(batchPoints);
+			for (BatchPoints batchPoints : databaseToBatchPoints.values()) {
+				BatchProcessor.this.influxDB.write(batchPoints);
+			}
+		} catch (Throwable t) {
+			// any exception would stop the scheduler
+			logger.log(Level.SEVERE, "Batch could not be sent. Data will be lost", t);
 		}
 	}
 

src/test/java/org/influxdb/impl/BatchProcessorTest.java

@@ -0,0 +1,41 @@
+package org.influxdb.impl;
+
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.util.concurrent.TimeUnit;
+
+import org.influxdb.InfluxDB;
+import org.influxdb.dto.BatchPoints;
+import org.influxdb.dto.Point;
+import org.testng.annotations.Test;
+
+public class BatchProcessorTest {
+
+    @Test
+    public void testSchedulerExceptionHandling() throws InterruptedException {
+        InfluxDB mockInfluxDB = mock(InfluxDBImpl.class);
+        BatchProcessor batchProcessor = BatchProcessor.builder(mockInfluxDB).actions(Integer.MAX_VALUE)
+            .interval(1, TimeUnit.NANOSECONDS).build();
+
+        doThrow(new RuntimeException()).when(mockInfluxDB).write(any(BatchPoints.class));
+
+        Point point = Point.measurement("cpu").field("6", "").build();
+        BatchProcessor.BatchEntry batchEntry1 = new BatchProcessor.BatchEntry(point, "db1", "");
+        BatchProcessor.BatchEntry batchEntry2 = new BatchProcessor.BatchEntry(point, "db2", "");
+
+        batchProcessor.put(batchEntry1);
+        Thread.sleep(200); // wait for scheduler
+
+        // first try throws an exception
+        verify(mockInfluxDB, times(1)).write(any(BatchPoints.class));
+
+        batchProcessor.put(batchEntry2);
+        Thread.sleep(200); // wait for scheduler
+        // without try catch the 2nd time does not occur
+        verify(mockInfluxDB, times(2)).write(any(BatchPoints.class));
+    }
+}