feat: new WriteOption config for capturing backpressure data

Author: Fred [C] Park

README.md

@@ -418,6 +418,8 @@ If you have Docker running, but it is not available over localhost (e.g. you are
 - `INFLUXDB_PORT_API`
 - `INFLUXDB_2_IP`
 - `INFLUXDB_2_PORT_API`
+- `INFLUXDB_2_ONBOARDING_IP`
+- `INFLUXDB_2_ONBOARDING_PORT`
 
 ```bash
 $ export INFLUXDB_IP=192.168.99.100

client/README.md

@@ -611,6 +611,10 @@ The writes are processed in batches which are configurable by `WriteOptions`:
 | **exponentialBase** | the base for the exponential retry delay, the next delay is computed using random exponential backoff as a random value within the interval  ``retryInterval * exponentialBase^(attempts-1)`` and ``retryInterval * exponentialBase^(attempts)``. Example for ``retryInterval=5_000, exponentialBase=2, maxRetryDelay=125_000, total=5`` Retry delays are random distributed values within the ranges of ``[5_000-10_000, 10_000-20_000, 20_000-40_000, 40_000-80_000, 80_000-125_000]``
 | **bufferLimit** | the maximum number of unwritten stored points | 10000 |
 | **backpressureStrategy** | the strategy to deal with buffer overflow | DROP_OLDEST |
+| **captureBackpressureData** | whether to capture affected data points in backpressure events | false |
+| **concatMapPrefetch** | the number of upstream items to prefetch for the concatMapMaybe operator | 2 |
+
+There is also a synchronous blocking version of `WriteApi` - [WriteApiBlocking](#writing-data-using-synchronous-blocking-api).
 
 #### Backpressure
 The backpressure presents the problem of what to do with a growing backlog of unconsumed data points. 
@@ -640,7 +644,41 @@ writeApi.listenEvents(BackpressureEvent.class, value -> {
 });
 ```
 
-There is also a synchronous blocking version of `WriteApi` - [WriteApiBlocking](#writing-data-using-synchronous-blocking-api).
+##### Backpressure Event Data Snapshots
+
+When backpressure occurs, enable `captureBackpressureData` to capture a snapshot of the affected data points from the `BackpressureEvent`. The content of this snapshot depends on the configured backpressure strategy:
+
+- **`DROP_OLDEST`**: The snapshot contains only the data points that will be dropped (the oldest points in the buffer). This allows you to log, persist, or handle the specific data that is being lost due to backpressure.
+
+- **`DROP_LATEST`**: The snapshot contains only the newest data points that are being added to the buffer. This represents the most recent data that triggered the backpressure condition.
+
+Logging dropped data points:
+```java
+WriteOptions writeOptions = WriteOptions.builder()
+    .backpressureStrategy(BackpressureOverflowStrategy.DROP_OLDEST)
+    .bufferLimit(1000)
+    .captureBackpressureData(true)
+    .build();
+
+WriteApi writeApi = influxDBClient.getWriteApi(writeOptions);
+
+writeApi.listenEvents(BackpressureEvent.class, backpressureEvent -> {
+    List<String> affectedPoints = backpressureEvent.getBufferedLineProtocol();
+
+    if (backpressureEvent.getReason() == BackpressureEvent.BackpressureReason.TOO_MUCH_BATCHES) {
+        logger.warn("Backpressure occurred. Affected {} data points:", affectedPoints.size());
+
+        // For DROP_OLDEST: these are the points that were dropped from the buffer
+        // For DROP_LATEST: these are the newest points that triggered the condition
+        affectedPoints.forEach(point -> logger.debug("Affected point: {}", point));
+
+        // Do something with affected points ie. requeue for retry
+        requeue(affectedPoints);
+    }
+});
+```
+
+Note: Disabling `captureBackpressureData` can improve performance when backpressure data capture is not needed.
 
 #### Writing data
 

client/src/main/java/com/influxdb/client/WriteOptions.java

@@ -43,6 +43,8 @@
  * <li>retryInterval = 5000 ms</li>
  * <li>jitterInterval = 0</li>
  * <li>bufferLimit = 10_000</li>
+ * <li>concatMapPrefetch = 2</li>
+ * <li>captureBackpressureData = false</li>
  * </ul>
  * <p>
  * The default backpressure strategy is {@link BackpressureOverflowStrategy#DROP_OLDEST}.
@@ -62,6 +64,8 @@ public final class WriteOptions implements WriteApi.RetryOptions {
     public static final int DEFAULT_MAX_RETRY_TIME = 180_000;
     public static final int DEFAULT_EXPONENTIAL_BASE = 2;
     public static final int DEFAULT_BUFFER_LIMIT = 10000;
+    public static final int DEFAULT_CONCAT_MAP_PREFETCH = 2;
+    public static final boolean DEFAULT_CAPTURE_BACKPRESSURE_DATA = false;
 
     /**
      * Default configuration with values that are consistent with Telegraf.
@@ -77,8 +81,10 @@ public final class WriteOptions implements WriteApi.RetryOptions {
     private final int maxRetryTime;
     private final int exponentialBase;
     private final int bufferLimit;
+    private final int concatMapPrefetch;
     private final Scheduler writeScheduler;
     private final BackpressureOverflowStrategy backpressureStrategy;
+    private final boolean captureBackpressureData;
 
     /**
      * @return the number of data point to collect in batch
@@ -171,6 +177,17 @@ public int getBufferLimit() {
         return bufferLimit;
     }
 
+    /**
+     * The number of upstream items to prefetch so that fresh items are ready to be mapped when a previous
+     * MaybeSource terminates.
+     *
+     * @return the prefetch value for concatMapMaybe operator
+     * @see WriteOptions.Builder#concatMapPrefetch(int)
+     */
+    public int getConcatMapPrefetch() {
+        return concatMapPrefetch;
+    }
+
     /**
      * @return The scheduler which is used for write data points.
      * @see WriteOptions.Builder#writeScheduler(Scheduler)
@@ -189,6 +206,14 @@ public BackpressureOverflowStrategy getBackpressureStrategy() {
         return backpressureStrategy;
     }
 
+    /**
+     * @return whether to capture affected data points in backpressure events
+     * @see WriteOptions.Builder#captureBackpressureData(boolean)
+     */
+    public boolean getCaptureBackpressureData() {
+        return captureBackpressureData;
+    }
+
     private WriteOptions(@Nonnull final Builder builder) {
 
         Arguments.checkNotNull(builder, "WriteOptions.Builder");
@@ -202,8 +227,10 @@ private WriteOptions(@Nonnull final Builder builder) {
         maxRetryTime = builder.maxRetryTime;
         exponentialBase = builder.exponentialBase;
         bufferLimit = builder.bufferLimit;
+        concatMapPrefetch = builder.concatMapPrefetch;
         writeScheduler = builder.writeScheduler;
         backpressureStrategy = builder.backpressureStrategy;
+        captureBackpressureData = builder.captureBackpressureData;
     }
 
     /**
@@ -231,8 +258,10 @@ public static class Builder {
         private int maxRetryTime = DEFAULT_MAX_RETRY_TIME;
         private int exponentialBase = DEFAULT_EXPONENTIAL_BASE;
         private int bufferLimit = DEFAULT_BUFFER_LIMIT;
+        private int concatMapPrefetch = DEFAULT_CONCAT_MAP_PREFETCH;
         private Scheduler writeScheduler = Schedulers.newThread();
         private BackpressureOverflowStrategy backpressureStrategy = BackpressureOverflowStrategy.DROP_OLDEST;
+        private boolean captureBackpressureData = DEFAULT_CAPTURE_BACKPRESSURE_DATA;
 
         /**
          * Set the number of data point to collect in batch.
@@ -339,7 +368,9 @@ public Builder maxRetryTime(final int maxRetryTime) {
          */
         @Nonnull
         public Builder exponentialBase(final int exponentialBase) {
-            Arguments.checkPositiveNumber(exponentialBase, "exponentialBase");
+            if (exponentialBase < 2) {
+                throw new IllegalArgumentException("Expecting a number >= 2 for exponentialBase");
+            }
             this.exponentialBase = exponentialBase;
             return this;
         }
@@ -354,11 +385,27 @@ public Builder exponentialBase(final int exponentialBase) {
          */
         @Nonnull
         public Builder bufferLimit(final int bufferLimit) {
-            Arguments.checkNotNegativeNumber(bufferLimit, "bufferLimit");
+            Arguments.checkPositiveNumber(bufferLimit, "bufferLimit");
             this.bufferLimit = bufferLimit;
             return this;
         }
 
+        /**
+         * Set the prefetch value for the concatMapMaybe operator that processes write batches.
+         *
+         * The number of upstream items to prefetch so that fresh items are ready to be mapped when a previous
+         * MaybeSource terminates.
+         *
+         * @param concatMapPrefetch the prefetch value for concatMapMaybe operator (must be positive)
+         * @return {@code this}
+         */
+        @Nonnull
+        public Builder concatMapPrefetch(final int concatMapPrefetch) {
+            Arguments.checkPositiveNumber(concatMapPrefetch, "concatMapPrefetch");
+            this.concatMapPrefetch = concatMapPrefetch;
+            return this;
+        }
+
         /**
          * Set the scheduler which is used for write data points. It is useful for disabling batch writes or
          * for tuning the performance. Default value is {@link Schedulers#newThread()}.
@@ -389,6 +436,25 @@ public Builder backpressureStrategy(@Nonnull final BackpressureOverflowStrategy
             return this;
         }
 
+        /**
+         * Set whether to capture affected data points in backpressure events.
+         *
+         * When enabled, BackpressureEvent will include the specific line protocol points
+         * that are affected by the backpressure condition:
+         * - For DROP_OLDEST strategy: points that will be dropped
+         * - For DROP_LATEST strategy: newest points being added
+         *
+         * Disabling this can improve performance when backpressure data capture is not needed.
+         *
+         * @param captureBackpressureData whether to capture affected data points. Default is false.
+         * @return {@code this}
+         */
+        @Nonnull
+        public Builder captureBackpressureData(final boolean captureBackpressureData) {
+            this.captureBackpressureData = captureBackpressureData;
+            return this;
+        }
+
         /**
          * Build an instance of WriteOptions.
          *

client/src/main/java/com/influxdb/client/internal/AbstractWriteClient.java

@@ -161,12 +161,15 @@ public AbstractWriteClient(@Nonnull final WriteOptions writeOptions,
                 //
                 .lift(new BackpressureBatchesBufferStrategy(
                         writeOptions.getBufferLimit(),
-                        () -> publish(new BackpressureEvent(BackpressureEvent.BackpressureReason.TOO_MUCH_BATCHES)),
-                        writeOptions.getBackpressureStrategy()))
+                        bufferedPoints -> publish(new BackpressureEvent(
+                                BackpressureEvent.BackpressureReason.TOO_MUCH_BATCHES, bufferedPoints)),
+                        writeOptions.getBackpressureStrategy(),
+                        writeOptions.getCaptureBackpressureData()))
                 //
-                // Use concat to process batches one by one
+                // Use concat to process batches with configurable prefetch
                 //
-                .concatMapMaybe(new ToWritePointsMaybe(processorScheduler, writeOptions))
+                .concatMapMaybe(new ToWritePointsMaybe(processorScheduler, writeOptions),
+                        writeOptions.getConcatMapPrefetch())
                 .doFinally(() -> finished.set(true))
                 .subscribe(responseNotification -> {
 

client/src/main/java/com/influxdb/client/internal/flowable/BackpressureBatchesBufferStrategy.java

@@ -1,9 +1,14 @@
 package com.influxdb.client.internal.flowable;
 
+import java.util.Collections;
 import java.util.ArrayDeque;
+import java.util.Arrays;
 import java.util.Deque;
+import java.util.List;
+import java.util.stream.Collectors;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
 
 import com.influxdb.client.internal.AbstractWriteClient;
 
@@ -13,7 +18,6 @@
 import io.reactivex.rxjava3.core.FlowableSubscriber;
 import io.reactivex.rxjava3.exceptions.Exceptions;
 import io.reactivex.rxjava3.exceptions.MissingBackpressureException;
-import io.reactivex.rxjava3.functions.Action;
 import io.reactivex.rxjava3.internal.operators.flowable.FlowableOnBackpressureBufferStrategy;
 import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.rxjava3.internal.util.BackpressureHelper;
@@ -34,21 +38,32 @@ public final class BackpressureBatchesBufferStrategy implements
 
     final long bufferSize;
 
-    final Action onOverflow;
+    final Consumer<List<String>> onOverflow;
 
     final BackpressureOverflowStrategy strategy;
 
-    public BackpressureBatchesBufferStrategy(long bufferSize, Action onOverflow,
-                                             BackpressureOverflowStrategy strategy) {
+    final boolean captureBackpressureData;
+
+    public BackpressureBatchesBufferStrategy(long bufferSize,
+            Consumer<List<String>> onOverflow,
+            BackpressureOverflowStrategy strategy) {
+        this(bufferSize, onOverflow, strategy, false);
+    }
+
+    public BackpressureBatchesBufferStrategy(long bufferSize,
+            Consumer<List<String>> onOverflow,
+            BackpressureOverflowStrategy strategy,
+            boolean captureBackpressureData) {
         this.bufferSize = bufferSize;
         this.onOverflow = onOverflow;
         this.strategy = strategy;
+        this.captureBackpressureData = captureBackpressureData;
     }
 
     @Override
     public @NonNull Subscriber<? super AbstractWriteClient.BatchWriteItem> apply(
             @NonNull final Subscriber<? super AbstractWriteClient.BatchWriteItem> subscriber) throws Throwable {
-        return new OnBackpressureBufferStrategySubscriber(subscriber, onOverflow, strategy, bufferSize);
+        return new OnBackpressureBufferStrategySubscriber(subscriber, onOverflow, strategy, bufferSize, captureBackpressureData);
     }
 
     static final class OnBackpressureBufferStrategySubscriber
@@ -59,8 +74,6 @@ static final class OnBackpressureBufferStrategySubscriber
 
         final Subscriber<? super AbstractWriteClient.BatchWriteItem> downstream;
 
-        final Action onOverflow;
-
         final BackpressureOverflowStrategy strategy;
 
         final long bufferSize;
@@ -76,14 +89,20 @@ static final class OnBackpressureBufferStrategySubscriber
         volatile boolean done;
         Throwable error;
 
+        final Consumer<List<String>> onOverflow;
+
+        final boolean captureBackpressureData;
+
         OnBackpressureBufferStrategySubscriber(Subscriber<? super AbstractWriteClient.BatchWriteItem> actual,
-                                               Action onOverflow,
+                                               Consumer<List<String>> onOverflow,
                                                BackpressureOverflowStrategy strategy,
-                                               long bufferSize) {
+                                               long bufferSize,
+                                               boolean captureBackpressureData) {
             this.downstream = actual;
             this.onOverflow = onOverflow;
             this.strategy = strategy;
             this.bufferSize = bufferSize;
+            this.captureBackpressureData = captureBackpressureData;
             this.requested = new AtomicLong();
             this.deque = new ArrayDeque<>();
         }
@@ -107,18 +126,25 @@ public void onNext(AbstractWriteClient.BatchWriteItem t) {
             boolean callOnOverflow = false;
             boolean callError = false;
             Deque<AbstractWriteClient.BatchWriteItem> dq = deque;
+            List<String> overflowSnapshot = null;
             synchronized (dq) {
                 AtomicLong size = new AtomicLong(t.length());
                 dq.forEach(batchWriteItem -> size.addAndGet(batchWriteItem.length()));
                 if (size.get() > bufferSize) {
                     switch (strategy) {
                         case DROP_LATEST:
+                            if (captureBackpressureData) {
+                                overflowSnapshot = captureBatch(t);
+                            }
                             dq.pollLast();
                             dq.offer(t);
                             callOnOverflow = true;
                             break;
                         case DROP_OLDEST:
-                            dq.poll();
+                            AbstractWriteClient.BatchWriteItem droppedBatch = dq.poll();
+                            if (captureBackpressureData && droppedBatch != null) {
+                                overflowSnapshot = captureBatch(droppedBatch);
+                            }
                             dq.offer(t);
                             callOnOverflow = true;
                             break;
@@ -135,7 +161,13 @@ public void onNext(AbstractWriteClient.BatchWriteItem t) {
             if (callOnOverflow) {
                 if (onOverflow != null) {
                     try {
-                        onOverflow.run();
+                        List<String> bufferedPoints;
+                        if (captureBackpressureData && overflowSnapshot != null) {
+                            bufferedPoints = overflowSnapshot;
+                        } else {
+                            bufferedPoints = Collections.emptyList();
+                        }
+                        onOverflow.accept(bufferedPoints);
                     } catch (Throwable ex) {
                         Exceptions.throwIfFatal(ex);
                         upstream.cancel();
@@ -150,6 +182,24 @@ public void onNext(AbstractWriteClient.BatchWriteItem t) {
             }
         }
 
+        /**
+         * Captures snapshot of a single batch item for overflow handling.
+         *
+         * @param item the batch item to capture
+         * @return list of line protocol points from the item
+         */
+        List<String> captureBatch(AbstractWriteClient.BatchWriteItem item) {
+            String lp = item.toLineProtocol();
+            if (lp == null || lp.isEmpty()) {
+                return Collections.emptyList();
+            }
+            
+            return Arrays.stream(lp.split("\n"))
+                .map(String::trim)
+                .filter(s -> !s.isEmpty())
+                .collect(Collectors.toList());
+        }
+
         @Override
         public void onError(Throwable t) {
             if (done) {