This is an automated email from the ASF dual-hosted git repository.
Abacn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 08e0efcfebf feat(firestore): add option to disable write ramp-up
throttling (#40390)
08e0efcfebf is described below
commit 08e0efcfebf1e8b7eca1a222be11bb964c4e64bb
Author: Balázs Németh <[email protected]>
AuthorDate: Thu Oct 15 16:13:46 2026 +0200
feat(firestore): add option to disable write ramp-up throttling (#40390)
Add RpcQosOptions.Builder.withRampupThrottlingDisabled(), mirroring
DatastoreV1.Write.withRampupThrottlingDisabled(). When set, writes no
longer wait for the 500/50/5 ramp-up budget and batch sizes are no longer
capped by it; adaptive throttling on failures still applies.
Addresses #27170.
Claude-Session: https://claude.ai/code/session_01BofSrVTSEGJvEzsSNXfU2o
Co-authored-by: Claude <[email protected]>
Co-authored-by: Yi Hu <[email protected]>
---
CHANGES.md | 2 +
.../beam/sdk/io/gcp/firestore/RpcQosImpl.java | 8 ++-
.../beam/sdk/io/gcp/firestore/RpcQosOptions.java | 61 ++++++++++++++++++++--
.../sdk/io/gcp/firestore/RpcQosOptionsTest.java | 16 ++++++
.../beam/sdk/io/gcp/firestore/RpcQosTest.java | 39 ++++++++++++++
5 files changed, 122 insertions(+), 4 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index c787b83dde7..bb5ab431e05 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -65,6 +65,8 @@
## I/Os
* Support for X source added (Java/Python)
([#X](https://github.com/apache/beam/issues/X)).
+* (Java) FirestoreIO: added
`RpcQosOptions.Builder.withRampupThrottlingDisabled()`, mirroring the existing
DatastoreIO option, to opt out of the 500/50/5 write ramp-up
([#27170](https://github.com/apache/beam/issues/27170)).
+
## New Features / Improvements
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java
index 06b7dbe3801..c647b3d084b 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java
@@ -376,6 +376,9 @@ final class RpcQosImpl implements RpcQos {
@Override
public boolean awaitSafeToProceed(Instant instant) throws
InterruptedException {
state.checkActive();
+ if (!options.isShouldThrottleRampup()) {
+ return super.awaitSafeToProceed(instant);
+ }
Optional<Duration> shouldThrottle = writeRampUp.shouldThrottle(instant);
if (shouldThrottle.isPresent()) {
Duration throttleDuration = shouldThrottle.get();
@@ -392,7 +395,10 @@ final class RpcQosImpl implements RpcQos {
public <ElementT extends Element<?>> FlushBufferImpl<ElementT>
newFlushBuffer(
Instant instantSinceEpoch) {
state.checkActive();
- int availableWriteCountBudget =
writeRampUp.getAvailableWriteCountBudget(instantSinceEpoch);
+ int availableWriteCountBudget =
+ options.isShouldThrottleRampup()
+ ? writeRampUp.getAvailableWriteCountBudget(instantSinceEpoch)
+ : Integer.MAX_VALUE;
int nextBatchMaxCount = wb.nextBatchMaxCount(instantSinceEpoch);
int batchMaxCount =
Ints.min(
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosOptions.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosOptions.java
index 950dbe9374b..247f1d42cb1 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosOptions.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosOptions.java
@@ -76,6 +76,7 @@ public final class RpcQosOptions implements Serializable,
HasDisplayData {
private final long batchMaxBytes;
private final Duration batchTargetLatency;
private final int hintMaxNumWorkers;
+ private final boolean shouldThrottleRampup;
private final boolean shouldReportDiagnosticMetrics;
private RpcQosOptions(
@@ -90,6 +91,7 @@ public final class RpcQosOptions implements Serializable,
HasDisplayData {
long batchMaxBytes,
Duration batchTargetLatency,
int hintMaxNumWorkers,
+ boolean shouldThrottleRampup,
boolean shouldReportDiagnosticMetrics) {
this.maxAttempts = maxAttempts;
this.initialBackoff = initialBackoff;
@@ -102,6 +104,7 @@ public final class RpcQosOptions implements Serializable,
HasDisplayData {
this.batchMaxBytes = batchMaxBytes;
this.batchTargetLatency = batchTargetLatency;
this.hintMaxNumWorkers = hintMaxNumWorkers;
+ this.shouldThrottleRampup = shouldThrottleRampup;
this.shouldReportDiagnosticMetrics = shouldReportDiagnosticMetrics;
}
@@ -125,6 +128,9 @@ public final class RpcQosOptions implements Serializable,
HasDisplayData {
.withLabel("batchTargetLatency"))
.add(
DisplayData.item("hintMaxNumWorkers",
hintMaxNumWorkers).withLabel("hintMaxNumWorkers"))
+ .add(
+ DisplayData.item("shouldThrottleRampup", shouldThrottleRampup)
+ .withLabel("shouldThrottleRampup"))
.add(
DisplayData.item("shouldReportDiagnosticMetrics",
shouldReportDiagnosticMetrics)
.withLabel("shouldReportDiagnosticMetrics"));
@@ -274,6 +280,23 @@ public final class RpcQosOptions implements Serializable,
HasDisplayData {
return hintMaxNumWorkers;
}
+ /**
+ * Whether writes are throttled to follow the <a target="_blank"
rel="noopener noreferrer"
+ *
href="https://cloud.google.com/firestore/docs/best-practices#ramping_up_traffic";>500/50/5
+ * ramp-up</a> strategy.
+ *
+ * <p>When {@code false}, the ramp-up budget no longer delays writes nor
limits the size of a
+ * batch. Writes are still subject to adaptive throttling in response to
failed requests.
+ *
+ * <p><i>Default Value:</i> {@code true}
+ *
+ * @see RpcQosOptions.Builder#withRampupThrottlingDisabled()
+ * @see #getHintMaxNumWorkers()
+ */
+ public boolean isShouldThrottleRampup() {
+ return shouldThrottleRampup;
+ }
+
/**
* Whether additional diagnostic metrics should be reported for a Transform.
*
@@ -314,11 +337,13 @@ public final class RpcQosOptions implements Serializable,
HasDisplayData {
.withBatchMaxBytes(batchMaxBytes)
.withBatchTargetLatency(batchTargetLatency)
.withHintMaxNumWorkers(hintMaxNumWorkers);
+ if (!shouldThrottleRampup) {
+ builder = builder.withRampupThrottlingDisabled();
+ }
if (shouldReportDiagnosticMetrics) {
- return builder.withReportDiagnosticMetrics();
- } else {
- return builder;
+ builder = builder.withReportDiagnosticMetrics();
}
+ return builder;
}
@Override
@@ -336,6 +361,7 @@ public final class RpcQosOptions implements Serializable,
HasDisplayData {
&& batchMaxCount == that.batchMaxCount
&& batchMaxBytes == that.batchMaxBytes
&& hintMaxNumWorkers == that.hintMaxNumWorkers
+ && shouldThrottleRampup == that.shouldThrottleRampup
&& shouldReportDiagnosticMetrics == that.shouldReportDiagnosticMetrics
&& initialBackoff.equals(that.initialBackoff)
&& samplePeriod.equals(that.samplePeriod)
@@ -358,6 +384,7 @@ public final class RpcQosOptions implements Serializable,
HasDisplayData {
batchMaxBytes,
batchTargetLatency,
hintMaxNumWorkers,
+ shouldThrottleRampup,
shouldReportDiagnosticMetrics);
}
@@ -386,6 +413,8 @@ public final class RpcQosOptions implements Serializable,
HasDisplayData {
+ batchTargetLatency
+ ", hintMaxNumWorkers="
+ hintMaxNumWorkers
+ + ", shouldThrottleRampup="
+ + shouldThrottleRampup
+ ", shouldReportDiagnosticMetrics="
+ shouldReportDiagnosticMetrics
+ '}';
@@ -444,6 +473,7 @@ public final class RpcQosOptions implements Serializable,
HasDisplayData {
private long batchMaxBytes;
private Duration batchTargetLatency;
private int hintMaxNumWorkers;
+ private boolean shouldThrottleRampup;
private boolean shouldReportDiagnosticMetrics;
private Builder() {
@@ -458,6 +488,7 @@ public final class RpcQosOptions implements Serializable,
HasDisplayData {
batchMaxBytes = FIRESTORE_RPC_BYTES_MAX;
batchTargetLatency = Duration.standardSeconds(5);
hintMaxNumWorkers = 500;
+ shouldThrottleRampup = true;
shouldReportDiagnosticMetrics = false;
}
@@ -646,6 +677,29 @@ public final class RpcQosOptions implements Serializable,
HasDisplayData {
return this;
}
+ /**
+ * Disable the <a target="_blank" rel="noopener noreferrer"
+ *
href="https://cloud.google.com/firestore/docs/best-practices#ramping_up_traffic";>500/50/5
+ * ramp-up</a> throttling of writes.
+ *
+ * <p>By default, each worker starts with a budget of {@code 500 /
hintMaxNumWorkers} writes per
+ * second, which grows by 50% every 5 minutes. With the default {@code
hintMaxNumWorkers} this
+ * is 1 write per second, which can severely limit the throughput of
pipelines running on fewer
+ * workers. Disabling ramp-up throttling is appropriate when the target
collection is already
+ * serving traffic, or its documents are written with well distributed IDs.
+ *
+ * <p>Writes remain subject to adaptive throttling in response to failed
requests.
+ *
+ * <p>This mirrors {@code
DatastoreIO.v1().write().withRampupThrottlingDisabled()}.
+ *
+ * @return this builder
+ * @see RpcQosOptions#isShouldThrottleRampup()
+ */
+ public Builder withRampupThrottlingDisabled() {
+ this.shouldThrottleRampup = false;
+ return this;
+ }
+
/**
* Whether additional diagnostic metrics should be reported for a
Transform.
*
@@ -698,6 +752,7 @@ public final class RpcQosOptions implements Serializable,
HasDisplayData {
batchMaxBytes,
batchTargetLatency,
hintMaxNumWorkers,
+ shouldThrottleRampup,
shouldReportDiagnosticMetrics);
}
diff --git
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosOptionsTest.java
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosOptionsTest.java
index 997d94b3d2b..601525c02ac 100644
---
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosOptionsTest.java
+++
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosOptionsTest.java
@@ -19,6 +19,7 @@ package org.apache.beam.sdk.io.gcp.firestore;
import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists.newArrayList;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@@ -59,6 +60,20 @@ public final class RpcQosOptionsTest {
assertEquals(newMaxAttempts, build.getMaxAttempts());
}
+ @Test
+ public void rampupThrottlingEnabledByDefault() {
+ assertTrue(RpcQosOptions.defaultOptions().isShouldThrottleRampup());
+ }
+
+ @Test
+ public void withRampupThrottlingDisabled_survivesToBuilder() {
+ RpcQosOptions disabled =
RpcQosOptions.newBuilder().withRampupThrottlingDisabled().build();
+
+ assertFalse(disabled.isShouldThrottleRampup());
+ assertNotEquals(RpcQosOptions.defaultOptions(), disabled);
+ assertEquals(disabled, disabled.toBuilder().build());
+ }
+
@Test
public void populateDisplayData() {
//noinspection unchecked
@@ -85,6 +100,7 @@ public final class RpcQosOptionsTest {
"samplePeriod",
"samplePeriodBucketSize",
"shouldReportDiagnosticMetrics",
+ "shouldThrottleRampup",
"throttleDuration");
assertEquals(expectedKeys, actualKeys);
diff --git
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosTest.java
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosTest.java
index 8910d491c4f..367677055e8 100644
---
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosTest.java
+++
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosTest.java
@@ -562,6 +562,45 @@ public final class RpcQosTest {
doTest_initialBatchSizeRelativeToWorkerCount(1, 500);
}
+ @Test
+ public void rampupThrottlingDisabled_initialBatchSizeIgnoresWorkerCount() {
+ RpcQosOptions options =
+ RpcQosOptions.newBuilder()
+ .withHintMaxNumWorkers(10000)
+ .withBatchInitialCount(500)
+ .withRampupThrottlingDisabled()
+ .build();
+ RpcQosImpl qos = new RpcQosImpl(options, random, sleeper, counterFactory,
distributionFactory);
+ RpcWriteAttemptImpl attempt = qos.newWriteAttempt(RPC_ATTEMPT_CONTEXT);
+ FlushBufferImpl<Element<Object>> buffer =
attempt.newFlushBuffer(Instant.EPOCH);
+ assertEquals(500, buffer.nextBatchMaxCount);
+ }
+
+ @Test
+ public void rampupThrottlingDisabled_writesDoNotWaitForRampUpBudget()
+ throws InterruptedException {
+ RpcQosOptions options =
+ RpcQosOptions.newBuilder()
+ .withHintMaxNumWorkers(10000)
+ .withRampupThrottlingDisabled()
+ .build();
+ RpcQos qos = new RpcQosImpl(options, random, sleeper, counterFactory,
distributionFactory);
+
+ // with ramp-up throttling enabled, the budget of 1 write per second would
be exhausted after
+ // the first successful request within the same second
+ Instant now = Instant.EPOCH;
+ for (int i = 0; i < 10; i++) {
+ RpcWriteAttempt attempt = qos.newWriteAttempt(RPC_ATTEMPT_CONTEXT);
+ assertTrue(attempt.awaitSafeToProceed(now));
+ attempt.recordRequestStart(now, 500);
+ attempt.recordWriteCounts(now, 500, 0);
+ attempt.recordRequestSuccessful(now);
+ }
+
+ verify(sleeper, times(0)).sleep(anyLong());
+ verify(counterThrottlingMs, times(0)).inc(anyLong());
+ }
+
@Test
public void isCodeRetryable() {
doTest_isCodeRetryable(Code.ABORTED, true);
