From 581edc7e249c12f68d3cfcd02d7473ed56575ea4 Mon Sep 17 00:00:00 2001 From: Maksym Tymoshyk Date: Mon, 31 Aug 2026 21:23:41 +0300 Subject: [PATCH 1/4] [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 8fa59e7d5ea0..b39b9c495a67 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -125,6 +125,7 @@ * (Go) Fixed pubsubio importing a `google.golang.org/genproto` package removed in recent releases, which broke builds of Go modules depending on a current `genproto` version ([#40018](https://github.com/apache/beam/issues/40018)). * (Python) `VertexAIFeatureStoreEnrichmentHandler` now logs a warning instead of failing when the caller lacks permission to read the feature store metadata ([#40442](https://github.com/apache/beam/issues/40442)). * (Java) BigQueryIO now treats a 404 when deleting a temporary table or dataset as success, so a replayed work item whose earlier attempt already deleted it no longer retries forever ([#24997](https://github.com/apache/beam/issues/24997)). +* (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)). * (Java) IcebergIO now writes rows containing `EnumerationType` (proto enum) fields as strings, instead of throwing `Unsupported Beam logical type Enum` ([#40299](https://github.com/apache/beam/issues/40299)). * (Python) Fixed stateful DoFns with side inputs sometimes taking the timer key coder from a side input instead of the main input, which could make the worker fail to decode timer keys with `Unknown type tag` ([#40374](https://github.com/apache/beam/issues/40374)). * (Python) Fixed `AsyncWrapper` finalizing bundles before generator processing, which could drop buffered writes, and nesting iterable outputs ([#40000](https://github.com/apache/beam/pull/40000)). 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 d12af24e407d..c0dff11ac4d4 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 07fa1a320e92..a77f5d79934d 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 @@ -2775,7 +2775,6 @@ public static Write write() { .setAutoSchemaUpdate(false) .setAutoSchemaUpdateStrictTimeout(null) .setDeterministicRecordIdFn(null) - .setMaxRetryJobs(1000) .setPropagateSuccessfulStorageApiWrites(false) .setPropagateSuccessfulStorageApiWritesPredicate(Predicates.alwaysTrue()) .setDirectWriteProtos(true) @@ -3038,7 +3037,7 @@ public enum Method { abstract boolean getIgnoreInsertIds(); - abstract int getMaxRetryJobs(); + abstract @Nullable Integer getMaxRetryJobs(); abstract @Nullable String getKmsKey(); @@ -3172,7 +3171,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); @@ -3825,7 +3824,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(); } @@ -4549,10 +4558,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 62e57ec848cb..c09c47acc997 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 @@ -2208,12 +2208,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 @@ -3114,7 +3142,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 4aafd272854f9667c01a72010063d0fd114e553c Mon Sep 17 00:00:00 2001 From: Maksym Tymoshyk Date: Tue, 1 Sep 2026 15:40:12 +0300 Subject: [PATCH 2/4] 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 9e81b675a755..c2f0b0ee1e5e 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 @@ -898,8 +898,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 4391a66e27ee..671adc56762a 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; @@ -278,6 +279,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. From 1a3f5279b4741d67682d4ef64e4aa95a23379fd9 Mon Sep 17 00:00:00 2001 From: Maksym Tymoshyk Date: Fri, 25 Sep 2026 13:47:09 +0300 Subject: [PATCH 3/4] Move the maxRetryJobs upgrade threshold to 2.78.0 release-2.77 was cut before this change landed, so 2.77.0 still seeds maxRetryJobs with 1000 in every config row. Treat rows from 2.77.0 and earlier as legacy. --- .../sdk/io/gcp/bigquery/BigQueryIOTranslation.java | 4 ++-- .../sdk/io/gcp/bigquery/BigQueryIOTranslationTest.java | 10 +++++----- 2 files changed, 7 insertions(+), 7 deletions(-) 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 c2f0b0ee1e5e..4720d4168195 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 @@ -898,7 +898,7 @@ 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 + // before 2.78.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 @@ -906,7 +906,7 @@ public Write fromConfigRow(Row configRow, PipelineOptions options) { // unbounded. any other value was chosen deliberately and is carried over Integer maxRetryJobs = configRow.getInt32("max_retry_jobs"); boolean isPreservedLegacyDefault = - TransformUpgrader.compareVersions(updateCompatibilityBeamVersion, "2.77.0") < 0 + TransformUpgrader.compareVersions(updateCompatibilityBeamVersion, "2.78.0") < 0 && Integer.valueOf(BatchLoads.DEFAULT_MAX_RETRY_JOBS_UNBOUNDED) .equals(maxRetryJobs); if (maxRetryJobs != null && !isPreservedLegacyDefault) { 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 671adc56762a..9cf04834f707 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 @@ -281,7 +281,7 @@ public void testReCreateWriteTransformFromRowTable() { @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 + // an SDK older than 2.78.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() @@ -293,7 +293,7 @@ public void testReCreateWriteTransformDropsLegacyMaxRetryJobsDefault() { Row row = translator.toConfigRow(writeTransform); PipelineOptions options = PipelineOptionsFactory.create(); - options.as(StreamingOptions.class).setUpdateCompatibilityVersion("2.76.0"); + options.as(StreamingOptions.class).setUpdateCompatibilityVersion("2.77.0"); BigQueryIO.Write writeTransformFromRow = (BigQueryIO.Write) translator.fromConfigRow(row, options); @@ -312,7 +312,7 @@ public void testReCreateWriteTransformKeepsLegacyMaxRetryJobsOtherThanDefault() Row row = translator.toConfigRow(writeTransform); PipelineOptions options = PipelineOptionsFactory.create(); - options.as(StreamingOptions.class).setUpdateCompatibilityVersion("2.76.0"); + options.as(StreamingOptions.class).setUpdateCompatibilityVersion("2.77.0"); BigQueryIO.Write writeTransformFromRow = (BigQueryIO.Write) translator.fromConfigRow(row, options); @@ -332,11 +332,11 @@ public void testReCreateWriteTransformKeepsMaxRetryJobsFromCurrentVersion() { Row row = translator.toConfigRow(writeTransform); PipelineOptions options = PipelineOptionsFactory.create(); - options.as(StreamingOptions.class).setUpdateCompatibilityVersion("2.77.0"); + options.as(StreamingOptions.class).setUpdateCompatibilityVersion("2.78.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 + // 2.78.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), From 2d0413f0cf2ff2f5b22f556db614d9f80807cb5d Mon Sep 17 00:00:00 2001 From: Maksym Tymoshyk Date: Fri, 25 Sep 2026 14:30:58 +0300 Subject: [PATCH 4/4] Skip a null maxRetryJobs when building the Write config row master now runs the nullness checker on BigQueryIOTranslation, and getMaxRetryJobs() is nullable after this change. --- .../beam/sdk/io/gcp/bigquery/BigQueryIOTranslation.java | 5 ++++- 1 file changed, 4 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 4720d4168195..66b1520e78e9 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 @@ -624,7 +624,10 @@ public Row toConfigRow(Write transform) { fieldValues.put("skip_invalid_rows", transform.getSkipInvalidRows()); fieldValues.put("ignore_unknown_values", transform.getIgnoreUnknownValues()); fieldValues.put("ignore_insert_ids", transform.getIgnoreInsertIds()); - fieldValues.put("max_retry_jobs", transform.getMaxRetryJobs()); + Integer maxRetryJobs = transform.getMaxRetryJobs(); + if (maxRetryJobs != null) { + fieldValues.put("max_retry_jobs", maxRetryJobs); + } fieldValues.put("propagate_successful", transform.getPropagateSuccessful()); String kmsKey = transform.getKmsKey(); if (kmsKey != null) {