diff --git a/.github/.OwlBot.lock.yaml b/.github/.OwlBot.lock.yaml index 480226ac0..44c78f7cc 100644 --- a/.github/.OwlBot.lock.yaml +++ b/.github/.OwlBot.lock.yaml @@ -13,4 +13,4 @@ # limitations under the License. docker: image: gcr.io/cloud-devrel-public-resources/owlbot-python:latest - digest: sha256:6162c384d685c5fe22521d3f37f6fc732bf99a085f6d47b677dbcae97fc21392 + digest: sha256:4e1991042fe54b991db9ca17c8fb386e61b22fe4d1472a568bf0fcac85dcf5d3 diff --git a/.github/workflows/unittest.yml b/.github/workflows/unittest.yml index e87fe5b7b..e5be6edbd 100644 --- a/.github/workflows/unittest.yml +++ b/.github/workflows/unittest.yml @@ -26,7 +26,7 @@ jobs: run: | nox -s unit-${{ matrix.python }} - name: Upload coverage results - uses: actions/upload-artifact@v2 + uses: actions/upload-artifact@v3 with: name: coverage-artifacts path: .coverage-${{ matrix.python }} @@ -47,7 +47,7 @@ jobs: python -m pip install --upgrade setuptools pip wheel python -m pip install coverage - name: Download coverage results - uses: actions/download-artifact@v2 + uses: actions/download-artifact@v3 with: name: coverage-artifacts path: .coverage-results/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 57732bacf..bac55a5dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,20 @@ [1]: https://pypi.org/project/google-cloud-pubsub/#history +## [2.11.0](https://github.com/googleapis/python-pubsub/compare/v2.10.0...v2.11.0) (2022-03-09) + + +### Features + +* retry temporary GRPC statuses for ack/modack/nack when exactly-once delivery is enabled ([#607](https://github.com/googleapis/python-pubsub/issues/607)) ([a91bed8](https://github.com/googleapis/python-pubsub/commit/a91bed829c9040fcc6c1e70b99b66188ac4ded40)) +* return singleton success future for exactly-once methods in Message ([#608](https://github.com/googleapis/python-pubsub/issues/608)) ([253ced2](https://github.com/googleapis/python-pubsub/commit/253ced28f308450c7a1a93cc38f6d101ecd7d4c0)) + + +### Bug Fixes + +* **deps:** require google-api-core>=1.31.5, >=2.3.2 ([#600](https://github.com/googleapis/python-pubsub/issues/600)) ([1608b7f](https://github.com/googleapis/python-pubsub/commit/1608b7ffdd5b5db87e1e55fde763440ca9a4086e)) +* **deps:** require proto-plus>=1.15.0 ([1608b7f](https://github.com/googleapis/python-pubsub/commit/1608b7ffdd5b5db87e1e55fde763440ca9a4086e)) + ## [2.10.0](https://github.com/googleapis/python-pubsub/compare/v2.9.0...v2.10.0) (2022-03-04) diff --git a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py index 5a9d08026..e098491fe 100644 --- a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py +++ b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py @@ -75,6 +75,14 @@ a subscription. We do this to reduce premature ack expiration. """ +_EXACTLY_ONCE_DELIVERY_TEMPORARY_RETRY_ERRORS = { + 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. @@ -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_"): @@ -176,9 +186,14 @@ 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 + ): + 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: @@ -188,11 +203,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]) @@ -580,7 +597,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( @@ -634,7 +653,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( diff --git a/google/cloud/pubsub_v1/subscriber/message.py b/google/cloud/pubsub_v1/subscriber/message.py index 5744aa71c..ab17bab78 100644 --- a/google/cloud/pubsub_v1/subscriber/message.py +++ b/google/cloud/pubsub_v1/subscriber/message.py @@ -40,6 +40,9 @@ attributes: {} }}""" +_SUCCESS_FUTURE = futures.Future() +_SUCCESS_FUTURE.set_result(AcknowledgeStatus.SUCCESS) + def _indent(lines: str, prefix: str = " ") -> str: """Indent some text. @@ -291,12 +294,13 @@ def ack_with_response(self) -> "futures.Future": pubsub_v1.subscriber.exceptions.AcknowledgeError exception will be thrown. """ - future = futures.Future() - req_future = None + req_future: Optional[futures.Future] if self._exactly_once_delivery_enabled_func(): + future = futures.Future() req_future = future else: - future.set_result(AcknowledgeStatus.SUCCESS) + future = _SUCCESS_FUTURE + req_future = None time_to_ack = math.ceil(time.time() - self._received_timestamp) self._request_queue.put( requests.AckRequest( @@ -390,12 +394,13 @@ def modify_ack_deadline_with_response(self, seconds: int) -> "futures.Future": will be thrown. """ - future = futures.Future() - req_future = None + req_future: Optional[futures.Future] if self._exactly_once_delivery_enabled_func(): + future = futures.Future() req_future = future else: - future.set_result(AcknowledgeStatus.SUCCESS) + future = _SUCCESS_FUTURE + req_future = None self._request_queue.put( requests.ModAckRequest( @@ -451,12 +456,13 @@ def nack_with_response(self) -> "futures.Future": will be thrown. """ - future = futures.Future() - req_future = None + req_future: Optional[futures.Future] if self._exactly_once_delivery_enabled_func(): + future = futures.Future() req_future = future else: - future.set_result(AcknowledgeStatus.SUCCESS) + future = _SUCCESS_FUTURE + req_future = None self._request_queue.put( requests.NackRequest( diff --git a/samples/snippets/noxfile.py b/samples/snippets/noxfile.py index 20cdfc620..4c808af73 100644 --- a/samples/snippets/noxfile.py +++ b/samples/snippets/noxfile.py @@ -188,42 +188,54 @@ def _session_tests( # check for presence of tests test_list = glob.glob("*_test.py") + glob.glob("test_*.py") test_list.extend(glob.glob("tests")) + if len(test_list) == 0: print("No tests found, skipping directory.") - else: - if TEST_CONFIG["pip_version_override"]: - pip_version = TEST_CONFIG["pip_version_override"] - session.install(f"pip=={pip_version}") - """Runs py.test for a particular project.""" - if os.path.exists("requirements.txt"): - if os.path.exists("constraints.txt"): - session.install("-r", "requirements.txt", "-c", "constraints.txt") - else: - session.install("-r", "requirements.txt") - - if os.path.exists("requirements-test.txt"): - if os.path.exists("constraints-test.txt"): - session.install( - "-r", "requirements-test.txt", "-c", "constraints-test.txt" - ) - else: - session.install("-r", "requirements-test.txt") - - if INSTALL_LIBRARY_FROM_SOURCE: - session.install("-e", _get_repo_root()) - - if post_install: - post_install(session) - - session.run( - "pytest", - *(PYTEST_COMMON_ARGS + session.posargs), - # Pytest will return 5 when no tests are collected. This can happen - # on travis where slow and flaky tests are excluded. - # See http://doc.pytest.org/en/latest/_modules/_pytest/main.html - success_codes=[0, 5], - env=get_pytest_env_vars(), - ) + return + + if TEST_CONFIG["pip_version_override"]: + pip_version = TEST_CONFIG["pip_version_override"] + session.install(f"pip=={pip_version}") + """Runs py.test for a particular project.""" + concurrent_args = [] + if os.path.exists("requirements.txt"): + if os.path.exists("constraints.txt"): + session.install("-r", "requirements.txt", "-c", "constraints.txt") + else: + session.install("-r", "requirements.txt") + with open("requirements.txt") as rfile: + packages = rfile.read() + + if os.path.exists("requirements-test.txt"): + if os.path.exists("constraints-test.txt"): + session.install( + "-r", "requirements-test.txt", "-c", "constraints-test.txt" + ) + else: + session.install("-r", "requirements-test.txt") + with open("requirements-test.txt") as rtfile: + packages += rtfile.read() + + if INSTALL_LIBRARY_FROM_SOURCE: + session.install("-e", _get_repo_root()) + + if post_install: + post_install(session) + + if "pytest-parallel" in packages: + concurrent_args.extend(['--workers', 'auto', '--tests-per-worker', 'auto']) + elif "pytest-xdist" in packages: + concurrent_args.extend(['-n', 'auto']) + + session.run( + "pytest", + *(PYTEST_COMMON_ARGS + session.posargs + concurrent_args), + # Pytest will return 5 when no tests are collected. This can happen + # on travis where slow and flaky tests are excluded. + # See http://doc.pytest.org/en/latest/_modules/_pytest/main.html + success_codes=[0, 5], + env=get_pytest_env_vars(), + ) @nox.session(python=ALL_VERSIONS) diff --git a/samples/snippets/requirements.txt b/samples/snippets/requirements.txt index 40078e73f..f47d14979 100644 --- a/samples/snippets/requirements.txt +++ b/samples/snippets/requirements.txt @@ -1,2 +1,2 @@ -google-cloud-pubsub==2.9.0 +google-cloud-pubsub==2.10.0 avro==1.11.0 diff --git a/setup.py b/setup.py index e1b259bd6..8624885b5 100644 --- a/setup.py +++ b/setup.py @@ -22,7 +22,7 @@ name = "google-cloud-pubsub" description = "Google Cloud Pub/Sub API client library" -version = "2.10.0" +version = "2.11.0" # Should be one of: # 'Development Status :: 3 - Alpha' # 'Development Status :: 4 - Beta' @@ -33,8 +33,8 @@ # NOTE: Maintainers, please do not require google-api-core>=2.x.x # Until this issue is closed # https://github.com/googleapis/google-cloud-python/issues/10566 - "google-api-core[grpc] >= 1.28.0, <3.0.0dev", - "proto-plus >= 1.7.1", + "google-api-core[grpc] >= 1.31.5, <3.0.0dev,!=2.0.*,!=2.1.*,!=2.2.*,!=2.3.0", + "proto-plus >= 1.15.0", "grpc-google-iam-v1 >= 0.12.3, < 0.13dev", "grpcio-status >= 1.16.0", ] diff --git a/testing/constraints-3.6.txt b/testing/constraints-3.6.txt index b89267633..0ce29f32c 100644 --- a/testing/constraints-3.6.txt +++ b/testing/constraints-3.6.txt @@ -5,7 +5,7 @@ # e.g., if setup.py has "foo >= 1.14.0, < 2.0.0dev", # Then this file should have foo==1.14.0 grpcio==1.38.1 -google-api-core==1.28.0 +google-api-core==1.31.5 libcst==0.3.10 -proto-plus==1.7.1 +proto-plus==1.15.0 grpc-google-iam-v1==0.12.3 diff --git a/tests/unit/pubsub_v1/subscriber/test_message.py b/tests/unit/pubsub_v1/subscriber/test_message.py index f5c7bf3c7..0debabaf3 100644 --- a/tests/unit/pubsub_v1/subscriber/test_message.py +++ b/tests/unit/pubsub_v1/subscriber/test_message.py @@ -156,6 +156,7 @@ def test_ack_with_response_exactly_once_delivery_disabled(): ) ) assert future.result() == AcknowledgeStatus.SUCCESS + assert future == message._SUCCESS_FUTURE check_call_types(put, requests.AckRequest) @@ -205,6 +206,7 @@ def test_modify_ack_deadline_with_response_exactly_once_delivery_disabled(): requests.ModAckRequest(ack_id="bogus_ack_id", seconds=60, future=None) ) assert future.result() == AcknowledgeStatus.SUCCESS + assert future == message._SUCCESS_FUTURE check_call_types(put, requests.ModAckRequest) @@ -242,6 +244,7 @@ def test_nack_with_response_exactly_once_delivery_disabled(): ) ) assert future.result() == AcknowledgeStatus.SUCCESS + assert future == message._SUCCESS_FUTURE check_call_types(put, requests.NackRequest) diff --git a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py index 9e8d6c5ed..36f82b621 100644 --- a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py +++ b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py @@ -1735,7 +1735,7 @@ def test_process_requests_permanent_error_raises_exception(): assert not requests_to_retry -def test_process_requests_transient_error_returns_request(): +def test_process_requests_transient_error_returns_request_for_retrying(): # a transient error returns the request in `requests_to_retry` future = futures.Future() ack_reqs_dict = { @@ -1772,6 +1772,38 @@ def test_process_requests_unknown_error_raises_exception(): assert not requests_to_retry +def test_process_requests_retriable_error_status_returns_request_for_retrying(): + # a retriable error status returns the request in `requests_to_retry` + retriable_errors = [ + code_pb2.DEADLINE_EXCEEDED, + code_pb2.RESOURCE_EXHAUSTED, + code_pb2.ABORTED, + code_pb2.INTERNAL, + code_pb2.UNAVAILABLE, + ] + + for retriable_error in retriable_errors: + future = futures.Future() + ack_reqs_dict = { + "ackid1": requests.AckRequest( + ack_id="ackid1", + byte_size=0, + time_to_ack=20, + ordering_key="", + future=future, + ) + } + st = status_pb2.Status() + st.code = retriable_error + ( + requests_completed, + requests_to_retry, + ) = streaming_pull_manager._process_requests(st, ack_reqs_dict, None) + assert not requests_completed + assert requests_to_retry[0].ack_id == "ackid1" + assert not future.done() + + def test_process_requests_permission_denied_error_status_raises_exception(): # a permission-denied error status raises an exception future = futures.Future()