Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat: retry temporary GRPC statuses for ack/modack/nack when exactly-once delivery is enabled #607

Merged
merged 4 commits into from
Mar 8, 2022
Merged

feat: retry temporary GRPC statuses for ack/modack/nack when exactly-once delivery is enabled #607

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0bae26c
Retry temporary GRPC statuses for ack/modack/nack when exactly-once d…
pradn Mar 8, 2022
ef8f8ba
Add more tests.
pradn Mar 8, 2022
eb3c4d2
Address Mike and Tianzi's comments
pradn Mar 8, 2022
290721e
Run linter
pradn Mar 8, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 29 additions & 4 deletions google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,14 @@
a subscription. We do this to reduce premature ack expiration.
"""

_EXACTLY_ONCE_DELIVERY_TEMPORARY_RETRY_ERRORS = [
pradn marked this conversation as resolved.
Show resolved Hide resolved
code_pb2.DEADLINE_EXCEEDED,
code_pb2.RESOURCE_EXHAUSTED,
code_pb2.ABORTED,
code_pb2.INTERNAL,
code_pb2.UNAVAILABLE,
]


def _wrap_as_exception(maybe_exception: Any) -> BaseException:
"""Wrap an object as a Python exception, if needed.
Expand Down Expand Up @@ -163,6 +171,8 @@ def _process_requests(
requests_completed = []
requests_to_retry = []
for ack_id in ack_reqs_dict:
# Handle special errors returned for ack/modack RPCs via the ErrorInfo
# sidecar metadata when exactly-once delivery is enabled.
if errors_dict and ack_id in errors_dict:
exactly_once_error = errors_dict[ack_id]
if exactly_once_error.startswith("TRANSIENT_"):
Expand All @@ -176,9 +186,18 @@ def _process_requests(
future = ack_reqs_dict[ack_id].future
future.set_exception(exc)
requests_completed.append(ack_reqs_dict[ack_id])
# Temporary GRPC errors are retried
elif (
error_status
and error_status.code in _EXACTLY_ONCE_DELIVERY_TEMPORARY_RETRY_ERRORS
):
import traceback

traceback.print_stack()
print("retrying temp error: " + str(error_status))
requests_to_retry.append(ack_reqs_dict[ack_id])
# Other GRPC errors are NOT retried
elif error_status:
# Only permanent errors are expected here b/c retriable errors are
# retried at the lower, GRPC level.
if error_status.code == code_pb2.PERMISSION_DENIED:
exc = AcknowledgeError(AcknowledgeStatus.PERMISSION_DENIED, info=None)
elif error_status.code == code_pb2.FAILED_PRECONDITION:
Expand All @@ -188,11 +207,13 @@ def _process_requests(
future = ack_reqs_dict[ack_id].future
future.set_exception(exc)
requests_completed.append(ack_reqs_dict[ack_id])
# Since no error occurred, requests with futures are completed successfully.
elif ack_reqs_dict[ack_id].future:
future = ack_reqs_dict[ack_id].future
# success
future.set_result(AcknowledgeStatus.SUCCESS)
requests_completed.append(ack_reqs_dict[ack_id])
# All other requests are considered completed.
else:
requests_completed.append(ack_reqs_dict[ack_id])

Expand Down Expand Up @@ -580,7 +601,9 @@ def send_unary_ack(
ack_errors_dict = _get_ack_errors(exc)
except exceptions.RetryError as exc:
status = status_pb2.Status()
status.code = code_pb2.DEADLINE_EXCEEDED
# Choose a non-retriable error code so the futures fail with
# exceptions.
status.code = code_pb2.UNKNOWN
# Makes sure to complete futures so they don't block forever.
_process_requests(status, ack_reqs_dict, None)
_LOGGER.debug(
Expand Down Expand Up @@ -634,7 +657,9 @@ def send_unary_modack(
modack_errors_dict = _get_ack_errors(exc)
except exceptions.RetryError as exc:
status = status_pb2.Status()
status.code = code_pb2.DEADLINE_EXCEEDED
# Choose a non-retriable error code so the futures fail with
# exceptions.
status.code = code_pb2.UNKNOWN
# Makes sure to complete futures so they don't block forever.
_process_requests(status, ack_reqs_dict, None)
_LOGGER.debug(
Expand Down
1 change: 1 addition & 0 deletions noxfile.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,6 +156,7 @@ def default(session):
"--cov-config=.coveragerc",
"--cov-report=",
"--cov-fail-under=0",
"-s",
pradn marked this conversation as resolved.
Show resolved Hide resolved
os.path.join("tests", "unit"),
*session.posargs,
)
Expand Down