Files that changed from the base of the PR and between 52fd565 and 62a0a14.

• rocketmq-broker/src/broker_runtime.rs (6 hunks)
• rocketmq-broker/src/client.rs (1 hunks)
• rocketmq-broker/src/client/client_channel_info.rs (3 hunks)
• rocketmq-broker/src/client/consumer_group_event.rs (1 hunks)
• rocketmq-broker/src/client/consumer_group_info.rs (8 hunks)
• rocketmq-broker/src/client/consumer_ids_change_listener.rs (1 hunks)
• rocketmq-broker/src/client/default_consumer_ids_change_listener.rs (1 hunks)
• rocketmq-broker/src/client/manager/consumer_manager.rs (1 hunks)
• rocketmq-broker/src/client/manager/producer_manager.rs (3 hunks)
• rocketmq-broker/src/filter.rs (1 hunks)
• rocketmq-broker/src/filter/expression_for_retry_message_filter.rs (1 hunks)
• rocketmq-broker/src/filter/expression_message_filter.rs (1 hunks)
• rocketmq-broker/src/filter/manager/consumer_filter_manager.rs (2 hunks)
• rocketmq-broker/src/offset/manager/consumer_offset_manager.rs (2 hunks)
• rocketmq-broker/src/processor/client_manage_processor.rs (2 hunks)
• rocketmq-broker/src/processor/pull_message_processor.rs (5 hunks)
• rocketmq-broker/src/processor/pull_message_result_handler.rs (2 hunks)
• rocketmq-broker/src/subscription/manager/subscription_group_manager.rs (1 hunks)
• rocketmq-broker/src/util/hook_utils.rs (2 hunks)
• rocketmq-common/src/common/broker/broker_config.rs (2 hunks)
• rocketmq-remoting/src/connection.rs (1 hunks)
• rocketmq-remoting/src/net/channel.rs (1 hunks)
• rocketmq-remoting/src/runtime/server.rs (1 hunks)
• rocketmq-store/src/base.rs (1 hunks)
• rocketmq-store/src/base/get_message_result.rs (1 hunks)
• rocketmq-store/src/base/message_result.rs (2 hunks)
• rocketmq-store/src/base/message_status_enum.rs (1 hunks)
• rocketmq-store/src/consume_queue.rs (1 hunks)
• rocketmq-store/src/filter.rs (2 hunks)
• rocketmq-store/src/lib.rs (1 hunks)
• rocketmq-store/src/log_file.rs (2 hunks)
• rocketmq-store/src/message_store/default_message_store.rs (3 hunks)
• rocketmq-store/src/queue.rs (2 hunks)
• rocketmq-store/src/queue/local_file_consume_queue_store.rs (2 hunks)

• rocketmq-store/src/message_store/default_message_store.rs (no review received)

• rocketmq-store/src/base.rs
• rocketmq-store/src/base/message_result.rs
• rocketmq-store/src/consume_queue.rs

rocketmq-broker/src/filter.rs (3)

18-18: Visibility of consumer_filter_data module set to pub(crate) is appropriate for encapsulation within the crate.

---

19-19: Adding expression_for_retry_message_filter as a pub(crate) module helps maintain encapsulation and focus on retry-specific message filtering logic.

---

20-20: The expression_message_filter module, marked as pub(crate), is well-scoped for internal use, supporting clean architecture and encapsulation.

rocketmq-broker/src/client.rs (2)

19-19: Introduction of consumer_group_event as a pub(crate) module aligns well with encapsulation principles and focuses on handling specific group events.

---

21-21: The consumer_ids_change_listener module, set as pub(crate), ensures that consumer ID change handling is encapsulated within the crate.

rocketmq-broker/src/client/consumer_ids_change_listener.rs (1)

22-26: The ConsumerIdsChangeListener trait, with methods handle and shutdown, provides a robust interface for managing consumer ID events and cleanup processes.

rocketmq-store/src/lib.rs (1)

22-22: The addition of consume_queue and filter modules in rocketmq-store enhances the store's capabilities to support new broker features.

rocketmq-store/src/filter.rs (1)

Line range hint 23-39: The MessageFilter trait definition is well-implemented and clear.

This provides a robust interface for message filtering, crucial for message processing based on different criteria.

rocketmq-broker/src/processor/pull_message_result_handler.rs (1)

