[Gemini] Migrate Python BigQuery away from apitools, deprecate old route - #39889

apache_beam.runners.dataflow.dataflow_runner.DataflowRuntimeException: Dataflow pipeline failed. State: FAILED, Error:
Traceback (most recent call last):
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/utils/retry.py", line 310, in wrapper
    sleep_interval = next(retry_intervals)
StopIteration

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "apache_beam/runners/common.py", line 1512, in apache_beam.runners.common.DoFnRunner.process
    return self.do_fn_invoker.invoke_process(windowed_value)
  File "apache_beam/runners/common.py", line 914, in apache_beam.runners.common.PerWindowInvoker.invoke_process
    self._invoke_process_per_window(
  File "apache_beam/runners/common.py", line 1059, in apache_beam.runners.common.PerWindowInvoker._invoke_process_per_window
    self.process_method(*args_for_process, **kwargs_for_process),
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery.py", line 1726, in process
    self._create_table_if_needed(
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
        bigquery_tools.parse_table_reference(destination), schema)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery.py", line 1690, in _create_table_if_needed
    self.bigquery_wrapper.get_or_create_table(
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
        table_reference.projectId,
        ^^^^^^^^^^^^^^^^^^^^^^^^^^
    ...<4 lines>...
        self.write_disposition,
        ^^^^^^^^^^^^^^^^^^^^^^^
        additional_create_parameters=self.additional_bq_parameters)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/utils/retry.py", line 313, in wrapper
    raise exn.with_traceback(exn_traceback)
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/utils/retry.py", line 300, in wrapper
    return fun(*args, **kwargs)
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery_tools.py", line 2954, in get_or_create_table
    created_table = self._create_table(
        project_id=project_id,
    ...<2 lines>...
        schema=schema or found_table.schema,
        additional_parameters=additional_create_parameters)
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery_tools.py", line 2391, in _create_table
    setattr(table, k, v)
    ~~~~~~~^^^^^^^^^^^^^
AttributeError: property '<lambda>' of 'Table' object has no setter

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/runners/worker/sdk_worker.py", line 310, in _execute
    response = task()
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/runners/worker/sdk_worker.py", line 384, in <lambda>
    lambda: self.create_worker().do_instruction(request), request)
            ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/runners/worker/sdk_worker.py", line 671, in do_instruction
    return getattr(self, request_type)(
           ~~~~~~~~~~~~~~~~~~~~~~~~~~~^
        getattr(request, request_type), request.instruction_id)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/runners/worker/sdk_worker.py", line 709, in process_bundle
    bundle_processor.process_bundle(instruction_id))
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1310, in process_bundle
    input_op_by_transform_id[element.transform_id].process_encoded(
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
        element.data)
        ^^^^^^^^^^^^^
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/runners/worker/bundle_processor.py", line 235, in process_encoded
    self.output(decoded_value)
    ~~~~~~~~~~~^^^^^^^^^^^^^^^
  File "apache_beam/runners/worker/operations.py", line 569, in apache_beam.runners.worker.operations.Operation.output
    def output(self, windowed_value, output_index=0):
  File "apache_beam/runners/worker/operations.py", line 571, in apache_beam.runners.worker.operations.Operation.output
    _cast_to_receiver(self.receivers[output_index]).receive(windowed_value)
  File "apache_beam/runners/worker/operations.py", line 261, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
    def receive(self, windowed_value):
  File "apache_beam/runners/worker/operations.py", line 264, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
    self.consumer.process(windowed_value)
  File "apache_beam/runners/worker/operations.py", line 955, in apache_beam.runners.worker.operations.DoOperation.process
    with self.scoped_process_state:
  File "apache_beam/runners/worker/operations.py", line 956, in apache_beam.runners.worker.operations.DoOperation.process
    delayed_applications = self.dofn_runner.process(o)
  File "apache_beam/runners/common.py", line 1514, in apache_beam.runners.common.DoFnRunner.process
    self._reraise_augmented(exn, windowed_value)
  File "apache_beam/runners/common.py", line 1602, in apache_beam.runners.common.DoFnRunner._reraise_augmented
    raise exn
  File "apache_beam/runners/common.py", line 1512, in apache_beam.runners.common.DoFnRunner.process
    return self.do_fn_invoker.invoke_process(windowed_value)
  File "apache_beam/runners/common.py", line 686, in apache_beam.runners.common.SimpleInvoker.invoke_process
    self.output_handler.handle_process_outputs(
  File "apache_beam/runners/common.py", line 1719, in apache_beam.runners.common._OutputHandler.handle_process_outputs
    self._write_value_to_tag(tag, windowed_value, watermark_estimator)
  File "apache_beam/runners/common.py", line 1832, in apache_beam.runners.common._OutputHandler._write_value_to_tag
    self.main_receivers.receive(windowed_value)
  File "apache_beam/runners/worker/operations.py", line 264, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
    self.consumer.process(windowed_value)
  File "apache_beam/runners/worker/operations.py", line 955, in apache_beam.runners.worker.operations.DoOperation.process
    with self.scoped_process_state:
  File "apache_beam/runners/worker/operations.py", line 956, in apache_beam.runners.worker.operations.DoOperation.process
    delayed_applications = self.dofn_runner.process(o)
  File "apache_beam/runners/common.py", line 1514, in apache_beam.runners.common.DoFnRunner.process
    self._reraise_augmented(exn, windowed_value)
  File "apache_beam/runners/common.py", line 1602, in apache_beam.runners.common.DoFnRunner._reraise_augmented
    raise exn
  File "apache_beam/runners/common.py", line 1512, in apache_beam.runners.common.DoFnRunner.process
    return self.do_fn_invoker.invoke_process(windowed_value)
  File "apache_beam/runners/common.py", line 686, in apache_beam.runners.common.SimpleInvoker.invoke_process
    self.output_handler.handle_process_outputs(
  File "apache_beam/runners/common.py", line 1719, in apache_beam.runners.common._OutputHandler.handle_process_outputs
    self._write_value_to_tag(tag, windowed_value, watermark_estimator)
  File "apache_beam/runners/common.py", line 1832, in apache_beam.runners.common._OutputHandler._write_value_to_tag
    self.main_receivers.receive(windowed_value)
  File "apache_beam/runners/worker/operations.py", line 264, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
    self.consumer.process(windowed_value)
  File "apache_beam/runners/worker/operations.py", line 955, in apache_beam.runners.worker.operations.DoOperation.process
    with self.scoped_process_state:
  File "apache_beam/runners/worker/operations.py", line 956, in apache_beam.runners.worker.operations.DoOperation.process
    delayed_applications = self.dofn_runner.process(o)
  File "apache_beam/runners/common.py", line 1514, in apache_beam.runners.common.DoFnRunner.process
    self._reraise_augmented(exn, windowed_value)
  File "apache_beam/runners/common.py", line 1602, in apache_beam.runners.common.DoFnRunner._reraise_augmented
    raise exn
  File "apache_beam/runners/common.py", line 1512, in apache_beam.runners.common.DoFnRunner.process
    return self.do_fn_invoker.invoke_process(windowed_value)
  File "apache_beam/runners/common.py", line 686, in apache_beam.runners.common.SimpleInvoker.invoke_process
    self.output_handler.handle_process_outputs(
  File "apache_beam/runners/common.py", line 1719, in apache_beam.runners.common._OutputHandler.handle_process_outputs
    self._write_value_to_tag(tag, windowed_value, watermark_estimator)
  File "apache_beam/runners/common.py", line 1832, in apache_beam.runners.common._OutputHandler._write_value_to_tag
    self.main_receivers.receive(windowed_value)
  File "apache_beam/runners/worker/operations.py", line 264, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
    self.consumer.process(windowed_value)
  File "apache_beam/runners/worker/operations.py", line 955, in apache_beam.runners.worker.operations.DoOperation.process
    with self.scoped_process_state:
  File "apache_beam/runners/worker/operations.py", line 956, in apache_beam.runners.worker.operations.DoOperation.process
    delayed_applications = self.dofn_runner.process(o)
  File "apache_beam/runners/common.py", line 1514, in apache_beam.runners.common.DoFnRunner.process
    self._reraise_augmented(exn, windowed_value)
  File "apache_beam/runners/common.py", line 1623, in apache_beam.runners.common.DoFnRunner._reraise_augmented
    raise new_exn
  File "apache_beam/runners/common.py", line 1512, in apache_beam.runners.common.DoFnRunner.process
    return self.do_fn_invoker.invoke_process(windowed_value)
  File "apache_beam/runners/common.py", line 914, in apache_beam.runners.common.PerWindowInvoker.invoke_process
    self._invoke_process_per_window(
  File "apache_beam/runners/common.py", line 1059, in apache_beam.runners.common.PerWindowInvoker._invoke_process_per_window
    self.process_method(*args_for_process, **kwargs_for_process),
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery.py", line 1726, in process
    self._create_table_if_needed(
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
        bigquery_tools.parse_table_reference(destination), schema)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery.py", line 1690, in _create_table_if_needed
    self.bigquery_wrapper.get_or_create_table(
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
        table_reference.projectId,
        ^^^^^^^^^^^^^^^^^^^^^^^^^^
    ...<4 lines>...
        self.write_disposition,
        ^^^^^^^^^^^^^^^^^^^^^^^
        additional_create_parameters=self.additional_bq_parameters)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/utils/retry.py", line 313, in wrapper
    raise exn.with_traceback(exn_traceback)
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/utils/retry.py", line 300, in wrapper
    return fun(*args, **kwargs)
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery_tools.py", line 2954, in get_or_create_table
    created_table = self._create_table(
        project_id=project_id,
    ...<2 lines>...
        schema=schema or found_table.schema,
        additional_parameters=additional_create_parameters)
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery_tools.py", line 2391, in _create_table
    setattr(table, k, v)
    ~~~~~~~^^^^^^^^^^^^^
AttributeError: property '<lambda>' of 'Table' object has no setter [while running 'WriteWithMultipleDests/_StreamToBigQuery/StreamInsertRows/ParDo(BigQueryWriteFn)-ptransform-60']
self = <apache_beam.io.gcp.bigquery_test.BigQueryStreamingInsertTransformIntegrationTests testMethod=test_value_provider_transform>

    @pytest.mark.it_postcommit
    def test_value_provider_transform(self):
      output_table_1 = '%s%s' % (self.output_table, 1)
      output_table_2 = '%s%s' % (self.output_table, 2)
      schema = {
          'fields': [{
              'name': 'name', 'type': 'STRING', 'mode': 'NULLABLE'
          }, {
              'name': 'language', 'type': 'STRING', 'mode': 'NULLABLE'
          }]
      }
    
      additional_bq_parameters = {
          'timePartitioning': {
              'type': 'DAY'
          },
          'clustering': {
              'fields': ['language']
          }
      }
    
      table_ref = bigquery_tools.parse_table_reference(output_table_1)
      table_ref2 = bigquery_tools.parse_table_reference(output_table_2)
    
      pipeline_verifiers = [
          BigQueryTableMatcher(
              project=self.project,
              dataset=table_ref.datasetId,
              table=table_ref.tableId,
              expected_properties=additional_bq_parameters),
          BigQueryTableMatcher(
              project=self.project,
              dataset=table_ref2.datasetId,
              table=table_ref2.tableId,
              expected_properties=additional_bq_parameters),
          BigqueryFullResultMatcher(
              project=self.project,
              query="SELECT name, language FROM %s" % output_table_1,
              data=[(d['name'], d['language']) for d in _ELEMENTS
                    if 'language' in d]),
          BigqueryFullResultMatcher(
              project=self.project,
              query="SELECT name, language FROM %s" % output_table_2,
              data=[(d['name'], d['language']) for d in _ELEMENTS
                    if 'language' in d])
      ]
    
      args = self.test_pipeline.get_full_options_as_args(
          on_success_matcher=hc.all_of(*pipeline_verifiers))
    
>     with beam.Pipeline(argv=args) as p:
           ^^^^^^^^^^^^^^^^^^^^^^^^

apache_beam/io/gcp/bigquery_test.py:2778: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
apache_beam/pipeline.py:652: in __exit__
    self.result = self.run()
                  ^^^^^^^^^^
apache_beam/pipeline.py:563: in run
    return self._run_internal(test_runner_api)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/pipeline.py:605: in _run_internal
    self._options).run(False)
                   ^^^^^^^^^^
apache_beam/pipeline.py:563: in run
    return self._run_internal(test_runner_api)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/pipeline.py:629: in _run_internal
    return self.runner.run_pipeline(self, self._options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/runners/dataflow/test_dataflow_runner.py:66: in run_pipeline
    self.result.wait_until_finish(duration=wait_duration)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <DataflowPipelineResult id: "2026-08-27_11_01_30-11499914128251294059"
project_id: "apache-beam-testing"
name: "beamap...y_name: "Apache Beam Python 3.14 SDK"
  }
}
start_time {
  seconds: 1787853691
  nanos: 627652000
}
 at 0x7e5e14187610>
duration = None

    def wait_until_finish(self, duration=None):
      if not self.is_in_terminal_state():
        if not self.has_job:
          raise IOError('Failed to get the Dataflow job id.')
        gcp_options = self._options.view_as(GoogleCloudOptions)
        consoleUrl = (
            "Console URL: https://console.cloud.google.com/"
            f"dataflow/jobs/{gcp_options.region}/{self.job_id()}"
            f"?project={gcp_options.project}")
        thread = threading.Thread(
            target=DataflowRunner.poll_for_job_completion,
            args=(self._runner, self, duration))
    
        # Mark the thread as a daemon thread so a keyboard interrupt on the main
        # thread will terminate everything. This is also the reason we will not
        # use thread.join() to wait for the polling thread.
        thread.daemon = True
        thread.start()
        while thread.is_alive():
          time.sleep(5.0)
    
        # TODO: Merge the termination code in poll_for_job_completion and
        # is_in_terminal_state.
        terminated = self.is_in_terminal_state()
        assert duration or terminated, (
            'Job did not reach to a terminal state after waiting indefinitely. '
            '{}'.format(consoleUrl))
    
        if terminated and self.state != PipelineState.DONE:
          # TODO(BEAM-1290): Consider converting this to an error log based on
          # theresolution of the issue.
          _LOGGER.error(consoleUrl)
>         raise DataflowRuntimeException(
              'Dataflow pipeline failed. State: %s, Error:\n%s' %
              (self.state, getattr(self._runner, 'last_error_msg', None)),
E             apache_beam.runners.dataflow.dataflow_runner.DataflowRuntimeException: Dataflow pipeline failed. State: FAILED, Error:
E             Traceback (most recent call last):
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/utils/retry.py", line 310, in wrapper
E                 sleep_interval = next(retry_intervals)
E             StopIteration
E             
E             During handling of the above exception, another exception occurred:
E             
E             Traceback (most recent call last):
E               File "apache_beam/runners/common.py", line 1512, in apache_beam.runners.common.DoFnRunner.process
E                 return self.do_fn_invoker.invoke_process(windowed_value)
E               File "apache_beam/runners/common.py", line 914, in apache_beam.runners.common.PerWindowInvoker.invoke_process
E                 self._invoke_process_per_window(
E               File "apache_beam/runners/common.py", line 1059, in apache_beam.runners.common.PerWindowInvoker._invoke_process_per_window
E                 self.process_method(*args_for_process, **kwargs_for_process),
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery.py", line 1726, in process
E                 self._create_table_if_needed(
E                 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
E                     bigquery_tools.parse_table_reference(destination), schema)
E                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery.py", line 1690, in _create_table_if_needed
E                 self.bigquery_wrapper.get_or_create_table(
E                 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
E                     table_reference.projectId,
E                     ^^^^^^^^^^^^^^^^^^^^^^^^^^
E                 ...<4 lines>...
E                     self.write_disposition,
E                     ^^^^^^^^^^^^^^^^^^^^^^^
E                     additional_create_parameters=self.additional_bq_parameters)
E                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/utils/retry.py", line 313, in wrapper
E                 raise exn.with_traceback(exn_traceback)
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/utils/retry.py", line 300, in wrapper
E                 return fun(*args, **kwargs)
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery_tools.py", line 2954, in get_or_create_table
E                 created_table = self._create_table(
E                     project_id=project_id,
E                 ...<2 lines>...
E                     schema=schema or found_table.schema,
E                     additional_parameters=additional_create_parameters)
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery_tools.py", line 2391, in _create_table
E                 setattr(table, k, v)
E                 ~~~~~~~^^^^^^^^^^^^^
E             AttributeError: property '<lambda>' of 'Table' object has no setter
E             
E             During handling of the above exception, another exception occurred:
E             
E             Traceback (most recent call last):
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/runners/worker/sdk_worker.py", line 310, in _execute
E                 response = task()
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/runners/worker/sdk_worker.py", line 384, in <lambda>
E                 lambda: self.create_worker().do_instruction(request), request)
E                         ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/runners/worker/sdk_worker.py", line 671, in do_instruction
E                 return getattr(self, request_type)(
E                        ~~~~~~~~~~~~~~~~~~~~~~~~~~~^
E                     getattr(request, request_type), request.instruction_id)
E                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/runners/worker/sdk_worker.py", line 709, in process_bundle
E                 bundle_processor.process_bundle(instruction_id))
E                 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1310, in process_bundle
E                 input_op_by_transform_id[element.transform_id].process_encoded(
E                 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
E                     element.data)
E                     ^^^^^^^^^^^^^
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/runners/worker/bundle_processor.py", line 235, in process_encoded
E                 self.output(decoded_value)
E                 ~~~~~~~~~~~^^^^^^^^^^^^^^^
E               File "apache_beam/runners/worker/operations.py", line 569, in apache_beam.runners.worker.operations.Operation.output
E                 def output(self, windowed_value, output_index=0):
E               File "apache_beam/runners/worker/operations.py", line 571, in apache_beam.runners.worker.operations.Operation.output
E                 _cast_to_receiver(self.receivers[output_index]).receive(windowed_value)
E               File "apache_beam/runners/worker/operations.py", line 261, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
E                 def receive(self, windowed_value):
E               File "apache_beam/runners/worker/operations.py", line 264, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
E                 self.consumer.process(windowed_value)
E               File "apache_beam/runners/worker/operations.py", line 955, in apache_beam.runners.worker.operations.DoOperation.process
E                 with self.scoped_process_state:
E               File "apache_beam/runners/worker/operations.py", line 956, in apache_beam.runners.worker.operations.DoOperation.process
E                 delayed_applications = self.dofn_runner.process(o)
E               File "apache_beam/runners/common.py", line 1514, in apache_beam.runners.common.DoFnRunner.process
E                 self._reraise_augmented(exn, windowed_value)
E               File "apache_beam/runners/common.py", line 1602, in apache_beam.runners.common.DoFnRunner._reraise_augmented
E                 raise exn
E               File "apache_beam/runners/common.py", line 1512, in apache_beam.runners.common.DoFnRunner.process
E                 return self.do_fn_invoker.invoke_process(windowed_value)
E               File "apache_beam/runners/common.py", line 686, in apache_beam.runners.common.SimpleInvoker.invoke_process
E                 self.output_handler.handle_process_outputs(
E               File "apache_beam/runners/common.py", line 1719, in apache_beam.runners.common._OutputHandler.handle_process_outputs
E                 self._write_value_to_tag(tag, windowed_value, watermark_estimator)
E               File "apache_beam/runners/common.py", line 1832, in apache_beam.runners.common._OutputHandler._write_value_to_tag
E                 self.main_receivers.receive(windowed_value)
E               File "apache_beam/runners/worker/operations.py", line 264, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
E                 self.consumer.process(windowed_value)
E               File "apache_beam/runners/worker/operations.py", line 955, in apache_beam.runners.worker.operations.DoOperation.process
E                 with self.scoped_process_state:
E               File "apache_beam/runners/worker/operations.py", line 956, in apache_beam.runners.worker.operations.DoOperation.process
E                 delayed_applications = self.dofn_runner.process(o)
E               File "apache_beam/runners/common.py", line 1514, in apache_beam.runners.common.DoFnRunner.process
E                 self._reraise_augmented(exn, windowed_value)
E               File "apache_beam/runners/common.py", line 1602, in apache_beam.runners.common.DoFnRunner._reraise_augmented
E                 raise exn
E               File "apache_beam/runners/common.py", line 1512, in apache_beam.runners.common.DoFnRunner.process
E                 return self.do_fn_invoker.invoke_process(windowed_value)
E               File "apache_beam/runners/common.py", line 686, in apache_beam.runners.common.SimpleInvoker.invoke_process
E                 self.output_handler.handle_process_outputs(
E               File "apache_beam/runners/common.py", line 1719, in apache_beam.runners.common._OutputHandler.handle_process_outputs
E                 self._write_value_to_tag(tag, windowed_value, watermark_estimator)
E               File "apache_beam/runners/common.py", line 1832, in apache_beam.runners.common._OutputHandler._write_value_to_tag
E                 self.main_receivers.receive(windowed_value)
E               File "apache_beam/runners/worker/operations.py", line 264, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
E                 self.consumer.process(windowed_value)
E               File "apache_beam/runners/worker/operations.py", line 955, in apache_beam.runners.worker.operations.DoOperation.process
E                 with self.scoped_process_state:
E               File "apache_beam/runners/worker/operations.py", line 956, in apache_beam.runners.worker.operations.DoOperation.process
E                 delayed_applications = self.dofn_runner.process(o)
E               File "apache_beam/runners/common.py", line 1514, in apache_beam.runners.common.DoFnRunner.process
E                 self._reraise_augmented(exn, windowed_value)
E               File "apache_beam/runners/common.py", line 1602, in apache_beam.runners.common.DoFnRunner._reraise_augmented
E                 raise exn
E               File "apache_beam/runners/common.py", line 1512, in apache_beam.runners.common.DoFnRunner.process
E                 return self.do_fn_invoker.invoke_process(windowed_value)
E               File "apache_beam/runners/common.py", line 686, in apache_beam.runners.common.SimpleInvoker.invoke_process
E                 self.output_handler.handle_process_outputs(
E               File "apache_beam/runners/common.py", line 1719, in apache_beam.runners.common._OutputHandler.handle_process_outputs
E                 self._write_value_to_tag(tag, windowed_value, watermark_estimator)
E               File "apache_beam/runners/common.py", line 1832, in apache_beam.runners.common._OutputHandler._write_value_to_tag
E                 self.main_receivers.receive(windowed_value)
E               File "apache_beam/runners/worker/operations.py", line 264, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
E                 self.consumer.process(windowed_value)
E               File "apache_beam/runners/worker/operations.py", line 955, in apache_beam.runners.worker.operations.DoOperation.process
E                 with self.scoped_process_state:
E               File "apache_beam/runners/worker/operations.py", line 956, in apache_beam.runners.worker.operations.DoOperation.process
E                 delayed_applications = self.dofn_runner.process(o)
E               File "apache_beam/runners/common.py", line 1514, in apache_beam.runners.common.DoFnRunner.process
E                 self._reraise_augmented(exn, windowed_value)
E               File "apache_beam/runners/common.py", line 1623, in apache_beam.runners.common.DoFnRunner._reraise_augmented
E                 raise new_exn
E               File "apache_beam/runners/common.py", line 1512, in apache_beam.runners.common.DoFnRunner.process
E                 return self.do_fn_invoker.invoke_process(windowed_value)
E               File "apache_beam/runners/common.py", line 914, in apache_beam.runners.common.PerWindowInvoker.invoke_process
E                 self._invoke_process_per_window(
E               File "apache_beam/runners/common.py", line 1059, in apache_beam.runners.common.PerWindowInvoker._invoke_process_per_window
E                 self.process_method(*args_for_process, **kwargs_for_process),
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery.py", line 1726, in process
E                 self._create_table_if_needed(
E                 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
E                     bigquery_tools.parse_table_reference(destination), schema)
E                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery.py", line 1690, in _create_table_if_needed
E                 self.bigquery_wrapper.get_or_create_table(
E                 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
E                     table_reference.projectId,
E                     ^^^^^^^^^^^^^^^^^^^^^^^^^^
E                 ...<4 lines>...
E                     self.write_disposition,
E                     ^^^^^^^^^^^^^^^^^^^^^^^
E                     additional_create_parameters=self.additional_bq_parameters)
E                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/utils/retry.py", line 313, in wrapper
E                 raise exn.with_traceback(exn_traceback)
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/utils/retry.py", line 300, in wrapper
E                 return fun(*args, **kwargs)
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery_tools.py", line 2954, in get_or_create_table
E                 created_table = self._create_table(
E                     project_id=project_id,
E                 ...<2 lines>...
E                     schema=schema or found_table.schema,
E                     additional_parameters=additional_create_parameters)
E               File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.14/site-packages/apache_beam/io/gcp/bigquery_tools.py", line 2391, in _create_table
E                 setattr(table, k, v)
E                 ~~~~~~~^^^^^^^^^^^^^
E             AttributeError: property '<lambda>' of 'Table' object has no setter [while running 'WriteWithMultipleDests/_StreamToBigQuery/StreamInsertRows/ParDo(BigQueryWriteFn)-ptransform-60']

apache_beam/runners/dataflow/dataflow_runner.py:834: DataflowRuntimeException

apache_beam.runners.dataflow.dataflow_runner.DataflowRuntimeException: Dataflow pipeline failed. State: FAILED, Error:
Workflow failed.
self = <apache_beam.io.gcp.bigquery_write_it_test.BigQueryWriteIntegrationTests testMethod=test_big_query_write_insert_non_transient_api_call_error>

    @pytest.mark.it_postcommit
    def test_big_query_write_insert_non_transient_api_call_error(self):
      """
      Test that non-transient GoogleAPICallError errors returned
      by beam.io.WriteToBigQuery are not retried and result in
      FAILED_ROWS containing both the failed rows and the reason
      for failure.
      """
      table_name = 'this_table_does_not_exist'
      table_id = '{}.{}'.format(self.dataset_id, table_name)
    
      input_data = [{
          'number': 1,
          'str': 'some_string',
      }]
    
      table_schema = {
          "fields": [{
              "name": "number", "type": "INTEGER", 'mode': 'NULLABLE'
          }, {
              "name": "str", "type": "STRING", 'mode': 'NULLABLE'
          }]
      }
    
      bq_result_errors = [({
          'number': 1,
          'str': 'some_string',
      }, "Not Found")]
    
      args = self.test_pipeline.get_full_options_as_args()
    
>     with beam.Pipeline(argv=args) as p:
           ^^^^^^^^^^^^^^^^^^^^^^^^

apache_beam/io/gcp/bigquery_write_it_test.py:563: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
apache_beam/pipeline.py:652: in __exit__
    self.result = self.run()
                  ^^^^^^^^^^
apache_beam/pipeline.py:563: in run
    return self._run_internal(test_runner_api)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/pipeline.py:605: in _run_internal
    self._options).run(False)
                   ^^^^^^^^^^
apache_beam/pipeline.py:563: in run
    return self._run_internal(test_runner_api)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/pipeline.py:629: in _run_internal
    return self.runner.run_pipeline(self, self._options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/runners/dataflow/test_dataflow_runner.py:66: in run_pipeline
    self.result.wait_until_finish(duration=wait_duration)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <DataflowPipelineResult id: "2026-08-27_11_13_56-929348224382641902"
project_id: "apache-beam-testing"
name: "beamapp-...y_name: "Apache Beam Python 3.14 SDK"
  }
}
start_time {
  seconds: 1787854437
  nanos: 281328000
}
 at 0x7d82bc2abbb0>
duration = None

    def wait_until_finish(self, duration=None):
      if not self.is_in_terminal_state():
        if not self.has_job:
          raise IOError('Failed to get the Dataflow job id.')
        gcp_options = self._options.view_as(GoogleCloudOptions)
        consoleUrl = (
            "Console URL: https://console.cloud.google.com/"
            f"dataflow/jobs/{gcp_options.region}/{self.job_id()}"
            f"?project={gcp_options.project}")
        thread = threading.Thread(
            target=DataflowRunner.poll_for_job_completion,
            args=(self._runner, self, duration))
    
        # Mark the thread as a daemon thread so a keyboard interrupt on the main
        # thread will terminate everything. This is also the reason we will not
        # use thread.join() to wait for the polling thread.
        thread.daemon = True
        thread.start()
        while thread.is_alive():
          time.sleep(5.0)
    
        # TODO: Merge the termination code in poll_for_job_completion and
        # is_in_terminal_state.
        terminated = self.is_in_terminal_state()
        assert duration or terminated, (
            'Job did not reach to a terminal state after waiting indefinitely. '
            '{}'.format(consoleUrl))
    
        if terminated and self.state != PipelineState.DONE:
          # TODO(BEAM-1290): Consider converting this to an error log based on
          # theresolution of the issue.
          _LOGGER.error(consoleUrl)
>         raise DataflowRuntimeException(
              'Dataflow pipeline failed. State: %s, Error:\n%s' %
              (self.state, getattr(self._runner, 'last_error_msg', None)),
E             apache_beam.runners.dataflow.dataflow_runner.DataflowRuntimeException: Dataflow pipeline failed. State: FAILED, Error:
E             Workflow failed.

apache_beam/runners/dataflow/dataflow_runner.py:834: DataflowRuntimeException

google.api_core.exceptions.DeadlineExceeded: 504 Deadline Exceeded
args = (name: "projects/apache-beam-testing/subscriptions/psit_subscription_input62c4f98a-e038-426f-a859-91fdc5e7b460"
topic: "projects/apache-beam-testing/topics/psit_topic_input62c4f98a-e038-426f-a859-91fdc5e7b460"
,)
kwargs = {'metadata': [('x-goog-api-client', 'gl-python/3.14.7 grpc/1.83.0 gax/2.34.0 gccl/2.39.2 pb/6.33.6'), ('x-goog-request...ects/apache-beam-testing/subscriptions/psit_subscription_input62c4f98a-e038-426f-a859-91fdc5e7b460')], 'timeout': 60.0}

    @functools.wraps(callable_)
    def error_remapped_callable(*args, **kwargs):
        try:
>           return callable_(*args, **kwargs)
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^

../../build/gradleenv/417525527/lib/python3.14/site-packages/google/api_core/grpc_helpers.py:55: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../build/gradleenv/417525527/lib/python3.14/site-packages/grpc/_interceptor.py:276: in __call__
    response, ignored_call = self._with_call(
../../build/gradleenv/417525527/lib/python3.14/site-packages/grpc/_interceptor.py:331: in _with_call
    return call.result(), call
           ^^^^^^^^^^^^^
../../build/gradleenv/417525527/lib/python3.14/site-packages/grpc/_channel.py:447: in result
    raise self
../../build/gradleenv/417525527/lib/python3.14/site-packages/grpc/_interceptor.py:314: in continuation
    response, call = self._thunk(new_method).with_call(
../../build/gradleenv/417525527/lib/python3.14/site-packages/grpc/_channel.py:1182: in with_call
    return _end_unary_response_blocking(state, call, True, None)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

state = <grpc._channel._RPCState object at 0x7e0c881dd8b0>
call = <grpc._cython.cygrpc.SegregatedCall object at 0x7e0c8824ad40>
with_call = True, deadline = None

    def _end_unary_response_blocking(
        state: _RPCState,
        call: cygrpc.SegregatedCall,
        with_call: bool,
        deadline: Optional[float],
    ) -> Union[ResponseType, Tuple[ResponseType, grpc.Call]]:
        if state.code is grpc.StatusCode.OK:
            if with_call:
                rendezvous = _MultiThreadedRendezvous(state, call, None, deadline)
                return state.response, rendezvous
            return state.response
>       raise _InactiveRpcError(state)  # pytype: disable=not-instantiable
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E       grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
E       	status = StatusCode.DEADLINE_EXCEEDED
E       	details = "Deadline Exceeded"
E       	debug_error_string = "DEADLINE_EXCEEDED:Deadline Exceeded"
E       >

../../build/gradleenv/417525527/lib/python3.14/site-packages/grpc/_channel.py:999: _InactiveRpcError

The above exception was the direct cause of the following exception:

self = <apache_beam.io.gcp.pubsub_integration_test.PubSubIntegrationTest testMethod=test_streaming_with_attributes>

    def setUp(self):
      self.test_pipeline = TestPipeline(is_integration_test=True)
      self.runner_name = type(self.test_pipeline.runner).__name__
      self.project = self.test_pipeline.get_option('project')
      self.uuid = str(uuid.uuid4())
    
      # Set up PubSub environment.
      from google.cloud import pubsub
      self.pub_client = pubsub.PublisherClient()
      self.input_topic = self.pub_client.create_topic(
          name=self.pub_client.topic_path(self.project, INPUT_TOPIC + self.uuid))
      self.output_topic = self.pub_client.create_topic(
          name=self.pub_client.topic_path(self.project, OUTPUT_TOPIC + self.uuid))
    
      self.sub_client = pubsub.SubscriberClient()
>     self.input_sub = self.sub_client.create_subscription(
          name=self.sub_client.subscription_path(
              self.project, INPUT_SUB + self.uuid),
          topic=self.input_topic.name)

apache_beam/io/gcp/pubsub_integration_test.py:150: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../build/gradleenv/417525527/lib/python3.14/site-packages/google/pubsub_v1/services/subscriber/client.py:989: in create_subscription
    response = rpc(
../../build/gradleenv/417525527/lib/python3.14/site-packages/google/api_core/gapic_v1/method.py:189: in __call__
    return wrapped_func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
../../build/gradleenv/417525527/lib/python3.14/site-packages/google/api_core/retry/retry_unary.py:295: in retry_wrapped_func
    return retry_target(
../../build/gradleenv/417525527/lib/python3.14/site-packages/google/api_core/retry/retry_unary.py:157: in retry_target
    next_sleep = _retry_error_helper(
../../build/gradleenv/417525527/lib/python3.14/site-packages/google/api_core/retry/retry_base.py:215: in _retry_error_helper
    raise final_exc from source_exc
../../build/gradleenv/417525527/lib/python3.14/site-packages/google/api_core/retry/retry_unary.py:148: in retry_target
    result = target()
             ^^^^^^^^
../../build/gradleenv/417525527/lib/python3.14/site-packages/google/api_core/timeout.py:130: in func_with_timeout
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (name: "projects/apache-beam-testing/subscriptions/psit_subscription_input62c4f98a-e038-426f-a859-91fdc5e7b460"
topic: "projects/apache-beam-testing/topics/psit_topic_input62c4f98a-e038-426f-a859-91fdc5e7b460"
,)
kwargs = {'metadata': [('x-goog-api-client', 'gl-python/3.14.7 grpc/1.83.0 gax/2.34.0 gccl/2.39.2 pb/6.33.6'), ('x-goog-request...ects/apache-beam-testing/subscriptions/psit_subscription_input62c4f98a-e038-426f-a859-91fdc5e7b460')], 'timeout': 60.0}

    @functools.wraps(callable_)
    def error_remapped_callable(*args, **kwargs):
        try:
            return callable_(*args, **kwargs)
        except grpc.RpcError as exc:
>           raise exceptions.from_grpc_error(exc) from exc
E           google.api_core.exceptions.DeadlineExceeded: 504 Deadline Exceeded

../../build/gradleenv/417525527/lib/python3.14/site-packages/google/api_core/grpc_helpers.py:57: DeadlineExceeded