Repository navigation
[java-firestore] BulkWriter stops sending after 500 failed writes: a failed write never releases its pending-operation slot #14630
Description
Issue Details
After 500 writes have failed on one BulkWriter instance, every later write on that instance is buffered and never sent, so its future and flush() never complete. A failed write never releases its pending-operation slot.
BulkWriter.executeWrite increments pendingOpsCount when it sends an operation and decrements it in two places: in a transformAsync on success, and in a catchingAsync(processedOperationFuture, ApiException.class, ...) on failure. A write that fails completes its future with a BulkWriterException. That class extends FirestoreException, which extends BaseGrpcServiceException, not com.google.api.gax.rpc.ApiException, so the failure handler never matches and the slot is never released. Once pendingOpsCount reaches maxPendingOpCount (500), executeWrite puts each new operation in bufferedOperations, and nothing drains that buffer again.
The count is over the instance's whole life, not concurrent operations, so a long-lived BulkWriter that sees occasional permanent failures (ALREADY_EXISTS from create(), FAILED_PRECONDITION from a precondition, NOT_FOUND from update()) eventually stops writing without any error.
Catching Throwable.class (or BulkWriterException.class) in that handler would release the slot.
Found while building an Apache Flink sink on BulkWriter (flink-gcp/flink-connector-gcp), which currently works around it.
Environment
- OS Type and Version: macOS 26 (arm64)
- Java Version and JDK Vendor: Eclipse Temurin 17.0.20
The reproducer runs against the Firestore emulator (gcr.io/google.com/cloudsdktool/google-cloud-cli:587.0.0-emulators). The defect is in client-side logic, so no GCP project is involved.
Dependencies
- Libraries-Bom: 26.90.0
- google-cloud-firestore 3.49.0
- Gax: 2.87.0
- Auth: 1.54.0
The affected code is identical in 3.49.0 and on main.
Reproducer
Run with FIRESTORE_EMULATOR_HOST pointing at the emulator, e.g. docker run -p 8080:8080 gcr.io/google.com/cloudsdktool/google-cloud-cli:587.0.0-emulators gcloud emulators firestore start --host-port=0.0.0.0:8080. The argument is the number of failed writes; 499 shows the write still completing one short of the ceiling.
package repro;
import com.google.api.core.ApiFuture;
import com.google.cloud.firestore.BulkWriter;
import com.google.cloud.firestore.DocumentReference;
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.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/** Run with FIRESTORE_EMULATOR_HOST set. Argument: the number of failed writes (default 500). */
public class FailedWritesLeakPendingSlots {
public static void main(String[] args) throws Exception {
int failures = args.length > 0 ? Integer.parseInt(args[0]) : 500;
Firestore db = FirestoreOptions.newBuilder().setProjectId("demo").build().getService();
String collection = "leak-" + System.nanoTime();
// Seed the documents so that create() fails with ALREADY_EXISTS, which is not retried.
try (BulkWriter seeding = db.bulkWriter()) {
for (int i = 0; i < failures; i++) {
seeding.set(db.document(collection + "/d" + i), Map.of("v", 1));
}
}
BulkWriter bulkWriter = db.bulkWriter();
List<ApiFuture<WriteResult>> refused = new ArrayList<>();
for (int i = 0; i < failures; i++) {
refused.add(bulkWriter.create(db.document(collection + "/d" + i), Map.of()));
}
bulkWriter.flush().get();
int failed = 0;
for (ApiFuture<WriteResult> future : refused) {
try {
future.get();
} catch (ExecutionException e) {
failed++;
}
}
System.out.println("failed writes on this BulkWriter: " + failed);
DocumentReference next = db.document(collection + "/next");
ApiFuture<WriteResult> write = bulkWriter.set(next, Map.of("v", 1));
ApiFuture<Void> flush = bulkWriter.flush();
try {
System.out.println("next write completed: " + write.get(10, TimeUnit.SECONDS));
} catch (TimeoutException e) {
System.out.println("next write did not complete within 10 s");
System.out.println("flush() done: " + flush.isDone());
System.out.println("document exists: " + next.get().get().exists());
}
try (BulkWriter fresh = db.bulkWriter()) {
ApiFuture<WriteResult> control = fresh.set(db.document(collection + "/control"), Map.of("v", 1));
fresh.flush();
System.out.println("same write on a new BulkWriter completed: " + control.get(10, TimeUnit.SECONDS));
}
System.exit(0);
}
}Logs and Stack Trace
No exception is thrown; the write is never sent.
$ java ... repro.FailedWritesLeakPendingSlots 499
failed writes on this BulkWriter: 499
next write completed: com.google.cloud.firestore.WriteResult@1f44ecf4
same write on a new BulkWriter completed: com.google.cloud.firestore.WriteResult@203d1834
$ java ... repro.FailedWritesLeakPendingSlots 500
failed writes on this BulkWriter: 500
next write did not complete within 10 s
flush() done: false
document exists: false
same write on a new BulkWriter completed: com.google.cloud.firestore.WriteResult@f5af0ea8
Behavior
Deterministic: the 500th failure on an instance is always the threshold. It does not depend on timing or data volume beyond the failure count.
Activity