From 43c9f09d16ea398716fd016014133a344938f546 Mon Sep 17 00:00:00 2001 From: Maksym Tymoshyk Date: Mon, 31 Aug 2026 21:23:41 +0300 Subject: [PATCH 1/2] [Java] Honor withMaxRetryJobs for bounded BigQueryIO file loads BigQueryIO.Write only forwarded maxRetryJobs to BatchLoads when the input was unbounded, so every batch pipeline used BatchLoads' own default of 3 retries and withMaxRetryJobs did nothing at all. A user who asked for fewer retries on a permanently failing load job, or for more, got neither and only saw "reached max retries: 3" after the write had given up. The value now reaches BatchLoads in both modes. Write.getMaxRetryJobs becomes a nullable Integer that BigQueryIO.write() leaves unset, so an absent setting can still fall back to the two different defaults the two modes have always had: 3 for bounded, 1000 for unbounded. Also documents on withMaxRetryJobs that the setting applies to FILE_LOADS only and what happens when it is not called. --- CHANGES.md | 1 + .../beam/sdk/io/gcp/bigquery/BatchLoads.java | 4 +++ .../beam/sdk/io/gcp/bigquery/BigQueryIO.java | 30 ++++++++++++----- .../io/gcp/bigquery/BigQueryIOWriteTest.java | 32 +++++++++++++++++-- 4 files changed, 57 insertions(+), 10 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 365605c7065b..78dbb7064adf 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -96,6 +96,7 @@ * (Prism) Self-checkpointing splittable DoFns now resume after their requested delay instead of immediately, so polling SDFs no longer busy-spin ([#39848](https://github.com/apache/beam/issues/39848)). * (Java) MongoDbIO read splitting now preserves non-ObjectId `_id` types (e.g. string ids) instead of failing to parse the generated range filters ([#39900](https://github.com/apache/beam/issues/39900)). * (Go) Fixed GCS glob matching silently dropping objects when the glob pattern contains multi-byte characters ([#39969](https://github.com/apache/beam/issues/39969)). +* (Java) `BigQueryIO.Write.withMaxRetryJobs` is now honored for bounded (batch) pipelines using `FILE_LOADS`, which previously always retried failed load jobs 3 times. Pipelines that never call it keep their existing defaults ([#28281](https://github.com/apache/beam/issues/28281)). ## Security Fixes diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java index 252e55d34c07..f892a5c8c16e 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java @@ -135,7 +135,11 @@ class BatchLoads // It sets to {@code Integer.MAX_VALUE} to block until the BigQuery job finishes. static final int LOAD_JOB_POLL_MAX_RETRIES = Integer.MAX_VALUE; + // The number of times a failed load or copy job is retried in place before the bundle is failed. + // A bounded pipeline keeps this low because the runner can just rerun the failed bundle; an + // unbounded pipeline retries far more, because failing a bundle in streaming is expensive. static final int DEFAULT_MAX_RETRY_JOBS = 3; + static final int DEFAULT_MAX_RETRY_JOBS_UNBOUNDED = 1000; private BigQueryServices bigQueryServices; private final WriteDisposition writeDisposition; diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java index 3549b92209b2..c7f2f8c2494f 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java @@ -2532,7 +2532,6 @@ public static Write write() { .setPropagateSuccessful(true) .setAutoSchemaUpdate(false) .setDeterministicRecordIdFn(null) - .setMaxRetryJobs(1000) .setPropagateSuccessfulStorageApiWrites(false) .setPropagateSuccessfulStorageApiWritesPredicate(Predicates.alwaysTrue()) .setDirectWriteProtos(true) @@ -2793,7 +2792,7 @@ public enum Method { abstract boolean getIgnoreInsertIds(); - abstract int getMaxRetryJobs(); + abstract @Nullable Integer getMaxRetryJobs(); abstract @Nullable String getKmsKey(); @@ -2925,7 +2924,7 @@ abstract Builder setDefaultMissingValueInterpretation( abstract Builder setAutoSharding(boolean autoSharding); - abstract Builder setMaxRetryJobs(int maxRetryJobs); + abstract Builder setMaxRetryJobs(@Nullable Integer maxRetryJobs); abstract Builder setPropagateSuccessful(boolean propagateSuccessful); @@ -3576,7 +3575,17 @@ public Write withAutoSharding() { return toBuilder().setAutoSharding(true).build(); } - /** If set, this will set the max number of retry of batch load jobs. */ + /** + * Sets the maximum number of times a failed BigQuery load or copy job is retried before the + * write fails. + * + *

