(fix) Future queries do not trigger flush on BatchedPstmtHolder for #3319

This change is that findFutureCount, findFutureList, findFutureIds do not trigger a flush on BatchedPstmtHolder.

This is to address the possible ConcurrentModificationException that could occur at BatchedPstmtHolder.closeStatements(BatchedPstmtHolder.java:153)

ebean-core/src/main/java/io/ebeaninternal/api/SpiQuery.java

@@ -654,6 +654,16 @@ public interface SpiQuery<T> extends Query<T>, SpiQueryFetch, TxnProfileEventCod
    */
   ObjectGraphNode parentNode();
 
+  /**
+   * Set that this is a future query that will execute in the background.
+   */
+  void usingFuture();
+
+  /**
+   * Return true if this is a future query.
+   */
+  boolean isUsingFuture();
+
   /**
    * Return false when this is a lazy load or refresh query for a bean.
    * <p>

ebean-core/src/main/java/io/ebeaninternal/server/core/DefaultServer.java

@@ -1273,6 +1273,7 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
   @Override
   public <T> FutureRowCount<T> findFutureCount(SpiQuery<T> query) {
     SpiQuery<T> copy = query.copy();
+    copy.usingFuture();
     boolean createdTransaction = false;
     SpiTransaction transaction = query.transaction();
     if (transaction == null) {
@@ -1291,6 +1292,7 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
   @Override
   public <T> FutureIds<T> findFutureIds(SpiQuery<T> query) {
     SpiQuery<T> copy = query.copy();
+    copy.usingFuture();
     boolean createdTransaction = false;
     SpiTransaction transaction = query.transaction();
     if (transaction == null) {
@@ -1309,6 +1311,7 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
   @Override
   public <T> FutureList<T> findFutureList(SpiQuery<T> query) {
     SpiQuery<T> spiQuery = query.copy();
+    spiQuery.usingFuture();
     // FutureList query always run in it's own persistence content
     spiQuery.setPersistenceContext(new DefaultPersistenceContext());
     if (!spiQuery.isDisableReadAudit()) {

ebean-core/src/main/java/io/ebeaninternal/server/query/DefaultOrmQueryEngine.java

@@ -51,6 +51,10 @@ public final class DefaultOrmQueryEngine implements OrmQueryEngine {
    * Flushes the jdbc batch by default unless explicitly turned off on the transaction.
    */
   private <T> void flushJdbcBatchOnQuery(OrmQueryRequest<T> request) {
+    if (request.query().isUsingFuture()) {
+      // future queries never invoke a flush
+      return;
+    }
     SpiTransaction t = request.transaction();
     if (t.isFlushOnQuery()) {
       // before we perform a query, we need to flush any

ebean-core/src/main/java/io/ebeaninternal/server/querydefn/DefaultOrmQuery.java

@@ -58,6 +58,7 @@ public class DefaultOrmQuery<T> extends AbstractQuery implements SpiQuery<T> {
   private Type type;
   private String label;
   private Mode mode = Mode.NORMAL;
+  private boolean usingFuture;
   private Object tenantId;
   /**
    * Holds query in structured form.
@@ -971,6 +972,16 @@ public class DefaultOrmQuery<T> extends AbstractQuery implements SpiQuery<T> {
     this.mode = mode;
   }
 
+  @Override
+  public final void usingFuture() {
+    this.usingFuture = true;
+  }
+
+  @Override
+  public final boolean isUsingFuture() {
+    return usingFuture;
+  }
+
   @Override
   public final boolean isUsageProfiling() {
     return usageProfiling;

ebean-test/src/test/java/org/tests/query/TestFindFutureRowCount.java

@@ -6,11 +6,87 @@ import org.junit.jupiter.api.Test;
 import org.tests.model.basic.EBasic;
 
 import java.util.List;
+import java.util.concurrent.ExecutionException;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
 class TestFindFutureRowCount extends BaseTestCase {
 
+  @Test
+  void futureCount_doesNotTriggerFlush() throws ExecutionException, InterruptedException {
+    try (Transaction transaction = DB.beginTransaction()) {
+      transaction.setBatchMode(true);
+      transaction.setBatchSize(100);
+
+      EBasic basic = new EBasic("count_batched");
+      DB.save(basic);
+
+      FutureRowCount<EBasic> futureCount = DB.find(EBasic.class)
+        .where().eq("name", "count_batched")
+        .findFutureCount();
+
+      assertThat(futureCount.get()).isEqualTo(0);
+
+      transaction.flush();
+
+      FutureRowCount<EBasic> futureCountAfter = DB.find(EBasic.class)
+        .where().eq("name", "count_batched")
+        .findFutureCount();
+
+      assertThat(futureCountAfter.get()).isEqualTo(1);
+    }
+  }
+
+  @Test
+  void futureList_doesNotTriggerFlush() throws ExecutionException, InterruptedException {
+    try (Transaction transaction = DB.beginTransaction()) {
+      transaction.setBatchMode(true);
+      transaction.setBatchSize(100);
+
+      EBasic basic = new EBasic("list_batched");
+      DB.save(basic);
+
+      FutureList<EBasic> futureList = DB.find(EBasic.class)
+        .where().eq("name", "list_batched")
+        .findFutureList();
+
+      assertThat(futureList.get()).isEmpty();
+
+      transaction.flush();
+
+      FutureList<EBasic> futureListAfter = DB.find(EBasic.class)
+        .where().eq("name", "list_batched")
+        .findFutureList();
+
+      assertThat(futureListAfter.get()).hasSize(1);
+    }
+  }
+
+  @Test
+  void futureIds_doesNotTriggerFlush() throws ExecutionException, InterruptedException {
+    try (Transaction transaction = DB.beginTransaction()) {
+      transaction.setBatchMode(true);
+      transaction.setBatchSize(100);
+
+      EBasic basic = new EBasic("ids_batched");
+      DB.save(basic);
+
+      FutureIds<EBasic> futureIds = DB.find(EBasic.class)
+        .where().eq("name", "ids_batched")
+        .findFutureIds();
+
+      assertThat(futureIds.get()).isEmpty();
+
+      transaction.flush();
+
+      FutureIds<EBasic> futureIdsAfter = DB.find(EBasic.class)
+        .where().eq("name", "ids_batched")
+        .findFutureIds();
+
+      assertThat(futureIdsAfter.get()).hasSize(1);
+    }
+  }
+
   @Test
   void count_when_inTransaction() throws Exception {
     try (Transaction transaction = DB.beginTransaction()) {