Merge branch 'main' into otel_publish

Author: mukund-ananthu

.release-please-manifest.json

@@ -1,4 +1,4 @@
 {
- ".": "2.23.0"
+ ".": "2.23.1"
 }
 

CHANGELOG.md

@@ -5,6 +5,13 @@
 [1]: https://pypi.org/project/google-cloud-pubsub/#history
 
 
+## [2.23.1](https://github.com/googleapis/python-pubsub/compare/v2.23.0...v2.23.1) (2024-09-09)
+
+
+### Bug Fixes
+
+* Replace asserts with None checks for graceful shutdown ([#1244](https://github.com/googleapis/python-pubsub/issues/1244)) ([ced4f52](https://github.com/googleapis/python-pubsub/commit/ced4f527c7f918a87d1b89c2b5da59dbdf00e2c3))
+
 ## [2.23.0](https://github.com/googleapis/python-pubsub/compare/v2.22.0...v2.23.0) (2024-07-29)
 
 

google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

@@ -1104,8 +1104,11 @@ def _on_response(self, response: gapic_types.StreamingPullResponse) -> None:
  )
 
  with self._pause_resume_lock:
- assert self._scheduler is not None
- assert self._leaser is not None
+ if self._scheduler is None or self._leaser is None:
+ _LOGGER.debug(
+ f"self._scheduler={self._scheduler} or self._leaser={self._leaser} is None. Stopping further processing."
+ )
+ return
 
  for received_message in received_messages:
  if (

google/pubsub/gapic_version.py

@@ -13,4 +13,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-__version__ = "2.23.0" # {x-release-please-version}
+__version__ = "2.23.1" # {x-release-please-version}

google/pubsub_v1/gapic_version.py

@@ -13,4 +13,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-__version__ = "2.23.0" # {x-release-please-version}
+__version__ = "2.23.1" # {x-release-please-version}

samples/generated_samples/snippet_metadata_google.pubsub.v1.json

@@ -8,7 +8,7 @@
  ],
  "language": "PYTHON",
  "name": "google-cloud-pubsub",
- "version": "0.1.0"
+ "version": "2.23.1"
  },
  "snippets": [
  {

samples/snippets/requirements-test.txt

@@ -1,7 +1,7 @@
 backoff==2.2.1
 pytest===7.4.4; python_version == '3.7'
-pytest==8.2.2; python_version >= '3.8'
+pytest==8.3.2; python_version >= '3.8'
 mock==5.1.0
 flaky==3.8.1
 google-cloud-bigquery==3.25.0
-google-cloud-storage==2.17.0
+google-cloud-storage==2.18.2

samples/snippets/requirements.txt

@@ -1,5 +1,5 @@
-google-cloud-pubsub==2.22.0
-avro==1.11.3
+google-cloud-pubsub==2.23.0
+avro==1.12.0
 protobuf===4.24.4; python_version == '3.7'
-protobuf==5.27.2; python_version >= '3.8'
-avro==1.11.3
+protobuf==5.28.0; python_version >= '3.8'
+avro==1.12.0

tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py

@@ -1375,6 +1375,62 @@ def test_close_blocking_scheduler_shutdown():
  scheduler.shutdown.assert_called_once_with(await_msg_callbacks=True)
 
 
+def test__on_response_none_scheduler():
+ manager, _, _, _, _, _ = make_running_manager()
+
+ manager._callback = mock.sentinel.callback
+ manager._scheduler = None
+ # Set up the messages.
+ response = gapic_types.StreamingPullResponse(
+ received_messages=[
+ gapic_types.ReceivedMessage(
+ ack_id="ack1",
+ message=gapic_types.PubsubMessage(data=b"foo", message_id="1"),
+ ),
+ gapic_types.ReceivedMessage(
+ ack_id="ack2",
+ message=gapic_types.PubsubMessage(data=b"bar", message_id="2"),
+ delivery_attempt=6,
+ ),
+ ]
+ )
+
+ manager._maybe_release_messages = mock.Mock()
+
+ # adjust message bookkeeping in leaser
+ fake_leaser_add(leaser, init_msg_count=0, assumed_msg_size=42)
+ manager._on_response(response)
+
+ manager._maybe_release_messages.assert_not_called
+
+
+def test__on_response_none_leaser():
+ manager, _, _, _, _, _ = make_running_manager()
+
+ manager._callback = mock.sentinel.callback
+ manager._leaser = None
+ # Set up the messages.
+ response = gapic_types.StreamingPullResponse(
+ received_messages=[
+ gapic_types.ReceivedMessage(
+ ack_id="ack1",
+ message=gapic_types.PubsubMessage(data=b"foo", message_id="1"),
+ ),
+ gapic_types.ReceivedMessage(
+ ack_id="ack2",
+ message=gapic_types.PubsubMessage(data=b"bar", message_id="2"),
+ delivery_attempt=6,
+ ),
+ ]
+ )
+
+ manager._maybe_release_messages = mock.Mock()
+
+ manager._on_response(response)
+
+ manager._maybe_release_messages.assert_not_called
+
+
 def test_close_nonblocking_scheduler_shutdown():
  manager, _, _, _, _, _ = make_running_manager(await_callbacks_on_shutdown=False)
  scheduler = manager._scheduler