def process_computation(self, computation_work):
    computation_id = computation_work.computation_id
    input_data_watermark = windmillio.windmill_to_harness_timestamp(
        computation_work.input_data_watermark)
    if computation_id not in self.instruction_map:
      self.get_config(computation_id)
    map_task_proto = self.instruction_map[computation_id]
    for work_item in computation_work.work:
      retry_locally = True
      while retry_locally:
        try:
          self.process_work_item(computation_id, map_task_proto,
                                 input_data_watermark, work_item)
          break
        except:  # pylint: disable=bare-except
          logging.error(
              'Exception while processing work item for computation %r: '
              '%s, %s', computation_id, work_item, traceback.format_exc())

          # Send exception details to Windmill, retry locally if possible.
          retry_locally = self.report_failure(computation_id, work_item,
                                              sys.exc_info())

          # TODO(ccy): handle token expiration in retry logic.
          # TODO(ccy): handle out-of-memory error in retry logic.
          if retry_locally:
            logging.error('Execution of work in computation %s for key %r '
                          'failed; will retry locally.', computation_id,
                          work_item.key)
            time.sleep(StreamingWorker.RETRY_LOCALLY_DELAY)
          else:
            logging.error('Execution of work in computation %s for key %r '
                          'failed; Windmill indicated to not retry '
                          'locally.', computation_id, work_item.key)
dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/worker/streamingworker.py/StreamingWorker.process_computation/original