try reverting

Author: abbrowne126

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

@@ -60,12 +60,6 @@
 
 
 _LOGGER = logging.getLogger(__name__)
-_SLOW_ACK_LOGGER = logging.getLogger("slow-ack")
-_STREAMS_LOGGER = logging.getLogger("subscriber-streams")
-_FLOW_CONTROL_LOGGER = logging.getLogger("subscriber-flow-control")
-_CALLBACK_DELIVERY_LOGGER = logging.getLogger("callback-delivery")
-_CALLBACK_EXCEPTION_LOGGER = logging.getLogger("callback-exceptions")
-_EXPIRY_LOGGER = logging.getLogger("expiry")
 _REGULAR_SHUTDOWN_THREAD_NAME = "Thread-RegularStreamShutdown"
 _RPC_ERROR_THREAD_NAME = "Thread-OnRpcTerminated"
 _RETRYABLE_STREAM_ERRORS = (
@@ -151,14 +145,6 @@ def _wrap_callback_errors(
         callback: The user callback.
         message: The Pub/Sub message.
     """
-    _CALLBACK_DELIVERY_LOGGER.debug(
-        "Message (id=%s, ack_id=%s, ordering_key=%s, exactly_once=%s) received by subscriber callback",
-        message.message_id,
-        message.ack_id,
-        message.ordering_key,
-        message.exactly_once_enabled,
-    )
-
     try:
         if message.opentelemetry_data:
             message.opentelemetry_data.end_subscribe_concurrency_control_span()
@@ -170,15 +156,9 @@ def _wrap_callback_errors(
         # Note: the likelihood of this failing is extremely low. This just adds
         # a message to a queue, so if this doesn't work the world is in an
         # unrecoverable state and this thread should just bail.
-
-        _CALLBACK_EXCEPTION_LOGGER.exception(
-            "Message (id=%s, ack_id=%s, ordering_key=%s, exactly_once=%s)'s callback threw exception, nacking message.",
-            message.message_id,
-            message.ack_id,
-            message.ordering_key,
-            message.exactly_once_enabled,
+        _LOGGER.exception(
+            "Top-level exception occurred in callback while processing a message"
         )
-
         message.nack()
         on_callback_error(exc)
 
@@ -219,7 +199,6 @@ def _process_requests(
     error_status: Optional["status_pb2.Status"],
     ack_reqs_dict: Dict[str, requests.AckRequest],
     errors_dict: Optional[Dict[str, str]],
-    ack_histogram: Optional[histogram.Histogram] = None,
 ):
     """Process requests when exactly-once delivery is enabled by referring to
     error_status and errors_dict.
@@ -231,16 +210,6 @@ def _process_requests(
     requests_completed = []
     requests_to_retry = []
     for ack_id in ack_reqs_dict:
-        # Debug logging: slow acks
-        if ack_histogram and ack_reqs_dict[
-            ack_id
-        ].time_to_ack > ack_histogram.percentile(percent=99):
-            _SLOW_ACK_LOGGER.debug(
-                "Message (id=%s, ack_id=%s) ack duration of %s s is higher than the p99 ack duration",
-                ack_reqs_dict[ack_id].message_id,
-                ack_reqs_dict[ack_id].ack_id,
-            )
-
         # 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:
@@ -591,10 +560,8 @@ def maybe_pause_consumer(self) -> None:
         with self._pause_resume_lock:
             if self.load >= _MAX_LOAD:
                 if self._consumer is not None and not self._consumer.is_paused:
-                    _FLOW_CONTROL_LOGGER.debug(
-                        "Message backlog over load at %.2f (threshold %.2f), initiating client-side flow control",
-                        self.load,
-                        _RESUME_THRESHOLD,
+                    _LOGGER.debug(
+                        "Message backlog over load at %.2f, pausing.", self.load
                     )
                     self._consumer.pause()
 
@@ -621,18 +588,10 @@ def maybe_resume_consumer(self) -> None:
             self._maybe_release_messages()
 
             if self.load < _RESUME_THRESHOLD:
-                _FLOW_CONTROL_LOGGER.debug(
-                    "Current load is %.2f (threshold %.2f), suspending client-side flow control.",
-                    self.load,
-                    _RESUME_THRESHOLD,
-                )
+                _LOGGER.debug("Current load is %.2f, resuming consumer.", self.load)
                 self._consumer.resume()
             else:
-                _FLOW_CONTROL_LOGGER.debug(
-                    "Current load is %.2f (threshold %.2f), retaining client-side flow control.",
-                    self.load,
-                    _RESUME_THRESHOLD,
-                )
+                _LOGGER.debug("Did not resume, current load is %.2f.", self.load)
 
     def _maybe_release_messages(self) -> None:
         """Release (some of) the held messages if the current load allows for it.
@@ -743,7 +702,7 @@ def send_unary_ack(
 
         if self._exactly_once_delivery_enabled():
             requests_completed, requests_to_retry = _process_requests(
-                error_status, ack_reqs_dict, ack_errors_dict, self.ack_histogram
+                error_status, ack_reqs_dict, ack_errors_dict
             )
         else:
             requests_completed = []
@@ -837,7 +796,7 @@ def send_unary_modack(
 
         if self._exactly_once_delivery_enabled():
             requests_completed, requests_to_retry = _process_requests(
-                error_status, ack_reqs_dict, modack_errors_dict, self.ack_histogram
+                error_status, ack_reqs_dict, modack_errors_dict
             )
         else:
             requests_completed = []
@@ -1280,11 +1239,6 @@ def _on_response(self, response: gapic_types.StreamingPullResponse) -> None:
             receipt_modack=True,
         )
 
-        if len(expired_ack_ids):
-            _EXPIRY_LOGGER.debug(
-                "ack ids %s were dropped as they have already expired."
-            )
-
         with self._pause_resume_lock:
             if self._scheduler is None or self._leaser is None:
                 _LOGGER.debug(
@@ -1350,13 +1304,9 @@ def _should_recover(self, exception: BaseException) -> bool:
         # If this is in the list of idempotent exceptions, then we want to
         # recover.
         if isinstance(exception, _RETRYABLE_STREAM_ERRORS):
-            _STREAMS_LOGGER.debug(
-                "Observed recoverable stream error %s, reopening stream", exception
-            )
+            _LOGGER.debug("Observed recoverable stream error %s", exception)
             return True
-        _STREAMS_LOGGER.debug(
-            "Observed non-recoverable stream error %s, shutting down stream", exception
-        )
+        _LOGGER.debug("Observed non-recoverable stream error %s", exception)
         return False
 
     def _should_terminate(self, exception: BaseException) -> bool:
@@ -1376,13 +1326,9 @@ def _should_terminate(self, exception: BaseException) -> bool:
         is_api_error = isinstance(exception, exceptions.GoogleAPICallError)
         # Terminate any non-API errors, or non-retryable errors (permission denied, unauthorized, etc.)
         if not is_api_error or isinstance(exception, _TERMINATING_STREAM_ERRORS):
-            _STREAMS_LOGGER.debug(
-                "Observed terminating stream error %s, shutting down stream", exception
-            )
+            _LOGGER.debug("Observed terminating stream error %s", exception)
             return True
-        _STREAMS_LOGGER.debug(
-            "Observed non-terminating stream error %s, attempting to reopen", exception
-        )
+        _LOGGER.debug("Observed non-terminating stream error %s", exception)
         return False
 
     def _on_rpc_done(self, future: Any) -> None:
@@ -1402,4 +1348,4 @@ def _on_rpc_done(self, future: Any) -> None:
             name=_RPC_ERROR_THREAD_NAME, target=self._shutdown, kwargs={"reason": error}
         )
         thread.daemon = True
-        thread.start()
+        thread.start()