Repository: influxdata/influxdb-java (java)

Issue:
There is a thread safety issue with InfluxDBResultMapper
When I run the `toPOJO` method from the `InfluxDBResultMapper` in a multithreaded environment, some fields of the objects returned are null in some special cases.

Then I debugged and figured out that there is some problem in the `cacheMeasurementClass` method. When the first thread puts the measurement class into the cache map `CLASS_FIELD_CACHE` and before it finish putting all the field of the measurement class  into cache, other threads would get an incomplete cache object. 

I think it‘s better to make it synchronized or move the step of putting the measurement class into the cache map to the end, at least after all the field of the measurement class cached.

Pull request:
fix issue: There is a thread safety issue with InfluxDBResultMapper #683
Fixes #683 

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 55b5e69..534e59a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,10 @@
 
 ## 2.20 [unreleased]
 
+### Features
+- Add an option in `BatchOption` to prevent `InfluxDB#write` from blocking when actions queue is exhausted. [Issue #668](https://github.com/influxdata/influxdb-java/issues/688)
+
+
 ## 2.19 [2020-05-18]
 
 ## 2.18 [2020-04-17]
diff --git a/MANUAL.md b/MANUAL.md
index 49ec521..6c569cd 100644
--- a/MANUAL.md
+++ b/MANUAL.md
@@ -101,7 +101,18 @@ It's possible to use both approaches at the same time: set a default database us
   public static final int DEFAULT_JITTER_INTERVAL_DURATION = 0;
   public static final int DEFAULT_BUFFER_LIMIT = 10000;
   public static final TimeUnit DEFAULT_PRECISION = TimeUnit.NANOSECONDS;
+  public static final boolean DEFAULT_DROP_ACTIONS_ON_QUEUE_EXHAUSTION = false;
 ```
+#### Configuring behaviour of batch writes when the action queue exhausts
+With batching enabled, the client provides two options on how to deal with **action queue** (where the points are accumulated as a batch) exhaustion.
+1. When `dropActionsOnQueueExhaustion` is `false` (default value), `InfluxDB#write` will be blocked till the space is created in the action queue.
+2. When `dropActionsOnQueueExhaustion` is `true`, new writes using `InfluxDB#write` will dropped and `droppedActionHandler` will be called.
+   Example usage:
+   ```Java
+   influxDB.enableBatch(BatchOptions.DEFAULTS.dropActionsOnQueueExhaustion(true)
+                                             .droppedActionHandler((point) -> log.error("Point dropped due to action queue exhaustion.")));
+   ```
+    
 
 #### Configuring the jitter interval for batch writes
 
diff --git a/pom.xml b/pom.xml
index 361e952..fc21660 100644
--- a/pom.xml
+++ b/pom.xml
@@ -266,7 +266,7 @@
     <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>
-      <version>3.3.3</version>
+      <version>3.4.0</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -301,12 +301,12 @@
     <dependency>
       <groupId>com.squareup.okhttp3</groupId>
       <artifactId>okhttp</artifactId>
-      <version>4.7.2</version>
+      <version>4.8.0</version>
     </dependency>
     <dependency>
       <groupId>com.squareup.okhttp3</groupId>
       <artifactId>logging-interceptor</artifactId>
-      <version>4.7.2</version>
+      <version>4.8.0</version>
     </dependency>
   </dependencies>
   <profiles>
diff --git a/src/main/java/org/influxdb/BatchOptions.java b/src/main/java/org/influxdb/BatchOptions.java
index 0bf696b..af1302a 100644
--- a/src/main/java/org/influxdb/BatchOptions.java
+++ b/src/main/java/org/influxdb/BatchOptions.java
@@ -6,6 +6,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
+import java.util.function.Consumer;
 
 /**
  * BatchOptions are used to configure batching of individual data point writes
@@ -19,6 +20,8 @@ public final class BatchOptions implements Cloneable {
   public static final int DEFAULT_JITTER_INTERVAL_DURATION = 0;
   public static final int DEFAULT_BUFFER_LIMIT = 10000;
   public static final TimeUnit DEFAULT_PRECISION = TimeUnit.NANOSECONDS;
+  public static final boolean DEFAULT_DROP_ACTIONS_ON_QUEUE_EXHAUSTION = false;
+
 
   /**
    * Default batch options. This class is immutable, each configuration
@@ -32,10 +35,14 @@ public final class BatchOptions implements Cloneable {
   private int jitterDuration = DEFAULT_JITTER_INTERVAL_DURATION;
   private int bufferLimit = DEFAULT_BUFFER_LIMIT;
   private TimeUnit precision = DEFAULT_PRECISION;
+  private boolean dropActionsOnQueueExhaustion = DEFAULT_DROP_ACTIONS_ON_QUEUE_EXHAUSTION;
+  private Consumer<Point> droppedActionHandler = (point) -> {
+  };
 
   private ThreadFactory threadFactory = Executors.defaultThreadFactory();
   BiConsumer<Iterable<Point>, Throwable> exceptionHandler = (points, throwable) -> {
   };
+
   private InfluxDB.ConsistencyLevel consistency = InfluxDB.ConsistencyLevel.ONE;
 
   private BatchOptions() {
@@ -133,6 +140,33 @@ public final class BatchOptions implements Cloneable {
     return clone;
   }
 
+  /**
+   * Set to define the behaviour when the action queue exhausts. If unspecified, will default to false which means
+   * that the {@link InfluxDB#write(Point)} will be blocked till the space in the queue is created.
+   * true means that the newer actions being written to the queue will be dropped and
+   * {@link BatchOptions#droppedActionHandler} will be called.
+   * @param dropActionsOnQueueExhaustion sets the behavior
+   * @return the BatchOptions instance to be able to use it in a fluent manner.
+   */
+  public BatchOptions dropActionsOnQueueExhaustion(final boolean dropActionsOnQueueExhaustion) {
+    BatchOptions clone = getClone();
+    clone.dropActionsOnQueueExhaustion = dropActionsOnQueueExhaustion;
+    return clone;
+  }
+
+  /**
+   * Handler to handle dropped actions due to queue actions. This is only valid when
+   * {@link BatchOptions#dropActionsOnQueueExhaustion} is set to true.
+   * @param droppedActionHandler to handle action drops on action queue exhaustion.
+   * @return the BatchOptions instance to be able to use it in a fluent manner.
+   */
+  public BatchOptions droppedActionHandler(final Consumer<Point> droppedActionHandler) {
+    BatchOptions clone = getClone();
+    clone.droppedActionHandler = droppedActionHandler;
+    return clone;
+  }
+
+
   /**
    * @return actions the number of actions to collect
    */
@@ -190,6 +224,21 @@ public final class BatchOptions implements Cloneable {
     return precision;
   }
 
+
+  /**
+   * @return a boolean determining whether to drop actions on action queue exhaustion.
+   */
+  public boolean isDropActionsOnQueueExhaustion() {
+    return dropActionsOnQueueExhaustion;
+  }
+
+  /**
+   * @return a consumer function to handle actions drops on action queue exhaustion.
+   */
+  public Consumer<Point> getDroppedActionHandler() {
+    return droppedActionHandler;
+  }
+
   private BatchOptions getClone() {
     try {
       return (BatchOptions) this.clone();
diff --git a/src/main/java/org/influxdb/impl/BatchProcessor.java b/src/main/java/org/influxdb/impl/BatchProcessor.java
index 5cf5db0..4430072 100644
--- a/src/main/java/org/influxdb/impl/BatchProcessor.java
+++ b/src/main/java/org/influxdb/impl/BatchProcessor.java
@@ -19,6 +19,7 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
+import java.util.function.Consumer;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -43,6 +44,8 @@ public final class BatchProcessor {
   private final int jitterInterval;
   private final TimeUnit precision;
   private final BatchWriter batchWriter;
+  private boolean dropActionsOnQueueExhaustion;
+  Consumer<Point> droppedActionHandler;
 
   /**
    * The Builder to create a BatchProcessor instance.
@@ -62,6 +65,8 @@ public final class BatchProcessor {
     private BiConsumer<Iterable<Point>, Throwable> exceptionHandler = (entries, throwable) -> { };
     private ConsistencyLevel consistencyLevel;
 
+    private boolean dropActionsOnQueueExhaustion;
+    private Consumer<Point> droppedActionsHandler;
     /**
      * @param threadFactory
      *            is optional.
@@ -152,6 +157,37 @@ public final class BatchProcessor {
       return this;
     }
 
+    /**
+     * To define the behaviour when the action queue exhausts. If unspecified, will default to false which means that
+     * the {@link InfluxDB#write(Point)} will be blocked till the space in the queue is created.
+     * true means that the newer actions being written to the queue will dropped and
+     * {@link BatchProcessor#droppedActionHandler} will be called.
+     *
+     * @param dropActionsOnQueueExhaustion
+     *            the dropActionsOnQueueExhaustion
+     *
+     * @return this Builder to use it fluent
+     */
+    public Builder dropActionsOnQueueExhaustion(final boolean dropActionsOnQueueExhaustion) {
+      this.dropActionsOnQueueExhaustion = dropActionsOnQueueExhaustion;
+      return this;
+    }
+
+    /**
+     * A callback to be used when an actions are dropped on action queue exhaustion.
+     *
+     * @param handler
+     *            the handler
+     *
+     * @return this Builder to use it fluent
+     */
+    public Builder droppedActionHandler(final Consumer<Point> handler) {
+      this.droppedActionsHandler = handler;
+      return this;
+    }
+
+
+
     /**
      * Consistency level for batch write.
      *
@@ -200,7 +236,7 @@ public final class BatchProcessor {
       }
       return new BatchProcessor(this.influxDB, batchWriter, this.threadFactory, this.actions, this.flushIntervalUnit,
                                 this.flushInterval, this.jitterInterval, exceptionHandler, this.consistencyLevel,
-                                this.precision);
+                                this.precision, this.dropActionsOnQueueExhaustion, this.droppedActionsHandler);
     }
   }
 
@@ -262,7 +298,8 @@ public final class BatchProcessor {
   BatchProcessor(final InfluxDB influxDB, final BatchWriter batchWriter, final ThreadFactory threadFactory,
                  final int actions, final TimeUnit flushIntervalUnit, final int flushInterval, final int jitterInterval,
                  final BiConsumer<Iterable<Point>, Throwable> exceptionHandler,
-                 final ConsistencyLevel consistencyLevel, final TimeUnit precision) {
+                 final ConsistencyLevel consistencyLevel, final TimeUnit precision,
+                 final boolean dropActionsOnQueueExhaustion, final Consumer<Point> droppedActionHandler) {
     super();
     this.influxDB = influxDB;
     this.batchWriter = batchWriter;
@@ -274,6 +311,8 @@ public final class BatchProcessor {
     this.exceptionHandler = exceptionHandler;
     this.consistencyLevel = consistencyLevel;
     this.precision = precision;
+    this.dropActionsOnQueueExhaustion = dropActionsOnQueueExhaustion;
+    this.droppedActionHandler = droppedActionHandler;
     if (actions > 1 && actions < Integer.MAX_VALUE) {
         this.queue = new LinkedBlockingQueue<>(actions);
     } else {
@@ -359,7 +398,14 @@ public final class BatchProcessor {
    */
   void put(final AbstractBatchEntry batchEntry) {
     try {
-        this.queue.put(batchEntry);
+        if (this.dropActionsOnQueueExhaustion) {
+          if (!this.queue.offer(batchEntry)) {
+            this.droppedActionHandler.accept(batchEntry.getPoint());
+            return;
+          }
+        } else {
+          this.queue.put(batchEntry);
+        }
     } catch (InterruptedException e) {
         throw new RuntimeException(e);
     }
@@ -403,4 +449,11 @@ public final class BatchProcessor {
     return batchWriter;
   }
 
+  public boolean isDropActionsOnQueueExhaustion() {
+    return dropActionsOnQueueExhaustion;
+  }
+
+  public Consumer<Point> getDroppedActionHandler() {
+    return droppedActionHandler;
+  }
 }
diff --git a/src/main/java/org/influxdb/impl/InfluxDBImpl.java b/src/main/java/org/influxdb/impl/InfluxDBImpl.java
index 6643064..904744f 100644
--- a/src/main/java/org/influxdb/impl/InfluxDBImpl.java
+++ b/src/main/java/org/influxdb/impl/InfluxDBImpl.java
@@ -310,6 +310,8 @@ public class InfluxDBImpl implements InfluxDB {
             .bufferLimit(batchOptions.getBufferLimit())
             .consistencyLevel(batchOptions.getConsistency())
             .precision(batchOptions.getPrecision())
+            .dropActionsOnQueueExhaustion(batchOptions.isDropActionsOnQueueExhaustion())
+            .droppedActionHandler(batchOptions.getDroppedActionHandler())
             .build();
     this.batchEnabled.set(true);
     return this;
@@ -343,13 +345,14 @@ public class InfluxDBImpl implements InfluxDB {
   public InfluxDB enableBatch(final int actions, final int flushDuration, final TimeUnit flushDurationTimeUnit,
                               final ThreadFactory threadFactory,
                               final BiConsumer<Iterable<Point>, Throwable> exceptionHandler) {
-    enableBatch(actions, flushDuration, 0, flushDurationTimeUnit, threadFactory, exceptionHandler);
+    enableBatch(actions, flushDuration, 0, flushDurationTimeUnit, threadFactory, exceptionHandler, false, null);
     return this;
   }
 
   private InfluxDB enableBatch(final int actions, final int flushDuration, final int jitterDuration,
                                final TimeUnit durationTimeUnit, final ThreadFactory threadFactory,
-                               final BiConsumer<Iterable<Point>, Throwable> exceptionHandler) {
+                               final BiConsumer<Iterable<Point>, Throwable> exceptionHandler,
+                               final boolean dropActionsOnQueueExhaustion, final Consumer<Point> droppedActionHandler) {
     if (this.batchEnabled.get()) {
       throw new IllegalStateException("BatchProcessing is already enabled.");
     }
@@ -360,6 +363,8 @@ public class InfluxDBImpl implements InfluxDB {
             .interval(flushDuration, jitterDuration, durationTimeUnit)
             .threadFactory(threadFactory)
             .consistencyLevel(consistency)
+            .dropActionsOnQueueExhaustion(dropActionsOnQueueExhaustion)
+            .droppedActionHandler(droppedActionHandler)
             .build();
     this.batchEnabled.set(true);
     return this;
diff --git a/src/main/java/org/influxdb/impl/InfluxDBResultMapper.java b/src/main/java/org/influxdb/impl/InfluxDBResultMapper.java
index c94d848..d6edef1 100644
--- a/src/main/java/org/influxdb/impl/InfluxDBResultMapper.java
+++ b/src/main/java/org/influxdb/impl/InfluxDBResultMapper.java
@@ -211,11 +211,7 @@ public class InfluxDBResultMapper {
       if (CLASS_FIELD_CACHE.containsKey(clazz.getName())) {
         continue;
       }
-      ConcurrentMap<String, Field> initialMap = new ConcurrentHashMap<>();
-      ConcurrentMap<String, Field> influxColumnAndFieldMap = CLASS_FIELD_CACHE.putIfAbsent(clazz.getName(), initialMap);
-      if (influxColumnAndFieldMap == null) {
-        influxColumnAndFieldMap = initialMap;
-      }
+      ConcurrentMap<String, Field> influxColumnAndFieldMap = new ConcurrentHashMap<>();
 
       Class<?> c = clazz;
       while (c != null) {
@@ -227,6 +223,7 @@ public class InfluxDBResultMapper {
         }
         c = c.getSuperclass();
       }
+      CLASS_FIELD_CACHE.putIfAbsent(clazz.getName(), influxColumnAndFieldMap);
     }
   }
 


Test patch:
diff --git a/src/test/java/org/influxdb/BatchOptionsTest.java b/src/test/java/org/influxdb/BatchOptionsTest.java
index c8933c0..351321b 100644
--- a/src/test/java/org/influxdb/BatchOptionsTest.java
+++ b/src/test/java/org/influxdb/BatchOptionsTest.java
@@ -1,16 +1,19 @@
 package org.influxdb;
 
+import okhttp3.OkHttpClient;
 import org.influxdb.InfluxDB.ConsistencyLevel;
 import org.influxdb.InfluxDBException.DatabaseNotFoundException;
 import org.influxdb.dto.BatchPoints;
 import org.influxdb.dto.Point;
 import org.influxdb.dto.Query;
 import org.influxdb.dto.QueryResult;
+import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.platform.runner.JUnitPlatform;
 import org.junit.runner.RunWith;
+import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
@@ -18,10 +21,12 @@ import static org.mockito.Mockito.*;
 
 import java.io.IOException;
 import java.text.MessageFormat;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
+import java.util.function.Consumer;
 
 
 @RunWith(JUnitPlatform.class)
@@ -50,6 +55,9 @@ public class BatchOptionsTest {
 
   @Test
   public void testParametersSet() {
+
+    
+
     BatchOptions options = BatchOptions.DEFAULTS.actions(3);
     Assertions.assertEquals(3, options.getActions());
     options=options.consistency(InfluxDB.ConsistencyLevel.ANY);
@@ -71,6 +79,13 @@ public class BatchOptionsTest {
     ThreadFactory tf=Executors.defaultThreadFactory();
     options=options.threadFactory(tf);
     Assertions.assertEquals(tf, options.getThreadFactory());
+    Assertions.assertEquals(false, options.isDropActionsOnQueueExhaustion());
+
+    options=options.dropActionsOnQueueExhaustion(true);
+    Assertions.assertEquals(true, options.isDropActionsOnQueueExhaustion());
+    Consumer<Point> droppedActionHandler = (pt) -> {};
+    options=options.droppedActionHandler(droppedActionHandler);
+    Assertions.assertEquals(droppedActionHandler, options.getDroppedActionHandler());
   }
 
   /**
@@ -114,6 +129,74 @@ public class BatchOptionsTest {
     }
   }
 
+  /**
+   * Test the implementation of {@link BatchOptions#dropActionsOnQueueExhaustion(boolean)} and {@link BatchOptions#droppedActionHandler(Consumer)} }.
+   */
+  @Test
+  public void testDropActionsOnQueueExhaustionSettings() throws InterruptedException, IOException {
+
+    /*
+      To simulate a behavior where the action if exhausted because it is not cleared by the batchProcessor(may be because the latency of writes to influx server went up),
+      will have to artificially inrease latency of http calls to influx db server, for that need to add an interceptor to the http client.
+      Thus locally creating a new influx db server instead of using the global one.
+     */
+
+    CountDownLatch countDownLatch = new CountDownLatch(1);
+    try (InfluxDB influxDBLocal = TestUtils.connectToInfluxDB(new OkHttpClient.Builder().addInterceptor(chain -> {
+      try {
+        //blocking the http call for write with countdown latch
+        if (chain.request().url().encodedPath().contains("write"))
+          countDownLatch.await(5, TimeUnit.SECONDS);
+        return chain.proceed(chain.request());
+      } catch (Exception e) {
+        e.printStackTrace();
+      }
+      return chain.proceed(chain.request());
+    }), null, InfluxDB.ResponseFormat.JSON)) {
+      String dbName = "write_unittest_" + System.currentTimeMillis();
+      try {
+
+        Consumer<Point> droppedActionHandler = mock(Consumer.class);
+        BatchOptions options = BatchOptions.DEFAULTS.actions(2).flushDuration(1000)
+                .dropActionsOnQueueExhaustion(true)
+                .droppedActionHandler(droppedActionHandler);
+
+        influxDBLocal.query(new Query("CREATE DATABASE " + dbName));
+        influxDBLocal.setDatabase(dbName);
+        influxDBLocal.enableBatch(options);
+
+
+        //write 4 points and the first two will get flushed as well as actions = 2;
+        writeSomePoints(influxDBLocal, 1, 2);
+        //wait for the BatchProcessor$write() to flush the queue.
+        Thread.sleep(200);
+
+        //4 more points, last 2 of these should get dropped
+        writeSomePoints(influxDBLocal, 3, 6);
+        verify(droppedActionHandler, times(2)).accept(any());
+
+        //releasing the latch
+        countDownLatch.countDown();
+
+        //wait for the point to get written to influx db.
+        Thread.sleep(200);
+
+        QueryResult result = influxDBLocal.query(new Query("select * from weather", dbName));
+        //assert that 2 points were dropped
+        Assertions.assertEquals(4, result.getResults().get(0).getSeries().get(0).getValues().size());
+        //assert that that last 2 points were dropped.
+        org.assertj.core.api.Assertions.assertThat(result.getResults().get(0).getSeries().get(0).getValues()).extracting(a -> a.get(2))
+                .containsOnly(1.0, 2.0, 3.0, 4.0);
+
+      }
+      finally {
+        influxDBLocal.query(new Query("DROP DATABASE " + dbName));
+      }
+
+    }
+
+  }
+
   /**
    * Test the implementation of {@link BatchOptions#flushDuration(int)} }.
    * @throws InterruptedException
diff --git a/src/test/java/org/influxdb/impl/BatchProcessorTest.java b/src/test/java/org/influxdb/impl/BatchProcessorTest.java
index 7824e7d..213f6d3 100644
--- a/src/test/java/org/influxdb/impl/BatchProcessorTest.java
+++ b/src/test/java/org/influxdb/impl/BatchProcessorTest.java
@@ -8,12 +8,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.atLeastOnce;
-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 static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.*;
 import static org.mockito.hamcrest.MockitoHamcrest.argThat;
 
 import java.io.IOException;
@@ -170,6 +165,23 @@ public class BatchProcessorTest {
         assertThat(batchProcessor.getConsistencyLevel(), is(equalTo(InfluxDB.ConsistencyLevel.ANY)));
     }
 
+    @Test
+    public void testDropOnActionQueueExhaustionDefault()  {
+        InfluxDB mockInfluxDB = mock(InfluxDBImpl.class);
+        BatchProcessor batchProcessor = BatchProcessor.builder(mockInfluxDB).actions(Integer.MAX_VALUE)
+                .interval(1, TimeUnit.NANOSECONDS).build();
+        Assertions.assertEquals(false, batchProcessor.isDropActionsOnQueueExhaustion());
+    }
+
+
+    @Test
+    public void testDropOnActionQueueExhaustionUpdated()  {
+        InfluxDB mockInfluxDB = mock(InfluxDBImpl.class);
+        BatchProcessor batchProcessor = BatchProcessor.builder(mockInfluxDB).actions(Integer.MAX_VALUE)
+                .interval(1, TimeUnit.NANOSECONDS).dropActionsOnQueueExhaustion(true).build();
+        Assertions.assertEquals(true, batchProcessor.isDropActionsOnQueueExhaustion());
+    }
+
     @Test
     @SuppressWarnings("unchecked")
     public void precision() throws Exception {