def job_logs(self, job, delay=5):
    """Job log generator.

    :param job: job name
    :param delay: time in seconds between each server poll

    Yields line by line.

    """
    finishing = False
    offset = 0
    while True:
      try:
        logs = self._session.get_job_logs(
          exec_id=self.exec_id,
          job=job,
          offset=offset,
        )
      except HTTPError as err:
        # if Azkaban is hanging, the job might be stuck in preparing stage
        preparing = False
        while True:
          sleep(delay)
          preparing_jobs = set(
            e['id']
            for e in self.status['nodes']
            if e['status'] == 'PREPARING'
          )
          if job in preparing_jobs:
            if not preparing:
              preparing = True
              _logger.debug(
                'Job %s in execution %s is still preparing.', job, self.exec_id
              )
          else:
            break
        if not preparing:
          # something else is causing the error
          raise err
      else:
        if logs['length']:
          offset += logs['length']
          lines = (e for e in logs['data'].split('\n') if e)
          for line in lines:
            yield line
        elif finishing:
          break
        else:
          running_jobs = set(
            e['id']
            for e in self.status['nodes']
            if e['status'] == 'RUNNING'
          )
          if running_jobs not in job:
            finishing = True
      sleep(delay)
dataset/ETHPy150Open mtth/azkaban/azkaban/remote.py/Execution.job_logs/swapped operands of `not in`