title: Manually request a retry in Apache Beam
body: <p>I'm using Apache Beam to do ETLs, and have a problem working with a certain set of report API endpoints.</p>
  <p>The endpoints are designed to not return data immediately when requested, but just report IDs. Then the client is supposed to check the reports' state using the IDs. If a report is <code>COMPLETE</code> then it is downloaded, otherwise if it is <code>UNDER_CONSTRUCT</code> the client is supposed to retry later.</p>
  <p>My attempt is to decorate the <code>process</code> of a <code>DoFn</code> with <code>apache_beam.utils.retry.with_exponential_backoff</code>, and manually raise an exception to make it retry when report is not complete.</p>
  <p>However it doesn't seem to work in that way, and after the exception is raised the program terminates immediately without any wait or retry.</p>
  <p>I made a minimal example below, where the actual report endpoint replaced with some mocks.</p>
  <pre class="lang-py prettyprint-override"><code>from apache_beam import Pipeline, PTransform, DoFn, ParDo
  from apache_beam.utils.retry import with_exponential_backoff
  
  
  class CreateReports(PTransform):
      def expand(self, pvalue):
          return [
              '100000',
              '100001',
              '100002',
          ]
  
  
  class _MockEndpoint:
      def __init__(self):
          self.count = 0
  
      def check_report(self, report_id):
          self.count += 1
          if self.count % 3 != 2:
              return {'status': 'UNDER_CONSTRUCT'}
          return {'status': 'COMPLETE', 'data': (report_id, self.count)}
  
  
  class _ReportNotReady(Exception):
      @classmethod
      def _retry_filter(cls, ex):
          print('Filtering', ex, cls, isinstance(ex, cls))
          return isinstance(ex, cls)
  
  
  class CheckReport(DoFn):
      endpoint = _MockEndpoint()
  
      @with_exponential_backoff(retry_filter=_ReportNotReady._retry_filter)
      def process(self, element):
          response = self.endpoint.check_report(report_id=element)
          if response['status'] == 'COMPLETE':
              yield response['data']
          else:
              raise _ReportNotReady('Retry later')
  
  
  
  with Pipeline() as p:
      m = p | CreateReports() | ParDo(CheckReport())
      print('done', m)
  </code></pre>
  <p>My questions are</p>
  <ol>
  <li>Is there better way to implement the manual retry under certain conditions?</li>
  <li>If <code>retry.with_exponential_backoff</code> is the right solution, any critical arguments missing there so it doesn't work?</li>
  </ol>