[Bug]: The default config for FirestoreV1.write().batchWrite() on smaller java jobs has an abysmal throughput #27170

Activity

1. nbali

   added

   awaiting triage

   bug

   on Jun 19, 2023
2. github-actions

   added

   java

   io

   P2

   on Jun 19, 2023
3. github-actions commented on Oct 2, 2026

   

   github-actionsbot

   on Oct 2, 2026 – with GitHub Actions

   Contributor

   More actions

   This issue has been marked as stale due to 150 days of inactivity. It will be closed in 30 days if no further activity occurs. If you think that’s incorrect or this issue still needs to be addressed, please simply write any comment. If closed, you can reopen the issue at any time. Thank you for your contributions.
4. github-actions

   added

   stale

   on Oct 2, 2026
5. nbali commented on Oct 2, 2026

   

   nbali

   on Oct 2, 2026

   ContributorAuthor

   More actions

   Still valid on current master (15a7ce1). Compared to 69cd6ef, the code linked in the original report is unchanged apart from shifted line numbers and one cast (Math.pow(1.5, growth) → Math.pow(1.5, (double) growth)) that doesn't change behavior.

   Updated permalinks:

   beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java

   Lines 122 to 123 in 15a7ce1

   	writeRampUp =
   	new WriteRampUp(500.0 / options.getHintMaxNumWorkers(), filteringDistributionFactory);

   beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosOptions.java

   Lines 258 to 275 in 15a7ce1

   	/**
   	* A hint to the QoS system for the intended max number of workers for a pipeline. The provided
   	* value can be used to try to scale calculations to values appropriate for the pipeline as a
   	* whole.
   	*
   	* <p>If you are running your pipeline on <a target="_blank" rel="noopener noreferrer"
   	* href="https://cloud.google.com/dataflow">Cloud Dataflow</a> this parameter should be set to the
   	* same value as <a target="_blank" rel="noopener noreferrer"
   	* href="https://cloud.google.com/dataflow/docs/guides/specifying-exec-params#setting-other-cloud-dataflow-pipeline-options">{@code
   	* maxNumWorkers}</a>
   	*
   	* <p><i>Default Value:</i> 500
   	*
   	* @see RpcQosOptions.Builder#withBatchMaxCount(int)
   	*/
   	public int getHintMaxNumWorkers() {
   	return hintMaxNumWorkers;
   	}

   beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java

   Lines 597 to 683 in 15a7ce1

   	static final class WriteRampUp {
   	private static final Duration RAMP_UP_INTERVAL = Duration.standardMinutes(5);
   	private final double baseBatchBudget;
   	private final long rampUpIntervalMinutes;
   	private final MovingFunction writeCounts;
   	private final LinearBackoff backoff;
   	private final Distribution throttlingMs;
   	private final Distribution availableWriteCountBudget;
   	@SuppressWarnings("OptionalUsedAsFieldOrParameterType")
   	private Optional<Instant> firstInstant = Optional.empty();
   	WriteRampUp(double baseBatchBudget, DistributionFactory distributionFactory) {
   	this.baseBatchBudget = baseBatchBudget;
   	this.rampUpIntervalMinutes = RAMP_UP_INTERVAL.getStandardMinutes();
   	this.writeCounts =
   	createMovingFunction(
   	// track up to one second of budget usage.
   	// this determines the full duration of time we want to keep track of request counts
   	Duration.standardSeconds(1),
   	// refill the budget each second
   	// this determines the sub-granularity the full duration will be broken into. So if
   	// we wanted budget to refill twice per second, this could be passed
   	// Duration.millis(500)
   	Duration.standardSeconds(1));
   	this.backoff = new LinearBackoff(Duration.standardSeconds(1));
   	this.throttlingMs =
   	distributionFactory.get(RpcQos.class.getName(), "qos_rampUp_throttlingMs");
   	this.availableWriteCountBudget =
   	distributionFactory.get(RpcQos.class.getName(), "qos_rampUp_availableWriteCountBudget");
   	}
   	int getAvailableWriteCountBudget(Instant instant) {
   	if (!firstInstant.isPresent()) {
   	firstInstant = Optional.of(instant);
   	return (int) Math.max(1, baseBatchBudget);
   	}
   	Instant first = firstInstant.get();
   	double maxRequestBudget = calcMaxRequestBudget(instant, first);
   	long writeCount = writeCounts.get(instant.getMillis());
   	double availableBudget = maxRequestBudget - writeCount;
   	int budget = Ints.saturatedCast((long) availableBudget);
   	availableWriteCountBudget.update(budget);
   	return budget;
   	}
   	/**
   	* Calculate the value relative to the growth line relative to the {@code first}.
   	*
   	* <p>For a {@code baseBatchBudget} of 500 the line would look like <a target="_blank"
   	* rel="noopener noreferrer"
   	* href="https://www.wolframalpha.com/input/?i=500+*+1.5%5Emax%280%2C+%28x-5%29%2F5%29+from+0+to+30">this</a>
   	*/
   	private double calcMaxRequestBudget(Instant instant, Instant first) {
   	Duration durationSinceFirst = new Duration(first, instant);
   	long calculatedGrowth =
   	(durationSinceFirst.getStandardMinutes() - rampUpIntervalMinutes) / rampUpIntervalMinutes;
   	long growth = Math.max(0, calculatedGrowth);
   	return baseBatchBudget * Math.pow(1.5, (double) growth);
   	}
   	void recordWriteCount(Instant instant, int numWrites) {
   	writeCounts.add(instant.getMillis(), numWrites);
   	}
   	Optional<Duration> shouldThrottle(Instant instant) {
   	int availableWriteCountBudget = getAvailableWriteCountBudget(instant);
   	if (availableWriteCountBudget <= 0) {
   	long nextBackOffMillis = backoff.nextBackOffMillis();
   	if (nextBackOffMillis > BackOff.STOP) {
   	Duration throttleDuration = Duration.millis(nextBackOffMillis);
   	throttlingMs.update(throttleDuration.getMillis());
   	return Optional.of(throttleDuration);
   	} else {
   	// we've exhausted our backoff, and have moved into the next time window try again
   	backoff.reset();
   	return Optional.empty();
   	}
   	} else {
   	// budget is available reset backoff tracking
   	backoff.reset();
   	return Optional.empty();
   	}
   	}
   	}

   beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/FirestoreV1WriteFn.java

   Lines 328 to 360 in 15a7ce1

   	private void flushBatch(boolean finishingBundle, ContextAdapter<OutT> context)
   	throws InterruptedException {
   	while (!writes.isEmpty()) {
   	RpcWriteAttempt attempt = rpcQos.newWriteAttempt(getRpcAttemptContext());
   	Instant begin = clock.instant();
   	if (!attempt.awaitSafeToProceed(begin)) {
   	continue;
   	}
   	FlushBuffer<WriteElement> flushBuffer = getFlushBuffer(attempt, begin);
   	if (flushBuffer.isFull() || (finishingBundle && flushBuffer.isNonEmpty())) {
   	DoFlushStatus flushStatus = doFlush(attempt, flushBuffer, context);
   	if (flushStatus == DoFlushStatus.ONE_OR_MORE_FAILURES && !finishingBundle) {
   	break;
   	}
   	} else {
   	// since we're not going to perform a flush, we need to return the writes that were
   	// preemptively removed
   	flushBuffer.forEach(writes::offer);
   	if (!finishingBundle) {
   	// we're not on the final flush, so yield until more elements are delivered or
   	// finishingBundle is true
   	return;
   	}
   	}
   	}
   	if (writes.isEmpty()) {
   	// now that the queue has been emptied reset our priority back to 0 to try and ensure
   	// we won't run into overflow issues if a worker runs for a long time and processes
   	// many writes.
   	queueNextEntryPriority = 0;
   	}
   	}