17-23: New imports have been added to support the updated functionalities in PullMessageResultHandler. Ensure these libraries/modules are used efficiently.

rocketmq-store/src/base/message_status_enum.rs (1)

55-55: The GetMessageStatus enum provides a comprehensive set of statuses for message retrieval operations, enhancing the clarity and manageability of error handling.

rocketmq-broker/src/client/client_channel_info.rs (1)

19-38: The ClientChannelInfo struct has been updated to use a Channel type instead of a string address, reflecting a more robust handling of network channels. Ensure this change is well-integrated across all usages of ClientChannelInfo in the project.

Verification successful

---

The verification reveals that the channel field in ClientChannelInfo has been correctly integrated across various files. The instances of ClientChannelInfo in the files client_manage_processor.rs, consumer_group_info.rs, and producer_manager.rs all show appropriate usage of the channel field.

• rocketmq-broker/src/processor/client_manage_processor.rs
• rocketmq-broker/src/client/consumer_group_info.rs
• rocketmq-broker/src/client/manager/producer_manager.rs

This confirms that the change has been well-integrated across the project.

Scripts executed

The following scripts were executed for the analysis:

---

Script:

#!/bin/bash
# Ensure all usages of `ClientChannelInfo` have been updated to use the new `channel` field.
rg --type rust "ClientChannelInfo"

Length of output: 2345

---

---

Script:

#!/bin/bash
# Check if the new `channel` field is used correctly in all instances of `ClientChannelInfo`.

# Files identified by `rg` that use `ClientChannelInfo`
files=(
    "rocketmq-broker/src/processor/client_manage_processor.rs"
    "rocketmq-broker/src/client/consumer_group_info.rs"
    "rocketmq-broker/src/client/manager/producer_manager.rs"
)

# Check each file for correct usage of the `channel` field in `ClientChannelInfo`
for file in "${files[@]}"; do
    echo "Checking $file..."
    rg --context 5 "ClientChannelInfo" "$file" | grep -E "channel|ClientChannelInfo"
done

Length of output: 4069

rocketmq-broker/src/client/consumer_group_event.rs (1)

17-29: The ConsumerGroupEvent enum is well-defined, providing clear categorization of consumer group events. This should facilitate better event handling in consumer group management.

Tools

GitHub Check: codecov/patch

[warning] 17-17: rocketmq-broker/src/client/consumer_group_event.rs#L17
Added line #L17 was not covered by tests

rocketmq-remoting/src/connection.rs (1)

62-70: The implementation of accessor methods framed and channel is clean and follows Rust's idiomatic practices.

rocketmq-store/src/log_file.rs (1)

85-99: The newly added methods for managing message offsets and retrieving messages are well-implemented. These methods enhance the RocketMQMessageStore trait's capabilities in handling message offsets and fetching messages based on filters.

rocketmq-remoting/src/net/channel.rs (1)

22-22: The Channel struct is well-defined, with properties and methods that correctly manage channel information. Using Uuid for channel_id is a robust choice for ensuring uniqueness.

rocketmq-broker/src/client/manager/producer_manager.rs (2)

57-57: The transition to using Channel instead of String for client identification in unregister_producer aligns well with the overall improvements in channel management. Good job on maintaining the integrity of the group_channel_table by checking and potentially removing empty entries.

---

75-95: The updated register_producer function correctly handles both new and existing producers using the Channel type. Setting the last_update_timestamp on existing entries before returning is a good practice for ensuring consistency in state management.

rocketmq-broker/src/filter/manager/consumer_filter_manager.rs (2)

59-59: The addition of expression type handling in the build function is a crucial update for managing consumer filters based on the type of expression. This ensures that only relevant data is processed further.

---

98-104: Currently, the get_consumer_filter_data function always returns None. If this is a placeholder, please ensure that the implementation is completed in future iterations or clarify if this behavior is intentional.

rocketmq-broker/src/processor/client_manage_processor.rs (2)

80-80: The update to initialize ClientChannelInfo with Channel directly from the connection context in unregister_client is consistent with the overall improvements in channel management across the system.

---

106-106: The update in heart_beat to use Channel for creating ClientChannelInfo and the addition of new fields in the response command enhance the robustness and clarity of heartbeat handling.

rocketmq-broker/src/client/manager/consumer_manager.rs (5)

39-55: The new method in ConsumerManager effectively initializes all necessary components with appropriate defaults and thread safety measures using Arc and RwLock. This setup is essential for the reliable operation of consumer management.

