coverage tests

Author: Fred [C] Park

client/src/test/java/com/influxdb/client/internal/flowable/BackpressureDataCaptureTest.java

@@ -25,6 +25,7 @@
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
 
 import com.influxdb.client.domain.WritePrecision;
 import com.influxdb.client.internal.AbstractWriteClient;
@@ -57,7 +58,7 @@ public void testBackwardCompatibility() {
                                         BackpressureOverflowStrategy.DROP_OLDEST)))
                 .subscribe(testSubscriber);
 
-        testSubscriber.request(10);
+        testSubscriber.request(1);
         testSubscriber.awaitDone(5, TimeUnit.SECONDS).assertNoErrors();
 
         Assertions.assertThat(backpressureCount.get()).isGreaterThan(0);
@@ -106,7 +107,7 @@ public void testDropOldestVsDropLatest_CompareStrategies() {
                                         true)))
                 .subscribe(oldestSubscriber);
 
-        oldestSubscriber.request(1); // Only process 1 batch, causing backpressure
+        oldestSubscriber.request(1);
         oldestSubscriber.awaitDone(5, TimeUnit.SECONDS).assertNoErrors();
 
         // Test DROP_LATEST  
@@ -124,7 +125,7 @@ public void testDropOldestVsDropLatest_CompareStrategies() {
                                         true)))
                 .subscribe(latestSubscriber);
 
-        latestSubscriber.request(1); // Only process 1 batch, causing backpressure
+        latestSubscriber.request(1);
         latestSubscriber.awaitDone(5, TimeUnit.SECONDS).assertNoErrors();
 
         // Extract and verify the exact batch numbers captured
@@ -172,6 +173,68 @@ public void testDropOldestVsDropLatest_CompareStrategies() {
         }
     }
 
+    @Test
+    public void testCaptureBatchReturnsEmptyList() {
+        int bufferSize = 1;
+        List<String> capturedData = new ArrayList<>();
+
+        TestSubscriber<AbstractWriteClient.BatchWriteItem> testSubscriber = createTestSubscriber();
+
+        Flowable<AbstractWriteClient.BatchWriteItem> emptyBatchFlow = Flowable.unsafeCreate(s -> {
+            BooleanSubscription bs = new BooleanSubscription();
+            s.onSubscribe(bs);
+
+            WriteParameters writeParameters = new WriteParameters("test-bucket", "test-org", WritePrecision.NS);
+            AbstractWriteClient.BatchWriteDataGrouped emptyData = new AbstractWriteClient.BatchWriteDataGrouped(writeParameters);
+            AbstractWriteClient.BatchWriteItem emptyBatch = new AbstractWriteClient.BatchWriteItem(writeParameters, emptyData);
+
+            // Send multiple batches to trigger overflow
+            s.onNext(emptyBatch);
+            s.onNext(emptyBatch);
+
+            if (!bs.isCancelled()) {
+                s.onComplete();
+            }
+        });
+
+        Flowable.fromPublisher(
+                        emptyBatchFlow
+                                .lift(new BackpressureBatchesBufferStrategy(
+                                        bufferSize,
+                                        capturedData::addAll,
+                                        BackpressureOverflowStrategy.DROP_OLDEST,
+                                        true))) // capture enabled
+                .subscribe(testSubscriber);
+
+        testSubscriber.request(1);
+        testSubscriber.awaitDone(5, TimeUnit.SECONDS).assertNoErrors();
+
+        Assertions.assertThat(capturedData).isEmpty();
+    }
+
+    @Test
+    public void testOnOverflowExceptionHandling() {
+        int bufferSize = 1;
+        TestSubscriber<AbstractWriteClient.BatchWriteItem> testSubscriber = createTestSubscriber();
+
+        Consumer<List<String>> throwingConsumer = capturedData -> {
+            throw new RuntimeException("Test exception in onOverflow");
+        };
+        Flowable.fromPublisher(
+                        createSequentialBatches(3)
+                                .lift(new BackpressureBatchesBufferStrategy(
+                                        bufferSize,
+                                        throwingConsumer,
+                                        BackpressureOverflowStrategy.DROP_OLDEST,
+                                        true)))
+                .subscribe(testSubscriber);
+
+        testSubscriber.request(1);
+        testSubscriber.awaitDone(5, TimeUnit.SECONDS);
+
+        testSubscriber.assertError(RuntimeException.class);
+    }
+
     /**
      * Extract batch numbers from line protocol strings for verification
      */
@@ -180,11 +243,11 @@ private List<Integer> extractBatchNumbers(List<String> lineProtocolPoints) {
                 .map(line -> {
                     try {
                         int batchStart = line.indexOf("batch=") + 6;
-                        if (batchStart == 5) return -1; // "batch=" not found
-                        
+                        if (batchStart == 5) return -1;
+
                         int batchEnd = line.indexOf(" ", batchStart);
                         if (batchEnd == -1) batchEnd = line.length();
-                        
+
                         String batchNumStr = line.substring(batchStart, batchEnd);
                         return Integer.parseInt(batchNumStr);
                     } catch (Exception e) {
@@ -217,16 +280,14 @@ private Flowable<AbstractWriteClient.BatchWriteItem> createSequentialBatches(int
         return Flowable.unsafeCreate(s -> {
             BooleanSubscription bs = new BooleanSubscription();
             s.onSubscribe(bs);
-            
+
             for (int batchNum = 1; batchNum <= batchCount && !bs.isCancelled(); batchNum++) {
                 WriteParameters writeParameters = new WriteParameters("test-bucket", "test-org", WritePrecision.NS);
                 AbstractWriteClient.BatchWriteDataGrouped data = new AbstractWriteClient.BatchWriteDataGrouped(writeParameters);
-                
+
                 // Single point per batch with clear batch identification
                 data.append(String.format("sequential,batch=%d value=%d %d", batchNum, batchNum * 100, batchNum * 1000000L));
-                
                 s.onNext(new AbstractWriteClient.BatchWriteItem(writeParameters, data));
-                
                 // Small delay to ensure predictable ordering
                 try {
                     Thread.sleep(1);