6. nbali commented on Oct 2, 2026

   

   nbali

   on Oct 2, 2026

   ContributorAuthor

   More actions

   Two PRs that alleviate this (they don't fix it): both copy existing DatastoreIO behavior into FirestoreIO.

   • [Java] FirestoreIO: add RpcQosOptions.withRampupThrottlingDisabled() #40390 adds RpcQosOptions.Builder.withRampupThrottlingDisabled(), mirroring DatastoreV1.Write.withRampupThrottlingDisabled(). With it, writes no longer wait for the 500/50/5 ramp-up budget and batches are no longer capped by it. Error-based throttling and retry backoff still apply.
   • [Java] FirestoreIO: ramp up writes from a pipeline-wide start time #40391 makes the ramp-up grow from one pipeline-wide start time (a singleton side input), mirroring DatastoreV1.Mutate / RampupThrottlingFn, instead of from each DoFn instance's first write. DoFn instances created later (autoscaling, retried bundles) no longer start over at 1 write/s. It's skipped when --updateCompatibilityVersion is below 2.78.0.

   Neither changes the default: without the opt-out or withHintMaxNumWorkers, a small job still starts at 500 / 500 = 1 write/s and grows 50% every 5 minutes.
7. github-actions

   removed

   stale

   on Oct 3, 2026
8. Abacn

   added a commit that references this issue on Oct 15, 2026

   feat(firestore): add option to disable write ramp-up throttling (#40390)

   ...

   08e0efc
9. Abacn

   closed this as completedin #40390on Oct 15, 2026
10. github-actions

    added this to the 2.78.0 Release milestone on Oct 15, 2026