Only applies when the write method is {@link Method#FILE_LOADS}. The streaming insert and + * Storage Write API methods retry at the row level and ignore this setting. + * + *

If this is not called, a bounded (batch) pipeline retries 3 times and an unbounded + * (streaming) pipeline retries 1000 times. Streaming defaults higher because failing a bundle + * in streaming is far more expensive than retrying the load job. + */ public Write withMaxRetryJobs(int maxRetryJobs) { return toBuilder().setMaxRetryJobs(maxRetryJobs).build(); } @@ -4235,10 +4244,15 @@ private WriteResult continueExpandTyped( batchLoads.setMaxFilesPerPartition(getMaxFilesPerPartition()); batchLoads.setMaxBytesPerPartition(getMaxBytesPerPartition()); - // When running in streaming (unbounded mode) we want to retry failed load jobs - // indefinitely. Failing the bundle is expensive, so we set a fairly high limit on retries. - if (IsBounded.UNBOUNDED.equals(input.isBounded())) { - batchLoads.setMaxRetryJobs(getMaxRetryJobs()); + // an explicit withMaxRetryJobs applies to batch and streaming alike. left unset, streaming + // retries a failed load job far more often than batch does: failing the bundle in streaming + // is expensive, so we would rather keep retrying the job than hand the work back to the + // runner. batch leaves BatchLoads on its own lower default + Integer maxRetryJobs = getMaxRetryJobs(); + if (maxRetryJobs != null) { + batchLoads.setMaxRetryJobs(maxRetryJobs); + } else if (IsBounded.UNBOUNDED.equals(input.isBounded())) { + batchLoads.setMaxRetryJobs(BatchLoads.DEFAULT_MAX_RETRY_JOBS_UNBOUNDED); } batchLoads.setTriggeringFrequency(getTriggeringFrequency()); if (getAutoSharding()) { diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java index 7310e2f850e5..41c868570a0c 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java @@ -2154,12 +2154,40 @@ public void testWriteFailedJobs() throws Exception { thrown.expect(RuntimeException.class); thrown.expectMessage("Failed to create job with prefix"); - thrown.expectMessage("reached max retries"); + // a bounded write that never calls withMaxRetryJobs keeps BatchLoads' own default of 3 + thrown.expectMessage("reached max retries: 3"); thrown.expectMessage("last failed job"); p.run(); } + @Test + public void testWriteFailedJobsRespectsMaxRetryJobsWhenBounded() throws Exception { + assumeTrue(!useStorageApi); + assumeTrue(!useStreaming); + p.apply( + Create.of( + new TableRow().set("name", "a").set("number", 1), + new TableRow().set("name", "b").set("number", 2), + new TableRow().set("name", "c").set("number", 3)) + .withCoder(TableRowJsonCoder.of())) + .apply( + BigQueryIO.writeTableRows() + .to("dataset-id.table-id") + .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER) + .withMaxRetryJobs(1) + .withTestServices(fakeBqServices) + .withoutValidation()); + + thrown.expect(RuntimeException.class); + // the load job fails on every attempt because the destination table does not exist, and the + // failure message reports the retry limit that was actually applied. a bounded pipeline used to + // drop withMaxRetryJobs on the floor, so this used to read "reached max retries: 3" + thrown.expectMessage("reached max retries: 1"); + + p.run(); + } + @Test public void testWriteWithMissingSchemaFromView() throws Exception { // Because no messages @@ -3019,7 +3047,7 @@ public void testMaxRetryJobs() { .withSchemaUpdateOptions( EnumSet.of(BigQueryIO.Write.SchemaUpdateOption.ALLOW_FIELD_ADDITION)) .withMaxRetryJobs(500); - assertEquals(500, write.getMaxRetryJobs()); + assertEquals(Integer.valueOf(500), write.getMaxRetryJobs()); } @Test From 1dbfb0b8cf071be3aa2f64337be3b8820ac25dfb Mon Sep 17 00:00:00 2001 From: Maksym Tymoshyk Date: Tue, 1 Sep 2026 15:40:12 +0300 Subject: [PATCH 2/2] Keep the bounded retry default when upgrading a pre-2.77.0 config row BigQueryIO.write() used to seed maxRetryJobs with 1000 whether or not the pipeline ever called withMaxRetryJobs, so every config row written by an older SDK carries that value. fromConfigRow restored it, and the new expand() would have read it as an explicit setting, raising bounded FILE_LOADS retries from 3 to 1000 on upgrade. A pre-2.77.0 row holding exactly 1000 says nothing about what the user asked for, so drop it and let each mode fall back to the default it used before: 3 for bounded and 1000 for unbounded. Any other value was chosen deliberately and is carried over. --- .../gcp/bigquery/BigQueryIOTranslation.java | 12 +++- .../bigquery/BigQueryIOTranslationTest.java | 65 +++++++++++++++++++ 2 files changed, 76 insertions(+), 1 deletion(-) diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTranslation.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTranslation.java index dd59939726bf..6f8e19e0b969 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTranslation.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTranslation.java @@ -830,8 +830,18 @@ public Write fromConfigRow(Row configRow, PipelineOptions options) { if (ignoreInsertIds != null) { builder = builder.setIgnoreInsertIds(ignoreInsertIds); } + // before 2.77.0 BigQueryIO.write() always seeded maxRetryJobs with 1000, whether or not the + // pipeline ever called withMaxRetryJobs, and a bounded write ignored the value and used + // BatchLoads' own default of 3. so a row from one of those versions that holds exactly 1000 + // tells us nothing about what the user asked for. leaving it unset makes both modes fall + // back to the same numbers they used before the upgrade, 3 for bounded and 1000 for + // unbounded. any other value was chosen deliberately and is carried over Integer maxRetryJobs = configRow.getInt32("max_retry_jobs"); - if (maxRetryJobs != null) { + boolean isPreservedLegacyDefault = + TransformUpgrader.compareVersions(updateCompatibilityBeamVersion, "2.77.0") < 0 + && Integer.valueOf(BatchLoads.DEFAULT_MAX_RETRY_JOBS_UNBOUNDED) + .equals(maxRetryJobs); + if (maxRetryJobs != null && !isPreservedLegacyDefault) { builder = builder.setMaxRetryJobs(maxRetryJobs); } String kmsKey = configRow.getString("kms_key"); diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTranslationTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTranslationTest.java index de63120c93cc..59f84657f724 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTranslationTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTranslationTest.java @@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import com.google.api.services.bigquery.model.Clustering; @@ -276,6 +277,70 @@ public void testReCreateWriteTransformFromRowTable() { writeTransformFromRow.getJsonClustering().get()); } + @Test + public void testReCreateWriteTransformDropsLegacyMaxRetryJobsDefault() { + // an SDK older than 2.77.0 put 1000 in every config row, so this is what a pipeline that never + // called withMaxRetryJobs looks like once one of those versions has serialized it + BigQueryIO.Write writeTransform = + BigQueryIO.write() + .to("dummyproject:dummydataset.dummytable") + .withMaxRetryJobs(BatchLoads.DEFAULT_MAX_RETRY_JOBS_UNBOUNDED); + + BigQueryIOTranslation.BigQueryIOWriteTranslator translator = + new BigQueryIOTranslation.BigQueryIOWriteTranslator(); + Row row = translator.toConfigRow(writeTransform); + + PipelineOptions options = PipelineOptionsFactory.create(); + options.as(StreamingOptions.class).setUpdateCompatibilityVersion("2.76.0"); + BigQueryIO.Write writeTransformFromRow = + (BigQueryIO.Write) translator.fromConfigRow(row, options); + + // unset, so a bounded write falls back to BatchLoads' default of 3 as it did before the + // upgrade, rather than jumping to 1000 + assertNull(writeTransformFromRow.getMaxRetryJobs()); + } + + @Test + public void testReCreateWriteTransformKeepsLegacyMaxRetryJobsOtherThanDefault() { + BigQueryIO.Write writeTransform = + BigQueryIO.write().to("dummyproject:dummydataset.dummytable").withMaxRetryJobs(7); + + BigQueryIOTranslation.BigQueryIOWriteTranslator translator = + new BigQueryIOTranslation.BigQueryIOWriteTranslator(); + Row row = translator.toConfigRow(writeTransform); + + PipelineOptions options = PipelineOptionsFactory.create(); + options.as(StreamingOptions.class).setUpdateCompatibilityVersion("2.76.0"); + BigQueryIO.Write writeTransformFromRow = + (BigQueryIO.Write) translator.fromConfigRow(row, options); + + // 7 was never a default in any version, so the pipeline must have asked for it + assertEquals(Integer.valueOf(7), writeTransformFromRow.getMaxRetryJobs()); + } + + @Test + public void testReCreateWriteTransformKeepsMaxRetryJobsFromCurrentVersion() { + BigQueryIO.Write writeTransform = + BigQueryIO.write() + .to("dummyproject:dummydataset.dummytable") + .withMaxRetryJobs(BatchLoads.DEFAULT_MAX_RETRY_JOBS_UNBOUNDED); + + BigQueryIOTranslation.BigQueryIOWriteTranslator translator = + new BigQueryIOTranslation.BigQueryIOWriteTranslator(); + Row row = translator.toConfigRow(writeTransform); + + PipelineOptions options = PipelineOptionsFactory.create(); + options.as(StreamingOptions.class).setUpdateCompatibilityVersion("2.77.0"); + BigQueryIO.Write writeTransformFromRow = + (BigQueryIO.Write) translator.fromConfigRow(row, options); + + // 2.77.0 and later only write this field when the pipeline set it, so 1000 is a real choice + // here + assertEquals( + Integer.valueOf(BatchLoads.DEFAULT_MAX_RETRY_JOBS_UNBOUNDED), + writeTransformFromRow.getMaxRetryJobs()); + } + @Test public void testWriteTransformRowIncludesAllFields() { // These fields do not represent properties of the transform.