---

57-73: The new_with_broker_stats method extends the basic initialization of ConsumerManager by incorporating broker statistics, which is critical for a comprehensive management system. The use of configuration values for timeouts is a good practice.

---

77-79: The set_broker_stats_manager method provides necessary flexibility in managing broker statistics, which is essential for adapting to changes in system configuration or operational requirements.

---

83-85: The find_subscription_data method's delegation to find_subscription_data_internal for actual data retrieval is a good practice in maintaining separation of concerns and enhancing code maintainability.

---

128-139: The compensate_subscribe_data method securely updates the compensation table with subscription data, ensuring thread safety with write locks. This method is crucial for handling data discrepancies or temporary compensations.

rocketmq-broker/src/client/consumer_group_info.rs (1)

21-21: Ensure appropriate use of parking_lot for thread safety.

The use of parking_lot::Mutex and RwLock is appropriate here for providing finer-grained lock control which is more performant than the standard Rust mutexes in this context.

Also applies to: 22-22, 25-25

rocketmq-store/src/queue/local_file_consume_queue_store.rs (1)

321-326: Ensure that the method get_max_offset handles potential errors or edge cases.

The method should robustly handle cases where the topic or queue ID might not exist, returning None or an appropriate error.

rocketmq-broker/src/processor/pull_message_processor.rs (4)

28-40: Ensure that the newly added imports are used within the file to avoid unnecessary dependencies.

---

66-67: The addition of consumer_offset_manager and message_store fields should be properly initialized and integrated within the class methods.

---

485-494: The new query_broadcast_pull_init_offset method is critical for handling broadcast messages. Ensure it handles edge cases and errors gracefully.

---

Line range hint 96-474: The modifications in the process_request_inner method enhance message pulling capabilities. Ensure thorough testing, especially around the new error handling and branching logic.

rocketmq-broker/src/broker_runtime.rs (4)

51-51: Ensure that the new import DefaultConsumerIdsChangeListener is utilized properly within the file.

---

92-92: The addition of consumer_manager as an Arc<ConsumerManager> ensures thread-safe access across different parts of the application.

---

191-191: The reassignment of consumer_manager in the BrokerRuntime constructor ensures it's properly initialized for the runtime's lifecycle. This line is critical and should be covered by integration tests to verify its impact on the system's behavior.
[ISSNUE]

+ // TODO: Ensure integration tests cover the initialization and lifecycle of consumer_manager in BrokerRuntime

Tools

GitHub Check: codecov/patch

[warning] 191-191: rocketmq-broker/src/broker_runtime.rs#L191
Added line #L191 was not covered by tests

---

122-122: The cloning of consumer_manager in the Clone implementation is critical for maintaining state consistency across clones of BrokerRuntime.

Verification successful

---

The Clone implementation for BrokerRuntime has been located, and it includes the cloning of consumer_manager. This confirms that the cloning behavior is explicitly defined and should maintain state consistency across clones.

• rocketmq-broker/src/broker_runtime.rs, lines 103-133: Clone implementation for BrokerRuntime.

Scripts executed

The following scripts were executed for the analysis:

---

Script:

#!/bin/bash
# Description: Verify that the cloning behavior of `consumer_manager` is consistent and does not lead to state inconsistencies.

# Test: Search for usages of `BrokerRuntime::clone` to ensure it's being used safely.
rg --type rust $'BrokerRuntime::clone'

Length of output: 37

---

---

Script:

#!/bin/bash
# Locate the `BrokerRuntime` struct definition.
ast-grep --lang rust --pattern 'struct BrokerRuntime { $$$ }'

Length of output: 63

---

---

Script:

#!/bin/bash
# Search for the definition of `BrokerRuntime` in all Rust files.
rg --type rust 'struct BrokerRuntime'

Length of output: 188

---

---

Script:

#!/bin/bash
# Locate the `Clone` implementation for `BrokerRuntime` in `rocketmq-broker/src/broker_runtime.rs`.
ast-grep --lang rust --pattern 'impl Clone for BrokerRuntime { $$$ }' rocketmq-broker/src/broker_runtime.rs

Length of output: 3092

Tools

GitHub Check: codecov/patch

[warning] 122-122: rocketmq-broker/src/broker_runtime.rs#L122
Added line #L122 was not covered by tests