Occasional RPC DEADLINE_EXCEEDED with PubSub source on Python/DirectRunner

[Pradip Thachile <pr...@thachile.com>]

Hi,

I'm testing a pipeline using Python Beam and the DirectRunner that reads from a Pub/Sub subscription however after "some time" (i.e. I don't quite see a predictable pattern yet) I get a bundle failure exception that flags a failure with a gRPC call to Pub/Sub:

ERROR:apache_beam.runners.direct.executor:Exception at bundle <apache_beam.runners.direct.bundle_factory._Bundle object at 0x120aaf550>, due to an exception.
 Traceback (most recent call last):
  File "<redacted venv path>/lib/python3.7/site-packages/google/api_core/grpc_helpers.py", line 57, in error_remapped_callable
    return callable_(*args, **kwargs)
  File "<redacted venv path>/lib/python3.7/site-packages/grpc/_channel.py", line 826, in __call__
    return _end_unary_response_blocking(state, call, False, None)
  File "<redacted venv path>/lib/python3.7/site-packages/grpc/_channel.py", line 729, in _end_unary_response_blocking
    raise _InactiveRpcError(state)
grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
    status = StatusCode.DEADLINE_EXCEEDED
    details = "Deadline Exceeded"
    debug_error_string = "{"created":"@1591255982.041826000","description":"Deadline Exceeded","file":"src/core/ext/filters/deadline/deadline_filter.cc","file_line":69,"grpc_status":4}"

This then causes a number of other knock-on exceptions (shown farther below). When this happens, looking at the subscription its clear that the number of unacked/old messages both start increasing and there's a long ensuing delay until messages are output by downstream GBK/Combine operations. Not really sure how to begin debugging this and would love to get the communities feedback.

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "<redacted venv path>/lib/python3.7/site-packages/apache_beam/runners/direct/executor.py", line 381, in call
    finish_state)
  File "<redacted venv path>/lib/python3.7/site-packages/apache_beam/runners/direct/executor.py", line 421, in attempt_call
    result = evaluator.finish_bundle()
  File "<redacted venv path>/lib/python3.7/site-packages/apache_beam/runners/direct/transform_evaluator.py", line 574, in finish_bundle
    data = self._read_from_pubsub(self.source.timestamp_attribute)
  File "<redacted venv path>/lib/python3.7/site-packages/apache_beam/runners/direct/transform_evaluator.py", line 562, in _read_from_pubsub
    return_immediately=True)
  File "<redacted venv path>/lib/python3.7/site-packages/google/cloud/pubsub_v1/_gapic.py", line 40, in <lambda>
    fx = lambda self, *a, **kw: wrapped_fx(self.api, *a, **kw)  # noqa
  File "<redacted venv path>/lib/python3.7/site-packages/google/cloud/pubsub_v1/gapic/subscriber_client.py", line 1005, in pull
    request, retry=retry, timeout=timeout, metadata=metadata
  File "<redacted venv path>/lib/python3.7/site-packages/google/api_core/gapic_v1/method.py", line 143, in __call__
    return wrapped_func(*args, **kwargs)
  File "<redacted venv path>/lib/python3.7/site-packages/google/api_core/retry.py", line 286, in retry_wrapped_func
    on_error=on_error,
  File "<redacted venv path>/lib/python3.7/site-packages/google/api_core/retry.py", line 184, in retry_target
    return target()
  File "<redacted venv path>/lib/python3.7/site-packages/google/api_core/timeout.py", line 214, in func_with_timeout
    return func(*args, **kwargs)
  File "<redacted venv path>/lib/python3.7/site-packages/google/api_core/grpc_helpers.py", line 59, in error_remapped_callable
    six.raise_from(exceptions.from_grpc_error(exc), exc)
  File "<string>", line 3, in raise_from
google.api_core.exceptions.DeadlineExceeded: 504 Deadline Exceeded

ERROR:grpc._plugin_wrapping:AuthMetadataPluginCallback "<google.auth.transport.grpc.AuthMetadataPlugin object at 0x12093c1d0>" raised exception!
Traceback (most recent call last):
  File "<redacted venv path>/lib/python3.7/site-packages/urllib3/connection.py", line 160, in _new_conn
    (self._dns_host, self.port), self.timeout, **extra_kw
  File "<redacted venv path>/lib/python3.7/site-packages/urllib3/util/connection.py", line 84, in create_connection
    raise err
  File "<redacted venv path>/lib/python3.7/site-packages/urllib3/util/connection.py", line 74, in create_connection
    sock.connect(sa)
