[java-firestore] BulkWriter never sends the first batch when maxOpsPerSecond is below 20, and spins its executor #14632

package repro;

import com.google.api.core.ApiFuture;
import com.google.cloud.firestore.BulkWriter;
import com.google.cloud.firestore.BulkWriterOptions;
import com.google.cloud.firestore.Firestore;
import com.google.cloud.firestore.FirestoreOptions;
import com.google.cloud.firestore.WriteResult;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Run with FIRESTORE_EMULATOR_HOST set. Arguments: maxOpsPerSecond (default 10) and the number of
 * writes (default 20).
 */
public class RateBelowBatchSizeNeverSends {
  public static void main(String[] args) throws Exception {
    int maxOpsPerSecond = args.length > 0 ? Integer.parseInt(args[0]) : 10;
    int writes = args.length > 1 ? Integer.parseInt(args[1]) : 20;
    Firestore db = FirestoreOptions.newBuilder().setProjectId("demo").build().getService();
    String collection = "rate-" + System.nanoTime();

    CountingExecutor executor = new CountingExecutor();
    BulkWriter bulkWriter =
        db.bulkWriter(
            BulkWriterOptions.builder()
                .setMaxOpsPerSecond(maxOpsPerSecond)
                .setExecutor(executor)
                .build());
    List<ApiFuture<WriteResult>> futures = new ArrayList<>();
    for (int i = 0; i < writes; i++) {
      futures.add(bulkWriter.set(db.document(collection + "/d" + i), Map.of("v", 1)));
    }
    bulkWriter.flush();

    try {
      futures.get(0).get(10, TimeUnit.SECONDS);
      System.out.println("first write completed");
    } catch (TimeoutException e) {
      System.out.println("first write did not complete within 10 s");
    }
    System.out.println("tasks scheduled on the BulkWriter executor: " + executor.scheduled.get());
    System.out.println(
        "documents written: " + db.collection(collection).get().get().size() + " of " + writes);
    System.exit(0);
  }

  /** Counts every task scheduled on it, to show the retry loop. */
  static final class CountingExecutor extends ScheduledThreadPoolExecutor {
    final AtomicLong scheduled = new AtomicLong();

    CountingExecutor() {
      super(1);
    }

    @Override
    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
      scheduled.incrementAndGet();
      return super.schedule(command, delay, unit);
    }

    @Override
    public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
      scheduled.incrementAndGet();
      return super.schedule(callable, delay, unit);
    }
  }
}