You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2022/04/19 06:04:43 UTC
[rocketmq-apis] branch v2 updated: Refactor client settings (#21)
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch v2
in repository https://gitbox.apache.org/repos/asf/rocketmq-apis.git
The following commit(s) were added to refs/heads/v2 by this push:
new 64d0928 Refactor client settings (#21)
64d0928 is described below
commit 64d092872de98f4cacf38a36a8becf2aa9d8b005
Author: Aaron Ai <ya...@alibaba-inc.com>
AuthorDate: Tue Apr 19 14:04:39 2022 +0800
Refactor client settings (#21)
---
apache/rocketmq/v2/definition.proto | 66 ----------------------
apache/rocketmq/v2/service.proto | 109 ++++++++++++++++++++++++++----------
2 files changed, 79 insertions(+), 96 deletions(-)
diff --git a/apache/rocketmq/v2/definition.proto b/apache/rocketmq/v2/definition.proto
index cc5df5c..3856838 100644
--- a/apache/rocketmq/v2/definition.proto
+++ b/apache/rocketmq/v2/definition.proto
@@ -58,20 +58,6 @@ message FilterExpression {
string expression = 2;
}
-// Dead lettering is done on a best effort basis. The same message might be
-// dead lettered multiple times.
-//
-// If validation on any of the fields fails at subscription creation/update,
-// the create/update subscription request will fail.
-message DeadLetterPolicy {
- // The maximum number of delivery attempts for any message.
- //
- // This field will be honored on a best effort basis.
- //
- // If this parameter is 0, a default value of 16 is used.
- int32 max_delivery_attempts = 1;
-}
-
message RetryPolicy {
int32 max_attempts = 1;
float initial_backoff = 2;
@@ -87,58 +73,6 @@ message Resource {
string name = 2;
}
-message Tracing {
- bool on = 1;
- // Tracing access point for client, which is essential if tracing is on.
- optional Endpoints service_access_point = 2;
-}
-
-message Publishing {
- // Publisher normally registers topics in interest, such that
- // pre-conditions may be examined and validated.
- repeated Resource topics = 1;
-
- // If publishing message experiences RPC failure, `retry_policy` describes
- // backoff policy before retries are made.
- RetryPolicy retry_policy = 2;
-
- // If message body size exceeds `compress_threshold`, it would be desirable to
- // compress it to relieve network overhead.
- int32 compress_threshold = 3;
-
- // Max message size in bytes permitted by server.
- int32 max_message_bytes = 4;
-}
-
-message CacheLimits {
- int32 count = 1;
- int64 bytes = 2;
-}
-
-message Subscription {
- Resource group = 1;
-
- repeated SubscriptionEntry subscriptions = 2;
-
- DeadLetterPolicy dead_letter_policy = 3;
-
- bool fifo = 5;
-
- int32 max_receive_batch_size = 8;
-
- // After messages are received from servers, consumers normally split them
- // into multiple mini-batches. Each mini-batch is assigned to a dedicated
- // task, which will be submitted to thread-pool to run concurrently.
- int32 messages_per_task = 9;
-
- CacheLimits cache_limits = 10;
-
- int32 consume_thread_count = 11;
-
- // Up to server
- google.protobuf.Duration max_invisible_duration = 12;
-}
-
message SubscriptionEntry {
Resource topic = 1;
FilterExpression expression = 2;
diff --git a/apache/rocketmq/v2/service.proto b/apache/rocketmq/v2/service.proto
index da47882..0ff6fb2 100644
--- a/apache/rocketmq/v2/service.proto
+++ b/apache/rocketmq/v2/service.proto
@@ -46,9 +46,7 @@ option java_outer_classname = "MQService";
// access-point, which annotates tenant-id, instance-id or other
// vendor-specific settings. Purpose-built name servers may respond customized
// results based on these particular requirements.
-message QueryRouteRequest {
- Resource topic = 1;
-}
+message QueryRouteRequest { Resource topic = 1; }
message QueryRouteResponse {
Status status = 1;
@@ -198,51 +196,102 @@ message RecoverOrphanedTransactionCommand {
string transaction_id = 2;
}
-message Settings {
- optional Publishing publishing = 1;
- optional Subscription subscription = 2;
- Tracing tracing = 3;
+message ActivePublishingSettings {
+ repeated Resource publishing_topics = 1;
+ RetryPolicy retry_policy = 2;
}
-enum Direction {
- DIRECTION_UNSPECIFIED = 0;
- REQUEST = 1;
- RESPONSE = 2;
+message ActiveSubscriptionSettings {
+ Resource group = 1;
+ repeated SubscriptionEntry subscriptions = 2;
+ // Consume thread count is essential if `ClientType` is `PushConsumer`.
+ optional int32 consume_thread_count = 3;
+ // Consume max batch size is essential if `ClientType` is `PushConsumer`.
+ optional int32 consume_max_batch_size = 4;
+ // Max cache message bytes is essential if `ClientType` is `PushConsumer`.
+ optional int64 max_cache_message_bytes = 5;
+ // Await duration here is essential if `ClientType` is `SimpleConsumer`.
+ optional google.protobuf.Duration simple_consumer_receive_await_duration = 6;
}
-message ClientSettings {
+message ReportActiveSettingsCommand {
string nonce = 1;
- Direction direction = 2;
- ClientType client_type = 3;
- Endpoints access_point = 4;
- Settings settings = 5;
+ ClientType client_type = 2;
+ Endpoints access_point = 3;
+ google.protobuf.Duration connection_timeout = 4;
+ bool trace_on = 5;
+ // Settings will be ignored by server if it is not match with the client type.
+ oneof settings {
+ ActivePublishingSettings active_publishing_settings = 6;
+ ActiveSubscriptionSettings active_subscription_settings = 7;
+ }
}
-message ClientOverwrittenSettings {
+message ReportActiveSettingsResult { string nonce = 1; }
+
+message PassivePublishingSettings {
+ // Message is compressed if the body size exceeds the threshold.
+ int32 message_body_compression_bytes_threshold = 1;
+ // Publishing is rejected if the message body size exceeds the threshold.
+ int32 max_message_body_bytes = 2;
+ google.protobuf.Duration orphaned_transaction_recovery_duration = 3;
+}
+
+message PassiveSubscriptionSettings {
+ bool fifo = 1;
+ // Max fifo consume attempts is essential if `fifo` flag is true, a default
+ // value of 17 is used if this parameter is not set.
+ optional int32 max_fifo_delivery_attempts = 2;
+ // Fifo consume retry interval is essential if `fifo` flag is true, a default
+ // value of 1s is used if this parameter is not set.
+ optional google.protobuf.Duration fifo_delivery_retry_interval = 3;
+ // Message receive batch size here is essential if `ClientType` is
+ // `PushConsumer`,
+ optional int32 push_consumer_receive_batch_size = 4;
+ // Long-polling timeout for `ReceiveMessageRequest`, which is essential if
+ // `ClientType` is `PushCosnumer`. A default value of 30s is
+ // used if this parameter is not set.
+ optional google.protobuf.Duration push_consumer_receive_await_duration = 5;
+ // Consume timeout is only works if the fifo flag is false. A default value of
+ // 15min is used if this parameter is not set.
+ optional google.protobuf.Duration consume_timeout = 6;
+}
+
+message ApplyPassiveSettingsCommand {
string nonce = 1;
- Direction direction = 2;
- Settings settings = 3;
+ Endpoints trace_access_point = 2;
+ // Settings will be ignored by server if it is not match with the client type.
+ oneof settings {
+ PassivePublishingSettings passive_publishing_settings = 3;
+ PassiveSubscriptionSettings passive_subscription_settings = 4;
+ }
}
+message ApplyPassiveSettingsResult { string nonce = 1; }
+
message TelemetryCommand {
oneof command {
// These messages are from client.
- // Report local client setting to server.
- ClientSettings client_settings = 1;
+ // Report active setting to server.
+ ReportActiveSettingsCommand report_active_settings_command = 1;
+ // Ack `ApplyPassiveSettingsCommand` from server.
+ ApplyPassiveSettingsResult apply_passive_settings_result = 2;
// Report thread stack trace to server.
- ThreadStackTrace thread_stack_trace = 2;
+ ThreadStackTrace thread_stack_trace = 3;
// Repost message verify result to server.
- VerifyMessageResult verify_message_result = 3;
+ VerifyMessageResult verify_message_result = 4;
// There messages are from server.
- // Overwrite client settings.
- ClientOverwrittenSettings client_overwritten_settings = 4;
+ // Deliver passive settings to client.
+ ApplyPassiveSettingsCommand apply_passive_settings_command = 5;
+ // Ack `ReportActiveSettingsCommand` from client.
+ ReportActiveSettingsResult report_active_settings_result = 6;
// Request client to recover the orphaned transaction message.
- RecoverOrphanedTransactionCommand recover_orphaned_transaction_command = 5;
+ RecoverOrphanedTransactionCommand recover_orphaned_transaction_command = 7;
// Request client to print thread stack trace.
- PrintThreadStackTraceCommand print_thread_stack_trace_command = 6;
+ PrintThreadStackTraceCommand print_thread_stack_trace_command = 8;
// Request client to verify the consumption of the appointed message.
- VerifyMessageCommand verify_message_command = 7;
+ VerifyMessageCommand verify_message_command = 9;
}
}
@@ -354,8 +403,8 @@ service MessagingService {
// `NOT_FOUND`.
rpc NackMessage(NackMessageRequest) returns (NackMessageResponse) {}
- // Forwards one message to dead letter queue if the DeadLetterPolicy is
- // triggered by this message at client-side, return `OK` if success.
+ // Forwards one message to dead letter queue if the max delivery attempts is
+ // exceeded by this message at client-side, return `OK` if success.
rpc ForwardMessageToDeadLetterQueue(ForwardMessageToDeadLetterQueueRequest)
returns (ForwardMessageToDeadLetterQueueResponse) {}