TimeoutError: [Errno 60] Operation timed out

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<redacted venv path>/lib/python3.7/site-packages/urllib3/connectionpool.py", line 677, in urlopen
    chunked=chunked,
  File "<redacted venv path>/lib/python3.7/site-packages/urllib3/connectionpool.py", line 381, in _make_request
    self._validate_conn(conn)
  File "<redacted venv path>/lib/python3.7/site-packages/urllib3/connectionpool.py", line 976, in _validate_conn
    conn.connect()
  File "<redacted venv path>/lib/python3.7/site-packages/urllib3/connection.py", line 308, in connect
    conn = self._new_conn()
  File "<redacted venv path>/lib/python3.7/site-packages/urllib3/connection.py", line 172, in _new_conn
    self, "Failed to establish a new connection: %s" % e
urllib3.exceptions.NewConnectionError: <urllib3.connection.HTTPSConnection object at 0x120257d90>: Failed to establish a new connection: [Errno 60] Operation timed out

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<redacted venv path>/lib/python3.7/site-packages/requests/adapters.py", line 449, in send
    timeout=timeout
  File "<redacted venv path>/lib/python3.7/site-packages/urllib3/connectionpool.py", line 725, in urlopen
    method, url, error=e, _pool=self, _stacktrace=sys.exc_info()[2]
  File "<redacted venv path>/lib/python3.7/site-packages/urllib3/util/retry.py", line 439, in increment
    raise MaxRetryError(_pool, url, error or ResponseError(cause))
urllib3.exceptions.MaxRetryError: HTTPSConnectionPool(host='oauth2.googleapis.com', port=443): Max retries exceeded with url: /token (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x120257d90>: Failed to establish a new connection: [Errno 60] Operation timed out'))

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<redacted venv path>/lib/python3.7/site-packages/google/auth/transport/requests.py", line 181, in __call__
    method, url, data=body, headers=headers, timeout=timeout, **kwargs
  File "<redacted venv path>/lib/python3.7/site-packages/requests/sessions.py", line 530, in request
    resp = self.send(prep, **send_kwargs)
  File "<redacted venv path>/lib/python3.7/site-packages/requests/sessions.py", line 643, in send
    r = adapter.send(request, **kwargs)
  File "<redacted venv path>/lib/python3.7/site-packages/requests/adapters.py", line 516, in send
    raise ConnectionError(e, request=request)
requests.exceptions.ConnectionError: HTTPSConnectionPool(host='oauth2.googleapis.com', port=443): Max retries exceeded with url: /token (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x120257d90>: Failed to establish a new connection: [Errno 60] Operation timed out'))

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "<redacted venv path>/lib/python3.7/site-packages/grpc/_plugin_wrapping.py", line 78, in __call__
    context, _AuthMetadataPluginCallback(callback_state, callback))
  File "<redacted venv path>/lib/python3.7/site-packages/google/auth/transport/grpc.py", line 84, in __call__
    callback(self._get_authorization_headers(context), None)
  File "<redacted venv path>/lib/python3.7/site-packages/google/auth/transport/grpc.py", line 71, in _get_authorization_headers
    self._request, context.method_name, context.service_url, headers
  File "<redacted venv path>/lib/python3.7/site-packages/google/auth/credentials.py", line 124, in before_request
    self.refresh(request)
  File "<redacted venv path>/lib/python3.7/site-packages/google/oauth2/service_account.py", line 334, in refresh
    access_token, expiry, _ = _client.jwt_grant(request, self._token_uri, assertion)
  File "<redacted venv path>/lib/python3.7/site-packages/google/oauth2/_client.py", line 153, in jwt_grant
    response_data = _token_endpoint_request(request, token_uri, body)
  File "<redacted venv path>/lib/python3.7/site-packages/google/oauth2/_client.py", line 105, in _token_endpoint_request
    response = request(method="POST", url=token_uri, headers=headers, body=body)
  File "<redacted venv path>/lib/python3.7/site-packages/google/auth/transport/requests.py", line 186, in __call__
    six.raise_from(new_exc, caught_exc)
  File "<string>", line 3, in raise_from
google.auth.exceptions.TransportError: HTTPSConnectionPool(host='oauth2.googleapis.com', port=443): Max retries exceeded with url: /token (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x120257d90>: Failed to establish a new connection: [Errno 60] Operation timed out'))

-Pradip