Apache Airflow version: 1.10.12

What happened:

If any Python code run in an operator, has imported absl.logging (directly or indirectly), airflow run on that task ends with AttributeError: 'StreamLogWriter' object has no attribute 'close'. This is not normally seen, as this only happens in the --raw inner stage. The task is marked as successful, but the task exited with exit code 1.

The full traceback is:

Traceback (most recent call last):
  File "/.../bin/airflow", line 37, in <module>
    args.func(args)
  File "/.../lib/python3.6/site-packages/airflow/utils/cli.py", line 76, in wrapper
    return f(*args, **kwargs)
  File "/.../lib/python3.6/site-packages/airflow/bin/cli.py", line 588, in run
    logging.shutdown()
  File "/.../lib/python3.6/logging/__init__.py", line 1946, in shutdown
    h.close()
  File "/.../lib/python3.6/site-packages/absl/logging/__init__.py", line 864, in close
    self.stream.close()
AttributeError: 'StreamLogWriter' object has no attribute 'close'
Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "/.../lib/python3.6/logging/__init__.py", line 1946, in shutdown
    h.close()
  File "/.../lib/python3.6/site-packages/absl/logging/__init__.py", line 864, in close
    self.stream.close()
AttributeError: 'StreamLogWriter' object has no attribute 'close'

Abseil is Google's utility package, and is used in Tensorflow. The same issue would be seen if you used import tensorflow.

What you expected to happen:

I expected the task to exit with exit code 0, without issues at exit.

How to reproduce it:

• Install abseil: pip install absl-py
• Create a test dag with a single operator that only uses import abs.logging
• Trigger a dagrun (no scheduler needs to be running)
• execute airflow run [dagid] [taskid] [execution-date] --raw

Anything else we need to know:

What happens is that absl.logging sets up a logger with a custom handler (absl.logging.ABSLHandler(), which is a proxy for either absl.logging.PythonHandler()), and that proxy will call .close() on its stream. By default that stream is sys.stderr. However, airflow has swapped out sys.stderr for a StreamLogWriter object when it runs the task under the airflow.utils.log.logging_mixin.redirect_stderr context manager. When the context manager exits, the logger is still holding on to the surrogate stderr object.

Normally, the abseil handler would handle such cases, it deliberately won't close a stream that is the same object as sys.stderr or sys.__stderr__ (see their source code). But at exit time that's no longer the case. At that time logging.shutdown() is called, and that leads to the above exception.

Since sys.stderr has a close method, the best fix would be for StreamLogWriter to also have one.