RockteMQ-AI commented on code in PR #1372:
URL: https://github.com/apache/rocketmq-clients/pull/1372#discussion_r4053327875
##########
python/rocketmq/v5/consumer/push/push_consumer.py:
##########
@@ -75,47 +102,58 @@ def __init__(
self.__receive_batch_size = 32
self.__long_polling_timeout = 30 # seconds
- """ override """
-
def unsubscribe(self, topic):
+ """Unsubscribe from a topic and drop associated process queues.
+
+ Args:
+ topic: The topic name to unsubscribe from.
+ """
super().unsubscribe(topic)
for message_queue, process_queue in self.__process_queues.items():
if message_queue.topic == topic:
self.__drop_message_queue(message_queue)
self.__assignments.remove(topic)
def shutdown(self):
- logger.info(f"begin to to shutdown {self}.")
+ """Shutdown the PushConsumer and release all resources.
+
+ Stops the assignment scan scheduler, all consumption threads,
+ and closes gRPC connections.
+
+ Raises:
+ IllegalStateException: If consumer is not running or already
shutdown.
+ """
+ logger.info(f"begin to shutdown {self}.")
super().shutdown()
- self.__scan_assignment_scheduler.stop_scheduler()
logger.info(f"shutdown {self} success.")
def reset_setting(self, settings):
+ """Reset consumer settings from server-side configuration.
+
+ Updates long_polling_timeout and consumption configuration
+ (backoff policy, FIFO mode, thread count).
+
+ Args:
+ settings: The :class:`Settings` protobuf from the server.
+ """
if settings:
self.__long_polling_timeout =
settings.subscription.long_polling_timeout.seconds
self.__configure_consumer_consumption(settings)
if not self._init_settings_event.is_set():
self._init_settings_event.set()
def reset_metric(self, metric):
+ """Reset a specific metric for this consumer.
+
+ Args:
Review Comment:
**[Warning]** Leftover commented-out code: `#
self.__start_async_executor()`. Since the ThreadPoolExecutor-based approach has
been replaced by coroutine-based async, this line (and the
`__start_async_executor` method itself) should be fully removed rather than
left as a comment. Dead code adds cognitive overhead for future readers.
##########
python/rocketmq/v5/client/client.py:
##########
@@ -93,20 +91,34 @@ def startup(self):
logger.error(f"{self} startup exception: {e}")
raise e
+ def __str__(self):
+ return f"{ClientType.Name(self.client_type)},
client_id:{self.client_id}"
+
def shutdown(self):
+ """Shutdown the client and release all resources.
+
+ Stops all schedulers, the async callback executor, closes gRPC
connections,
+ clears topic route cache, and sends a termination notification to the
server.
+
+ Raises:
+ IllegalStateException: If client is not running or already
shutdown.
+ """
if not self.is_running:
- raise IllegalStateException(f"{self} is not running.")
+ logger.warn(f"{self} is not running, can't shutdown")
+ return
if self.__had_shutdown:
- raise IllegalStateException(f"{self} had shutdown.")
+ logger.warn(f"{self} had shutdown, can't shutdown again")
+ return
self._pre_shutdown()
try:
self.__stop_client_threads()
Review Comment:
**[Warning]** Commented-out code: `# self.__topic_route_cache.clear()`. The
route cache has been moved to `ClientRouteManager`, so this line in the old
`Client.shutdown` path is no longer needed. Please remove it to keep the
cleanup logic clear. (I see `self.__topic_route_cache.clear()` is correctly
placed in `ClientRouteManager.clear()` at line 884 of the diff.)
##########
python/rocketmq/v5/consumer/consumer.py:
##########
@@ -59,8 +81,28 @@ def __init__(
self._subscriptions = ConcurrentMap()
if subscription:
Review Comment:
**[Info]** `max_workers=os.cpu_count()` for the message decode executor may
create an unexpectedly large thread pool on high-core-count machines (e.g., 64+
cores in CI/production servers). Consider capping it, e.g., `min(os.cpu_count()
or 4, 8)`, to avoid excessive thread creation for a decode-only pool.
##########
python/rocketmq/v5/consumer/simple/lite_simple_consumer.py:
##########
@@ -0,0 +1,136 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from rocketmq.grpc_protocol import ClientType
+from rocketmq.v5.client import (ClientConfiguration,
+ LiteTopicSubscriptionManager)
+from rocketmq.v5.consumer.simple.simple_consumer import SimpleConsumer
+from rocketmq.v5.exception import IllegalArgumentException
+from rocketmq.v5.model import FilterExpression
+
+
+class LiteSimpleConsumer(SimpleConsumer):
Review Comment:
**[Info]** The `LiteSimpleConsumer` and `LitePushConsumer` are significant
new public API types, but the test changes only cover `SimpleConsumer`.
Consider adding at least basic unit tests for the lite variants — particularly
`subscribe_lite` / `unsubscribe_lite` lifecycle and the
`LiteTopicSubscriptionManager` quota enforcement logic.
##########
python/rocketmq/v5/consumer/push/push_consumer.py:
##########
@@ -223,73 +268,66 @@ def __query_assignment_req(self, topic, queue):
# receive #
def __execute_receive(self, message_queue, process_queue, attempt_id=None):
- if not self.is_running:
- logger.error(f"queue: {message_queue} end receive, because
consumer is not running.")
- return
- if process_queue.dropped:
- logger.info(f"queue: {message_queue} end receive, because queue is
dropped. ")
+ if not self.is_running or process_queue.dropped:
return
+
if not attempt_id:
Review Comment:
**[Info]** `__cleanup()` is called from both `_on_start_failure()` and
`_pre_shutdown()`. If `startup()` fails after partial initialization,
`_on_start_failure` runs `__cleanup()`, and then `shutdown()` may also invoke
`_pre_shutdown()` -> `__cleanup()`. Please verify that `__cleanup()` is
idempotent (e.g., null-check schedulers before stopping, check `__consumption`
state before shutting down) to avoid errors during double-cleanup.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
