apache_beam.runners.dataflow.dataflow_runner.DataflowRuntimeException: Dataflow pipeline failed. State: FAILED, Error:
None
self = <apache_beam.transforms.managed_iceberg_it_test.ManagedIcebergIT testMethod=test_write_read_pipeline>

    def test_write_read_pipeline(self):
      iceberg_config = {
          "table": "test_iceberg_write_read.test_" + str(int(time.time())),
          "catalog_name": "default",
          "catalog_properties": {
              "type": "hadoop",
              "warehouse": self.WAREHOUSE,
          }
      }
    
      rows = [self._create_row(i) for i in range(100)]
      expected_dicts = [row.as_dict() for row in rows]
    
      with beam.Pipeline(argv=self.args) as write_pipeline:
>       _ = (
            write_pipeline
            | beam.Create(rows)
            | beam.managed.Write(beam.managed.ICEBERG, config=iceberg_config))

apache_beam/transforms/managed_iceberg_it_test.py:68: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
apache_beam/pipeline.py:620: in __exit__
    self.result = self.run()
apache_beam/pipeline.py:594: in run
    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 <Job
 clientRequestId: '20241218055550270621-4304'
 createTime: '2024-12-18T05:55:55.698506Z'
...024-12-18T05:55:55.698506Z'
 steps: []
 tempFiles: []
 type: TypeValueValuesEnum(JOB_TYPE_BATCH, 1)> at 0x7c86f1f60640>
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.')
        consoleUrl = (
            "Console URL: https://console.cloud.google.com/"
            f"dataflow/jobs/<RegionId>/{self.job_id()}"
            "?project=<ProjectId>")
        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)),
              self)
      elif PipelineState.is_terminal(
          self.state) and self.state == PipelineState.FAILED and self._runner:
>       raise DataflowRuntimeException(
            'Dataflow pipeline failed. State: %s, Error:\n%s' %
            (self.state, getattr(self._runner, 'last_error_msg', None)),
            self)
E       apache_beam.runners.dataflow.dataflow_runner.DataflowRuntimeException: Dataflow pipeline failed. State: FAILED, Error:
E       None

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