[Bug]: BigQuery IO Batch load using File_load causing the same job id ignoring inserts as the job_id is already completed

[yeshvantbhavnasi]

What happened?

BigQueryIO.Write<Row> batchWrite =
          rowWrite
              .withMethod(BigQueryIO.Write.Method.FILE_LOADS)
              .withTriggeringFrequency(Duration.standardMinutes(fileLoadSpecOptional.get().writeTriggerFrequency))
              .withAutoSharding()
              .withExtendedErrorInfo()
              .withMaxRetryJobs(fileLoadSpecOptional.get().withJobRetryCount)
              .withWriteTempDataset(TMP_TABLE_DATASET)
              .withCustomGcsTempLocation(ValueProvider.StaticValueProvider.of(fileLoadSpecOptional.get().stagingBucket));

Have following code to setup file_load operation for Dataflow streaming jobs using GCS and File_notification based ingestion into BigQuery

• It creates a JobId for new file after few minutes and load data into Live table with jobId beam_bq_job_LOAD_defaultnetworkvpcflowlogslive_e78561b1b6e147e7abafe83b9314c7f1_41dd2dbb7ce5b6e717e5aadb4244b4e0_00001_00000-0
  but after this every new files are added to staging bucket new set of files have exactly 1 partition and c.pane.index seems to be 0 causing the same job_id and ignoring file writes
  See the screenshot for different timestamp we have the same job_id ignoring inserts

Should we add random uuid to jobId logic to fix this issue here ?

by changing this line to:

 String jobIdPrefix =
          BigQueryResourceNaming.createJobIdWithDestination(
              c.sideInput(loadJobIdPrefixView), tableDestination, partition, c.pane().getIndex()) + "-" +  random5CharUUID();

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[maskalski-clgx]

We faced the same issue after we switched to Dataflow Runner v2. BigQuery Load job with the same jobId is used for every batch write so only the first batch is actually written.

In addition to @yeshvantbhavnasi 's logs, I have the following warnings:

[Abacn]

Thanks for reporting this. It's clearly a bug and should be fixed.

We faced the same issue after we switched to Dataflow Runner v2.

This also suggests runner v2 somehow generates smaller load jobs. Most of time, within one window it should only have one load job. Another issue to track.

[Abacn]

I see, they are different window. Should also add window to job name.

[Abacn]

There is a bug in Dataflow Runner V2 where ProcessContext c.pane().getIndex() always returns 0 here:

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

Line 254 in a636f25

c.sideInput(loadJobIdPrefixView), tableDestination, partition, c.pane().getIndex());

Causing duplicated BigQuery job id. However in Dataflow Runner V1 the pane index is increasing.