Skip to content

feat: add deployment and lite topic fields to Admin service #116

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lizhimins merged 1 commit into from
Jul 30, 2026
Merged

feat: add deployment and lite topic fields to Admin service #116

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions ChangeLog.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
9. Enums field number = 0 is redefined to meet requirement [Each enum value should end with a semicolon, not a comma. Prefer prefixing enum values instead of surrounding them in an enclosing message. The zero value enum should have the suffix UNSPECIFIED.](https://developers.google.com/protocol-buffers/docs/style)
10. Nested enumerations are externalized due to the same guide item as above.
11. Expanded the Admin service with control-plane RPCs for topic/subscription/consumer administration and diagnostics (DescribeTopicStatus, ListSubscription, DescribeSubscription, DeleteSubscription, DescribeGroupAccumulation, ListConsumerConnection, ResetGroupOffset, QueryMessage, PrintThreadStackTrace, VerifyMessage, AdminSendMessage, GetConsumerRunningInfo, GetTopicRoute, QueryTimeSpan, GetProxyRuntimeStats).
12. Admin control-plane requests carry an optional `deployment_name` to address a single deployment in multi-tenant environments, and the lite topic dimension is exposed through `lite_topic` / `liteTopic` / `lite_topic_accumulation` plus the `LITE_SELECTIVE` consumption model. Admin field numbers are assigned so that the definitions stay wire compatible with Admin services already deployed against these RPCs.

Remaining Issues:
How server publishes conf and conf changes to clients.
143 changes: 94 additions & 49 deletions apache/rocketmq/v2/admin.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,10 @@ message ChangeLogLevelResponse {

// Request the status and metadata of a topic.
message DescribeTopicStatusRequest {
Resource topic = 1;
// Identifies the target deployment (instance) in multi-tenant environments
// where a single endpoint serves several isolated deployments.
optional string deployment_name = 1;
Resource topic = 2;
}

message DescribeTopicStatusResponse {
Expand All @@ -73,8 +76,10 @@ message DescribeTopicStatusResponse {
// List subscription relationships filtered by topic and/or group.
// At least one of the two filters is expected to be set.
message ListSubscriptionRequest {
optional Resource topic = 1;
optional Resource group = 2;
// Identifies the target deployment (instance) in multi-tenant environments.
optional string deployment_name = 1;
optional Resource topic = 2;
optional Resource group = 3;
}

// A single subscription relationship between a group and a topic.
Expand Down Expand Up @@ -107,8 +112,10 @@ message ListSubscriptionResponse {
// ListSubscription, this returns the subscription reported by each individual
// client, which helps diagnose inconsistent subscriptions within a group.
message DescribeSubscriptionRequest {
optional Resource topic = 1;
optional Resource group = 2;
// Identifies the target deployment (instance) in multi-tenant environments.
optional string deployment_name = 1;
optional Resource topic = 2;
optional Resource group = 3;
}

message DescribeSubscriptionResponse {
Expand All @@ -124,11 +131,13 @@ message DescribeSubscriptionResponse {

// Delete a subscription relationship between a group and a topic.
message DeleteSubscriptionRequest {
Resource topic = 1;
Resource group = 2;
// Identifies the target deployment (instance) in multi-tenant environments.
optional string deployment_name = 1;
Resource topic = 2;
Resource group = 3;

// The filter expression that identifies the subscription to delete.
FilterExpression expression = 3;
FilterExpression expression = 4;
}

message DeleteSubscriptionResponse {
Expand Down Expand Up @@ -164,6 +173,10 @@ enum MessageModel {

// Every consumer within the group receives the full stream of messages.
BROADCASTING = 2;

// Consumers of the group selectively subscribe to individual lite topics
// that are multiplexed onto a shared physical topic.
LITE_SELECTIVE = 3;
}

// A message queue together with the process queue snapshot held by a consumer.
Expand Down Expand Up @@ -218,11 +231,16 @@ message ConsumerRunningInfo {

// Query the message accumulation (lag) of a consumer group.
message DescribeGroupAccumulationRequest {
Resource group = 1;
// Identifies the target deployment (instance) in multi-tenant environments.
optional string deployment_name = 1;
Resource group = 2;

// Optional set of topics to break the accumulation down by. When empty, the
// aggregated accumulation of the whole group is returned.
repeated Resource topics = 2;
repeated Resource topics = 3;

// Restrict the query to a single lite topic multiplexed onto the topics above.
optional string lite_topic = 4;
}

message DescribeGroupAccumulationResponse {
Expand Down Expand Up @@ -251,12 +269,17 @@ message DescribeGroupAccumulationResponse {

// Per-topic accumulation, keyed by topic name.
map<string, GroupAccumulation> topic_accumulation = 3;

// Per-lite-topic accumulation, keyed by lite topic name.
map<string, GroupAccumulation> lite_topic_accumulation = 4;
}

// Query the time span of messages consumed by a group across its topics.
message QueryTimeSpanRequest {
Resource group = 1;
repeated Resource topics = 2;
// Identifies the target deployment (instance) in multi-tenant environments.
optional string deployment_name = 1;
Resource group = 2;
repeated Resource topics = 3;
}

message QueryTimeSpanResponse {
Expand All @@ -283,10 +306,15 @@ message QueryTimeSpanResponse {

// List the online consumer connections of a group.
message ListConsumerConnectionRequest {
Resource group = 1;
// Identifies the target deployment (instance) in multi-tenant environments.
optional string deployment_name = 1;
Resource group = 2;

// Optional topic filter.
optional Resource topic = 2;
optional Resource topic = 3;

// Optional lite topic filter, multiplexed onto the topic above.
optional string liteTopic = 4;
}

message ListConsumerConnectionResponse {
Expand All @@ -296,11 +324,13 @@ message ListConsumerConnectionResponse {

// Reset the consume offset of a group on a topic to the given timestamp.
message ResetGroupOffsetRequest {
Resource group = 1;
Resource topic = 2;
// Identifies the target deployment (instance) in multi-tenant environments.
optional string deployment_name = 1;
Resource group = 2;
Resource topic = 3;

// Messages stored at or after this timestamp will be re-consumed.
google.protobuf.Timestamp reset_timestamp = 3;
google.protobuf.Timestamp reset_timestamp = 4;
}

message ResetGroupOffsetResponse {
Expand All @@ -309,32 +339,37 @@ message ResetGroupOffsetResponse {

// Query messages of a topic by id, key or subscription within a time range.
message ListMessageRequest {
Resource topic = 1;
// Identifies the target deployment (instance) in multi-tenant environments.
optional string deployment_name = 1;
Resource topic = 2;

// Maximum number of messages to return.
int32 max_message_nums = 2;
int32 max_message_nums = 3;

// Time range (inclusive begin, inclusive end) to scan.
google.protobuf.Timestamp begin_timestamp = 3;
google.protobuf.Timestamp end_timestamp = 4;
google.protobuf.Timestamp begin_timestamp = 4;
google.protobuf.Timestamp end_timestamp = 5;

// The key used to locate messages. Exactly one should be set.
oneof search_key {
string message_id = 5;
string message_key = 6;
string subscription = 7;
string message_id = 6;
string message_key = 7;
string subscription = 8;

// Scan every message belonging to the given lite topic.
string lite_topic = 14;
}

// Opaque cursor for scroll-style pagination, echoed from a previous response.
optional string scroll_id = 8;
optional string scroll_id = 9;

// Page-style pagination parameters.
optional int32 page_num = 9;
optional int32 page_size = 10;
optional int32 page_num = 10;
optional int32 page_size = 11;

// Restrict the query to a specific broker and queue.
optional string broker_name = 11;
optional int32 queue_id = 12;
optional string broker_name = 12;
optional int32 queue_id = 13;
}

message ListMessageResponse {
Expand All @@ -347,8 +382,10 @@ message ListMessageResponse {

// Print the thread stack trace of a specific client, for diagnostics.
message PrintThreadStackTraceRequest {
string client_id = 1;
Resource group = 2;
// Identifies the target deployment (instance) in multi-tenant environments.
optional string deployment_name = 1;
string client_id = 2;
Resource group = 3;
}

message PrintThreadStackTraceResponse {
Expand All @@ -359,10 +396,12 @@ message PrintThreadStackTraceResponse {
// Ask a specific client to consume the given message once, to verify that the
// consumer logic works as expected.
message VerifyMessageRequest {
string client_id = 1;
Resource group = 2;
Resource topic = 3;
string message_id = 4;
// Identifies the target deployment (instance) in multi-tenant environments.
optional string deployment_name = 1;
string client_id = 2;
Resource group = 3;
Resource topic = 4;
string message_id = 5;
}

message VerifyMessageResponse {
Expand All @@ -372,21 +411,23 @@ message VerifyMessageResponse {
// Send a message from the admin side, typically used to send a test message
// from a console.
message AdminSendMessageRequest {
Resource topic = 1;
// Identifies the target deployment (instance) in multi-tenant environments.
optional string deployment_name = 1;
Resource topic = 2;
// Tag, which is optional.
optional string tag = 2;
optional string tag = 3;
// Message key
optional string key = 3;
optional string key = 4;
// Message body
bytes body = 4;
bytes body = 5;
// User-defined properties of the message.
map<string, string> user_properties = 5;
map<string, string> user_properties = 6;
// System properties of the message.
optional SystemProperties system_properties = 6;
optional SystemProperties system_properties = 7;
// Request-scoped extension information. Unknown entries should be ignored.
// Known keys:
// - protocol_type: Protocol used by the original producer (REMOTING or GRPC_V2).
map<string, string> ext_info = 7;
map<string, string> ext_info = 8;
}

message AdminSendMessageResponse {
Expand All @@ -398,8 +439,10 @@ message AdminSendMessageResponse {

// Fetch the aggregated running information of a specific consumer client.
message GetConsumerRunningInfoRequest {
Resource group = 1;
string client_id = 2;
// Identifies the target deployment (instance) in multi-tenant environments.
optional string deployment_name = 1;
Resource group = 2;
string client_id = 3;
}

message GetConsumerRunningInfoResponse {
Expand All @@ -415,17 +458,19 @@ message GetTopicRouteRequest {
INTERNET = 1;
INTRANET = 2;
}
Resource topic = 1;
NetworkType network_type = 2;
// Identifies the target deployment (instance) in multi-tenant environments.
optional string deployment_name = 1;
Resource topic = 2;
NetworkType network_type = 3;

// Protocol type the client speaks, e.g. "grpc" or "remoting".
string protocol_type = 3;
string protocol_type = 4;

// Whether the request comes from a streaming client.
bool stream_request_type = 4;
bool stream_request_type = 5;

// Address of the requesting client.
string client_address = 5;
string client_address = 6;
}

message GetTopicRouteResponse {
Expand Down
Loading