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/07/01 08:04:13 UTC
[rocketmq-clients] branch cpp updated (5a6593d -> 6b31ad5)
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a change to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
discard 5a6593d Use official gRPC release v1.46.3
discard 25b47d7 Upgrade grpc to v1.46.3.2
discard cc4db86 Upgrade grpc to v1.46.3.1
discard c492fba Fix url of bazel_proto mirror
discard d83ee7b Add mirror for rules_proto
discard 8b947a4 Add io_bazel_rules_go mirror
discard ae31a80 Add dependent CDN through OSS
discard 5d080d4 Translate ClientIdRequired to InternalClientError
discard 747d0ab Add error handling
discard 770da2b Add error handling for NotifyClientTermination
discard b477272 Add error handling for EndTransaction
discard f695a67 Add error handling for Ack
discard 256f659 Add error handling for ReceiveMessage
discard 6c5c66e Complete error handling for QueryAssignment
discard 9d6b5ed Finish error handling of SendMessage
discard 3d9ec8f Complete error handling for Heartbeat
discard 5bc970f Complete error handling of QueryRoute
discard 979c226 Sync and update protobuf files to v2.0-alpha
add 8f208e7 Java: adapt with the latest protocol
add 78a4791 Java: bugfix for MessageIdImpl#toString
new 00080c3 Sync and update protobuf files to v2.0-alpha
new acee141 Complete error handling of QueryRoute
new 4a9426e Complete error handling for Heartbeat
new efdd9e1 Finish error handling of SendMessage
new 1b1dfd0 Complete error handling for QueryAssignment
new b4a2e50 Add error handling for ReceiveMessage
new ce9a35f Add error handling for Ack
new f40067b Add error handling for EndTransaction
new 9c859cb Add error handling for NotifyClientTermination
new bae23b8 Add error handling
new fcf63ce Translate ClientIdRequired to InternalClientError
new 390a87e Add dependent CDN through OSS
new a5aa904 Add io_bazel_rules_go mirror
new cde80b0 Add mirror for rules_proto
new 747d699 Fix url of bazel_proto mirror
new c5bbe34 Upgrade grpc to v1.46.3.1
new 056e68c Upgrade grpc to v1.46.3.2
new 6b31ad5 Use official gRPC release v1.46.3
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (5a6593d)
\
N -- N -- N refs/heads/cpp (6b31ad5)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 18 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../client/java/message/MessageIdImpl.java | 4 ++
.../apache/rocketmq/client/java/tool/TestBase.java | 9 ++--
java/pom.xml | 61 ++++++++++++++++++----
3 files changed, 58 insertions(+), 16 deletions(-)
[rocketmq-clients] 10/18: Add error handling
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit bae23b8f445ba2b7015782e1707c7797a583652c
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Thu Jun 30 19:26:28 2022 +0800
Add error handling
---
cpp/src/main/cpp/client/ClientManagerImpl.cpp | 41 +++++++++++++++++++++++++--
1 file changed, 38 insertions(+), 3 deletions(-)
diff --git a/cpp/src/main/cpp/client/ClientManagerImpl.cpp b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
index 8d182d8..0f4c7f5 100644
--- a/cpp/src/main/cpp/client/ClientManagerImpl.cpp
+++ b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
@@ -1117,29 +1117,64 @@ void ClientManagerImpl::changeInvisibleDuration(
std::error_code ec;
auto&& status = invocation_context->response.status();
+ auto&& peer_address = invocation_context->remote_address;
switch (status.code()) {
case rmq::Code::OK: {
- SPDLOG_DEBUG("Nack to {} OK", invocation_context->remote_address);
+ SPDLOG_DEBUG("ChangeInvisibleDuration to {} OK", peer_address);
break;
};
+
+ case rmq::Code::ILLEGAL_TOPIC: {
+ SPDLOG_WARN("IllegalTopic: {}. Host={}", status.message(), peer_address);
+ ec = ErrorCode::IllegalTopic;
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_CONSUMER_GROUP: {
+ SPDLOG_WARN("IllegalConsumerGroup: {}. Host={}", status.message(), peer_address);
+ ec = ErrorCode::IllegalConsumerGroup;
+ break;
+ }
+
case rmq::Code::UNAUTHORIZED: {
SPDLOG_WARN("Unauthorized: {}, host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::Unauthorized;
break;
}
+
+ case rmq::Code::INVALID_RECEIPT_HANDLE: {
+ SPDLOG_WARN("InvalidReceiptHandle: {}, host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::InvalidReceiptHandle;
+ break;
+ }
+
+ case rmq::Code::CLIENT_ID_REQUIRED: {
+ SPDLOG_WARN("ClientIdRequired: {}, host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::InternalClientError;
+ break;
+ }
+
case rmq::Code::FORBIDDEN: {
SPDLOG_WARN("Forbidden: {}, host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::Forbidden;
break;
}
+
case rmq::Code::INTERNAL_SERVER_ERROR: {
SPDLOG_WARN("InternalServerError: {}, host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::InternalServerError;
break;
}
+
+ case rmq::Code::TOO_MANY_REQUESTS: {
+ SPDLOG_WARN("TooManyRequests: {}, host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::TooManyRequests;
+ break;
+ }
+
default: {
- SPDLOG_WARN("NotImplemented: Please upgrade to latest SDK, host={}", invocation_context->remote_address);
- ec = ErrorCode::NotImplemented;
+ SPDLOG_WARN("NotSupported: Please upgrade to latest SDK, host={}", invocation_context->remote_address);
+ ec = ErrorCode::NotSupported;
break;
}
}
[rocketmq-clients] 07/18: Add error handling for Ack
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit ce9a35fd7bb4ee6f1179adf65cfb3d0586876f4a
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Thu Jun 30 17:32:04 2022 +0800
Add error handling for Ack
---
cpp/src/main/cpp/client/ClientManagerImpl.cpp | 71 ++++++++++++++++++++++++---
1 file changed, 64 insertions(+), 7 deletions(-)
diff --git a/cpp/src/main/cpp/client/ClientManagerImpl.cpp b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
index 8562c6d..3f572b3 100644
--- a/cpp/src/main/cpp/client/ClientManagerImpl.cpp
+++ b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
@@ -1010,23 +1010,80 @@ void ClientManagerImpl::ack(const std::string& target, const Metadata& metadata,
switch (status.code()) {
case rmq::Code::OK: {
SPDLOG_DEBUG("Ack OK. host={}", invocation_context->remote_address);
- } break;
+ break;
+ }
+
+ case rmq::Code::MULTIPLE_RESULTS: {
+ SPDLOG_DEBUG("Server returns multiple results. host={}", invocation_context->remote_address);
+ // Treat it as successful, allowing top tier processing according to response entries.
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_TOPIC: {
+ SPDLOG_WARN("IllegalTopic: {}, host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::IllegalTopic;
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_CONSUMER_GROUP: {
+ SPDLOG_WARN("IllegalConsumerGroup: {}, host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::IllegalConsumerGroup;
+ break;
+ }
+
+ case rmq::Code::INVALID_RECEIPT_HANDLE: {
+ SPDLOG_WARN("InvalidReceiptHandle: {}, host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::InvalidReceiptHandle;
+ break;
+ }
+
+ case rmq::Code::CLIENT_ID_REQUIRED: {
+ SPDLOG_WARN("ClientIdRequired: {}, host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::ClientIdRequired;
+ break;
+ }
+
+ case rmq::Code::TOPIC_NOT_FOUND: {
+ SPDLOG_WARN("TopicNotFound: {}, host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::TopicNotFound;
+ break;
+ }
+
case rmq::Code::UNAUTHORIZED: {
SPDLOG_WARN("Unauthorized: {}, host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::Unauthorized;
- } break;
+ break;
+ }
+
case rmq::Code::FORBIDDEN: {
SPDLOG_WARN("PermissionDenied: {}, host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::Forbidden;
- } break;
+ break;
+ }
+
+ case rmq::Code::TOO_MANY_REQUESTS: {
+ SPDLOG_WARN("TooManyRequests: {}, host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::TooManyRequests;
+ break;
+ }
+
case rmq::Code::INTERNAL_SERVER_ERROR: {
SPDLOG_WARN("InternalServerError: {}, host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::InternalServerError;
- } break;
+ break;
+ }
+
+ case rmq::Code::PROXY_TIMEOUT: {
+ SPDLOG_WARN("GatewayTimeout: {}, host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::GatewayTimeout;
+ break;
+ }
+
default: {
- SPDLOG_WARN("NotImplement: please upgrade SDK to latest release. host={}", invocation_context->remote_address);
- ec = ErrorCode::NotImplemented;
- } break;
+ SPDLOG_WARN("NotSupported: please upgrade SDK to latest release. host={}", invocation_context->remote_address);
+ ec = ErrorCode::NotSupported;
+ break;
+ }
}
cb(ec);
};
[rocketmq-clients] 02/18: Complete error handling of QueryRoute
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit acee141965ab701bdc2bc858a2715cb8d4b09b32
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Thu Jun 30 15:10:57 2022 +0800
Complete error handling of QueryRoute
---
cpp/api/rocketmq/ErrorCode.h | 95 +++++++++++++++++-----
cpp/src/main/cpp/base/ErrorCategory.cpp | 9 +-
cpp/src/main/cpp/client/ClientManagerImpl.cpp | 52 +++++++++---
.../main/cpp/client/ReceiveMessageStreamReader.cpp | 2 +-
4 files changed, 120 insertions(+), 38 deletions(-)
diff --git a/cpp/api/rocketmq/ErrorCode.h b/cpp/api/rocketmq/ErrorCode.h
index 5068c83..14ec504 100644
--- a/cpp/api/rocketmq/ErrorCode.h
+++ b/cpp/api/rocketmq/ErrorCode.h
@@ -33,6 +33,11 @@ enum class ErrorCode : int {
*/
IllegalState = 1,
+ /**
+ * @brief To publish FIFO messages, only synchronous API is supported.
+ */
+ BadRequestAsyncPubFifoMessage = 10001,
+
/**
* @brief Broker has processed the request but is not going to return any content.
*/
@@ -45,53 +50,92 @@ enum class ErrorCode : int {
BadConfiguration = 300,
/**
- * @brief The server cannot process the request due to apprent client-side
+ * @brief Generic code, representing multiple return results.
+ *
+ */
+ MultipleResults = 30000,
+
+ /**
+ * @brief The server cannot process the request due to apparent client-side
* error. For example, topic contains invalid character or is excessively
* long.
*
*/
- BadRequest = 400,
+ BadRequest = 40000,
/**
- * @brief To publish FIFO messages, only synchronous API is supported.
+ * @brief The access point is illegal
+ *
+ */
+ IllegalAccessPoint = 40001,
+
+ IllegalTopic = 40002,
+ IllegalConsumerGroup = 40003,
+ IllegalMessageTag = 40004,
+ IllegalMessageKey = 40005,
+ IllegalMessageGroup = 40006,
+ IllegalMessageProperty = 40007,
+ InvalidTransactionId = 40008,
+ IllegalMessageId = 40009,
+ IllegalFilterExpression = 40010,
+ InvalidReceiptHandle = 40011,
+
+ /**
+ * @brief Message property conflicts with its type.
*/
- BadRequestAsyncPubFifoMessage = 40001,
+ MessagePropertyConflictWithType = 40012,
+
+ // Client type is not recognized by server
+ UnsupportedClientType = 40013,
+
+ // Received message is corrupted, generally failing to pass integrity checksum validation.
+ MessageCorrupted = 40014,
+
+ // Request is rejected due to missing of x-mq-client-id HTTP header in the metadata frame.
+ ClientIdRequired = 40015,
/**
* @brief Authentication failed. Possibly caused by invalid credentials.
*
*/
- Unauthorized = 401,
+ Unauthorized = 40100,
/**
* @brief Credentials are understood by server but authenticated user does not
* have privilege to perform the requested action.
*
*/
- Forbidden = 403,
+ Forbidden = 40300,
/**
* @brief Topic not found, which should be created through console or
* administration API before hand.
*
*/
- NotFound = 404,
+ NotFound = 40400,
+
+ MessageNotFound = 40401,
- TopicNotFound = 404001,
+ TopicNotFound = 404002,
- GroupNotFound = 404002,
+ ConsumerGroupNotFound = 404003,
/**
* @brief Timeout when connecting, reading from or writing to brokers.
*
*/
- RequestTimeout = 408,
+ RequestTimeout = 40800,
/**
* @brief Message body is too large.
*
*/
- PayloadTooLarge = 413,
+ PayloadTooLarge = 41300,
+
+ /**
+ * @brief Message body size exceeds limited allowed by server.
+ */
+ MessageBodyTooLarge = 41301,
/**
* @brief When trying to perform an action whose dependent procedure state is
@@ -100,14 +144,14 @@ enum class ErrorCode : int {
* 2. Commit/Rollback a transactional message that does not exist;
* 3. Commit an offset which is greater than maximum of partition;
*/
- PreconditionRequired = 428,
+ PreconditionRequired = 42800,
/**
* @brief Quota exchausted. The user has sent too many requests in a given
* amount of time.
*
*/
- TooManyRequest = 429,
+ TooManyRequest = 42900,
/**
* @brief The server is unwilling to process the request because either an
@@ -115,62 +159,69 @@ enum class ErrorCode : int {
* large
*
*/
- HeaderFieldsTooLarge = 431,
+ HeaderFieldsTooLarge = 43100,
+
+ // Message properties total size exceeds the threshold.
+ MessagePropertiesTooLarge = 43101,
/**
* @brief A server operator has received a legal demand to deny access to a
* resource or to a set of resources that includes the requested resource.
*
*/
- UnavailableForLegalReasons = 451,
+ UnavailableForLegalReasons = 45100,
/**
* @brief Server side interval error
*
*/
- InternalServerError = 500,
+ InternalServerError = 50000,
/**
* @brief The server either does not recognize the request method, or it lacks
* the ability to fulfil the request.
*
*/
- NotImplemented = 501,
+ NotImplemented = 50100,
/**
* @brief The server was acting as a gateway or proxy and received an invalid
* response from the upstream server.
*
*/
- BadGateway = 502,
+ BadGateway = 50200,
/**
* @brief The server cannot handle the request (because it is overloaded or
* down for maintenance). Generally, this is a temporary state.
*
*/
- ServiceUnavailable = 503,
+ ServiceUnavailable = 50300,
/**
* @brief The server was acting as a gateway or proxy and did not receive a
* timely response from the upstream server.
*
*/
- GatewayTimeout = 504,
+ GatewayTimeout = 50400,
/**
* @brief The server does not support the protocol version used in the
* request.
*
*/
- ProtocolVersionNotSupported = 505,
+ NotSupported = 50500,
+
+ ProtocolUnsupported = 50501,
+
+ VerifyFifoMessageUnsupported = 50502,
/**
* @brief The server is unable to store the representation needed to complete
* the request.
*
*/
- InsufficientStorage = 507,
+ InsufficientStorage = 50700,
};
std::error_code make_error_code(ErrorCode code);
diff --git a/cpp/src/main/cpp/base/ErrorCategory.cpp b/cpp/src/main/cpp/base/ErrorCategory.cpp
index 4fe2ea0..4f1b290 100644
--- a/cpp/src/main/cpp/base/ErrorCategory.cpp
+++ b/cpp/src/main/cpp/base/ErrorCategory.cpp
@@ -53,7 +53,7 @@ std::string ErrorCategory::message(int code) const {
case ErrorCode::TopicNotFound:
return "Topic is not found. Verify the request topic has already been created through console or management API";
- case ErrorCode::GroupNotFound:
+ case ErrorCode::ConsumerGroupNotFound:
return "Group is not found. Verify the request group has already been created through console or management API";
case ErrorCode::RequestTimeout:
@@ -100,17 +100,16 @@ std::string ErrorCategory::message(int code) const {
"a timely response from the upstream "
"server.";
- case ErrorCode::ProtocolVersionNotSupported:
+ case ErrorCode::ProtocolUnsupported:
return "The server does not support the protocol version used in the "
"request.";
case ErrorCode::InsufficientStorage:
return "The server is unable to store the representation needed to "
"complete the request.";
-
- default:
- return "Not-Implemented";
}
+
+ return "";
}
ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/src/main/cpp/client/ClientManagerImpl.cpp b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
index 0c6e155..05a670b 100644
--- a/cpp/src/main/cpp/client/ClientManagerImpl.cpp
+++ b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
@@ -478,32 +478,64 @@ void ClientManagerImpl::resolveRoute(const std::string& target_host, const Metad
}
auto ptr = std::make_shared<TopicRouteData>(std::move(message_queues));
cb(ec, ptr);
- } break;
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_ACCESS_POINT: {
+ SPDLOG_WARN("IllegalAccessPoint: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::IllegalAccessPoint;
+ cb(ec, nullptr);
+ break;
+ }
+
case rmq::Code::UNAUTHORIZED: {
SPDLOG_WARN("Unauthorized: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::Unauthorized;
cb(ec, nullptr);
- } break;
- case rmq::Code::FORBIDDEN: {
- SPDLOG_WARN("Forbidden: {}. Host={}", status.message(), invocation_context->remote_address);
- ec = ErrorCode::Forbidden;
- cb(ec, nullptr);
- } break;
+ break;
+ }
+
case rmq::Code::TOPIC_NOT_FOUND: {
SPDLOG_WARN("TopicNotFound: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::NotFound;
cb(ec, nullptr);
- } break;
+ break;
+ }
+
+ case rmq::Code::TOO_MANY_REQUESTS: {
+ SPDLOG_WARN("TooManyRequest: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::TooManyRequest;
+ cb(ec, nullptr);
+ break;
+ }
+
+ case rmq::Code::CLIENT_ID_REQUIRED: {
+ SPDLOG_ERROR("ClientIdRequired: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::ClientIdRequired;
+ cb(ec, nullptr);
+ break;
+ }
+
case rmq::Code::INTERNAL_SERVER_ERROR: {
SPDLOG_WARN("InternalServerError: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::InternalServerError;
cb(ec, nullptr);
- } break;
+ break;
+ }
+
+ case rmq::Code::PROXY_TIMEOUT: {
+ SPDLOG_WARN("GatewayTimeout: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::GatewayTimeout;
+ cb(ec, nullptr);
+ break;
+ }
+
default: {
SPDLOG_WARN("NotImplement: Please upgrade to latest SDK release. Host={}", invocation_context->remote_address);
ec = ErrorCode::NotImplemented;
cb(ec, nullptr);
- } break;
+ break;
+ }
}
};
invocation_context->callback = callback;
diff --git a/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp b/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp
index 223c879..4667de4 100644
--- a/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp
+++ b/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp
@@ -69,7 +69,7 @@ void ReceiveMessageStreamReader::OnReadDone(bool ok) {
}
case rmq::Code::CONSUMER_GROUP_NOT_FOUND: {
- ec_ = ErrorCode::GroupNotFound;
+ ec_ = ErrorCode::ConsumerGroupNotFound;
break;
}
case rmq::Code::TOO_MANY_REQUESTS: {
[rocketmq-clients] 05/18: Complete error handling for QueryAssignment
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit 1b1dfd064ab85418cce9668d32fde3ffbef022ca
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Thu Jun 30 17:07:48 2022 +0800
Complete error handling for QueryAssignment
---
cpp/src/main/cpp/client/ClientManagerImpl.cpp | 70 +++++++++++++++++++++++----
1 file changed, 60 insertions(+), 10 deletions(-)
diff --git a/cpp/src/main/cpp/client/ClientManagerImpl.cpp b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
index b6e8a2e..8562c6d 100644
--- a/cpp/src/main/cpp/client/ClientManagerImpl.cpp
+++ b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
@@ -467,7 +467,7 @@ bool ClientManagerImpl::send(const std::string& target_host, const Metadata& met
break;
}
}
-
+
cb(ec, send_receipt);
};
@@ -683,25 +683,75 @@ void ClientManagerImpl::queryAssignment(
std::error_code ec;
switch (status.code()) {
case rmq::Code::OK: {
- SPDLOG_DEBUG("Query assignment OK");
- } break;
+ SPDLOG_DEBUG("Query assignment OK. Host={}", invocation_context->remote_address);
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_ACCESS_POINT: {
+ SPDLOG_WARN("IllegalAccessPoint: {}, host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::IllegalAccessPoint;
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_TOPIC: {
+ SPDLOG_WARN("IllegalAccessPoint: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::IllegalTopic;
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_CONSUMER_GROUP: {
+ SPDLOG_WARN("IllegalConsumerGroup: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::IllegalConsumerGroup;
+ break;
+ }
+
+ case rmq::Code::CLIENT_ID_REQUIRED: {
+ SPDLOG_WARN("ClientIdRequired: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::ClientIdRequired;
+ break;
+ }
+
case rmq::Code::UNAUTHORIZED: {
SPDLOG_WARN("Unauthorized: {}, host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::Unauthorized;
- } break;
+ break;
+ }
+
case rmq::Code::FORBIDDEN: {
SPDLOG_WARN("Forbidden: {}, host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::Forbidden;
- } break;
+ break;
+ }
+
+ case rmq::Code::TOPIC_NOT_FOUND: {
+ SPDLOG_WARN("TopicNotFound: {}, host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::TopicNotFound;
+ break;
+ }
+
+ case rmq::Code::CONSUMER_GROUP_NOT_FOUND: {
+ SPDLOG_WARN("ConsumerGroupNotFound: {}, host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::ConsumerGroupNotFound;
+ break;
+ }
+
case rmq::Code::INTERNAL_SERVER_ERROR: {
SPDLOG_WARN("InternalServerError: {}, host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::InternalServerError;
- } break;
+ break;
+ }
+
+ case rmq::Code::PROXY_TIMEOUT: {
+ SPDLOG_WARN("GatewayTimeout: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::GatewayTimeout;
+ break;
+ }
+
default: {
- SPDLOG_WARN("NotImplemented: please upgrade SDK to latest release. Host={}",
- invocation_context->remote_address);
- ec = ErrorCode::NotImplemented;
- } break;
+ SPDLOG_WARN("NotSupported: please upgrade SDK to latest release. Host={}", invocation_context->remote_address);
+ ec = ErrorCode::NotSupported;
+ break;
+ }
}
cb(ec, invocation_context->response);
};
[rocketmq-clients] 09/18: Add error handling for NotifyClientTermination
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit 9c859cbf74252196a352119d6fc1c92149800842
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Thu Jun 30 19:18:32 2022 +0800
Add error handling for NotifyClientTermination
---
cpp/src/main/cpp/client/ClientManagerImpl.cpp | 14 ++++++++++++--
1 file changed, 12 insertions(+), 2 deletions(-)
diff --git a/cpp/src/main/cpp/client/ClientManagerImpl.cpp b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
index 1602769..8d182d8 100644
--- a/cpp/src/main/cpp/client/ClientManagerImpl.cpp
+++ b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
@@ -1392,26 +1392,36 @@ std::error_code ClientManagerImpl::notifyClientTermination(const std::string& ta
SPDLOG_DEBUG("NotifyClientTermination OK. host={}", target_host);
break;
}
+
+ case rmq::Code::ILLEGAL_CONSUMER_GROUP: {
+ SPDLOG_ERROR("IllegalConsumerGroup: {}. Host={}", status.message(), target_host);
+ ec = ErrorCode::IllegalConsumerGroup;
+ break;
+ }
+
case rmq::Code::INTERNAL_SERVER_ERROR: {
SPDLOG_WARN("InternalServerError: Cause={}, host={}", status.message(), target_host);
ec = ErrorCode::InternalServerError;
break;
}
+
case rmq::Code::UNAUTHORIZED: {
SPDLOG_WARN("Unauthorized due to lack of valid authentication credentials: Cause={}, host={}", status.message(),
target_host);
ec = ErrorCode::Unauthorized;
break;
}
+
case rmq::Code::FORBIDDEN: {
SPDLOG_WARN("Forbidden due to insufficient permission to the resource: Cause={}, host={}", status.message(),
target_host);
ec = ErrorCode::Forbidden;
break;
}
+
default: {
- SPDLOG_WARN("NotImplemented. Please upgrade to latest SDK release. host={}", target_host);
- ec = ErrorCode::NotImplemented;
+ SPDLOG_WARN("NotSupported. Please upgrade to latest SDK release. host={}", target_host);
+ ec = ErrorCode::NotSupported;
break;
}
}
[rocketmq-clients] 01/18: Sync and update protobuf files to v2.0-alpha
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit 00080c3dd8f32ec7ef4cb94679dfecd496439173
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Thu Jun 30 10:51:43 2022 +0800
Sync and update protobuf files to v2.0-alpha
---
cpp/proto/apache/rocketmq/v2/definition.proto | 161 +++++++++++--------------
cpp/proto/apache/rocketmq/v2/service.proto | 7 ++
cpp/src/main/cpp/rocketmq/PushConsumerImpl.cpp | 2 +-
3 files changed, 81 insertions(+), 89 deletions(-)
diff --git a/cpp/proto/apache/rocketmq/v2/definition.proto b/cpp/proto/apache/rocketmq/v2/definition.proto
index 58bf06d..ab1fd09 100644
--- a/cpp/proto/apache/rocketmq/v2/definition.proto
+++ b/cpp/proto/apache/rocketmq/v2/definition.proto
@@ -292,106 +292,74 @@ message Assignment {
}
enum Code {
- // Success.
- OK = 0;
+ CODE_UNSPECIFIED = 0;
+
+ // Generic code for success.
+ OK = 20000;
+
+ // Generic code for multiple return results.
+ MULTIPLE_RESULTS = 30000;
+
+ // Generic code for bad request, indicating that required fields or headers are missing.
+ BAD_REQUEST = 40000;
// Format of access point is illegal.
- ILLEGAL_ACCESS_POINT = 1;
+ ILLEGAL_ACCESS_POINT = 40001;
// Format of topic is illegal.
- ILLEGAL_TOPIC = 2;
+ ILLEGAL_TOPIC = 40002;
// Format of consumer group is illegal.
- ILLEGAL_CONSUMER_GROUP = 3;
+ ILLEGAL_CONSUMER_GROUP = 40003;
// Format of message tag is illegal.
- ILLEGAL_MESSAGE_TAG = 4;
+ ILLEGAL_MESSAGE_TAG = 40004;
// Format of message key is illegal.
- ILLEGAL_MESSAGE_KEY = 5;
- // Size of message keys exceeds the threshold.
- MESSAGE_KEYS_TOO_LARGE = 6;
+ ILLEGAL_MESSAGE_KEY = 40005;
// Format of message group is illegal.
- ILLEGAL_MESSAGE_GROUP = 7;
+ ILLEGAL_MESSAGE_GROUP = 40006;
// Format of message property key is illegal.
- ILLEGAL_MESSAGE_PROPERTY_KEY = 8;
+ ILLEGAL_MESSAGE_PROPERTY_KEY = 40007;
// Message properties total size exceeds the threshold.
- MESSAGE_PROPERTIES_TOO_LARGE = 9;
+ MESSAGE_PROPERTIES_TOO_LARGE = 40008;
// Message body size exceeds the threshold.
- MESSAGE_BODY_TOO_LARGE = 10;
-
- // User does not have the permission to operate.
- // See https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/403
- FORBIDDEN = 403;
-
- // Code indicates that the client request has not been completed
- // because it lacks valid authentication credentials for the
- // requested resource.
- // See https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/401
- UNAUTHORIZED = 401;
-
- // Topic resource does not exist.
- TOPIC_NOT_FOUND = 13;
-
- // Consumer group resource does not exist.
- CONSUMER_GROUP_NOT_FOUND = 14;
-
- // Not allowed to verify message. Chances are that you are verifying
- // a FIFO message, as is violating FIFO semantics.
- VERIFY_MESSAGE_FORBIDDEN = 15;
-
- // Failed to consume message.
- FAILED_TO_CONSUME_MESSAGE = 16;
-
- // Message is corrupted.
- MESSAGE_CORRUPTED = 17;
-
- // Too many requests are made in short period of duration.
- // Requests are throttled.
- TOO_MANY_REQUESTS = 18;
-
- // Expired receipt-handle is used when trying to acknowledge or change
- // invisible duration of a message
- RECEIPT_HANDLE_EXPIRED = 19;
-
- // Message property is not match the message type.
- MESSAGE_PROPERTY_DOES_NOT_MATCH_MESSAGE_TYPE = 20;
-
- // Format of message id is illegal.
- ILLEGAL_MESSAGE_ID = 21;
-
+ MESSAGE_BODY_TOO_LARGE = 40009;
// Transaction id is invalid.
- INVALID_TRANSACTION_ID = 22;
-
+ INVALID_TRANSACTION_ID = 40010;
+ // Format of message id is illegal.
+ ILLEGAL_MESSAGE_ID = 40011;
// Format of filter expression is illegal.
- ILLEGAL_FILTER_EXPRESSION = 23;
-
+ ILLEGAL_FILTER_EXPRESSION = 40012;
// Receipt handle of message is invalid.
- INVALID_RECEIPT_HANDLE = 24;
-
- // Message persistence timeout.
- MASTER_PERSISTENCE_TIMEOUT = 25;
-
- // Slave persistence timeout.
- SLAVE_PERSISTENCE_TIMEOUT = 26;
+ INVALID_RECEIPT_HANDLE = 40013;
+ // Message property is not match the message type.
+ MESSAGE_PROPERTY_DOES_NOT_MATCH_MESSAGE_TYPE = 40014;
+ // Client type could not be recognized.
+ UNRECOGNIZED_CLIENT_TYPE = 40015;
+ // Message is corrupted.
+ MESSAGE_CORRUPTED = 40016;
+ // Request is rejected due to missing of x-mq-client-id header.
+ CLIENT_ID_REQUIRED = 40017;
- // The HA-mechanism is not working now.
- HA_NOT_AVAILABLE = 27;
+ // Generic code indicates that the client request lacks valid authentication
+ // credentials for the requested resource.
+ UNAUTHORIZED = 40100;
- // Operation is not allowed in current version.
- VERSION_UNSUPPORTED = 28;
+ // Generic code for the case that user does not have the permission to operate.
+ FORBIDDEN = 40300;
+ // Generic code for resource not found.
+ NOT_FOUND = 40400;
// Message not found from server.
- MESSAGE_NOT_FOUND = 29;
-
- // Message offset is illegal.
- ILLEGAL_MESSAGE_OFFSET = 30;
-
- // Illegal message is for the sake of backward compatibility. In most case,
- // more definitive code is better, e.g. `ILLEGAL_MESSAGE_TAG`.
- ILLEGAL_MESSAGE = 31;
-
- // Client type could not be recognized.
- UNRECOGNIZED_CLIENT_TYPE = 32;
+ MESSAGE_NOT_FOUND = 40401;
+ // Topic resource does not exist.
+ TOPIC_NOT_FOUND = 40402;
+ // Consumer group resource does not exist.
+ CONSUMER_GROUP_NOT_FOUND = 40403;
- // Return different results for entries in composite request.
- MULTIPLE_RESULTS = 33;
+ // Generic code indicates that too many requests are made in short period of duration.
+ // Requests are throttled.
+ TOO_MANY_REQUESTS = 42900;
+ // Generic code indicates that server/client encountered an unexpected
+ // condition that prevented it from fulfilling the request.
+ INTERNAL_ERROR = 50000;
// Code indicates that the server encountered an unexpected condition
// that prevented it from fulfilling the request.
// This error response is a generic "catch-all" response.
@@ -401,17 +369,34 @@ enum Code {
// to prevent the error from happening again in the future.
//
// See https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/500
- INTERNAL_SERVER_ERROR = 500;
+ INTERNAL_SERVER_ERROR = 50001;
+ // The HA-mechanism is not working now.
+ HA_NOT_AVAILABLE = 50002;
- // Code means that the server or client does not support the functionality
- // required to fulfill the request.
- NOT_IMPLEMENTED = 501;
+ // Generic code means that the server or client does not support the
+ // functionality required to fulfill the request.
+ NOT_IMPLEMENTED = 50100;
+ // Generic code for timeout.
+ TIMEOUT = 50400;
+ // Message persistence timeout.
+ MASTER_PERSISTENCE_TIMEOUT = 50401;
+ // Slave persistence timeout.
+ SLAVE_PERSISTENCE_TIMEOUT = 50402;
// Code indicates that the server, while acting as a gateway or proxy,
// did not get a response in time from the upstream server that
// it needed in order to complete the request.
- // See https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/504
- GATEWAY_TIMEOUT = 504;
+ PROXY_TIMEOUT = 50403;
+
+ UNSUPPORTED = 50500;
+ // Operation is not allowed in current version.
+ VERSION_UNSUPPORTED = 50501;
+ // Not allowed to verify message. Chances are that you are verifying
+ // a FIFO message, as is violating FIFO semantics.
+ VERIFY_FIFO_MESSAGE_UNSUPPORTED = 50502;
+
+ // Generic code for failed message consumption.
+ FAILED_TO_CONSUME_MESSAGE = 60000;
}
message Status {
diff --git a/cpp/proto/apache/rocketmq/v2/service.proto b/cpp/proto/apache/rocketmq/v2/service.proto
index fa2a2f0..c5d4cce 100644
--- a/cpp/proto/apache/rocketmq/v2/service.proto
+++ b/cpp/proto/apache/rocketmq/v2/service.proto
@@ -16,6 +16,7 @@
syntax = "proto3";
import "google/protobuf/duration.proto";
+import "google/protobuf/timestamp.proto";
import "apache/rocketmq/v2/definition.proto";
@@ -101,6 +102,8 @@ message ReceiveMessageResponse {
oneof content {
Status status = 1;
Message message = 2;
+ // The timestamp that brokers start to deliver status line or message.
+ google.protobuf.Timestamp delivery_timestamp = 3;
}
}
@@ -211,6 +214,10 @@ message Publishing {
// reject the request. As a result, it is advisable that Producer performs
// client-side check validation.
int32 max_body_size = 3;
+
+ // When `validate_message_type` flag set `false`, no need to validate message's type
+ // with messageQueue's `accept_message_types` before publising.
+ bool validate_message_type = 4;
}
message Subscription {
diff --git a/cpp/src/main/cpp/rocketmq/PushConsumerImpl.cpp b/cpp/src/main/cpp/rocketmq/PushConsumerImpl.cpp
index 89dc112..2acff23 100644
--- a/cpp/src/main/cpp/rocketmq/PushConsumerImpl.cpp
+++ b/cpp/src/main/cpp/rocketmq/PushConsumerImpl.cpp
@@ -550,7 +550,7 @@ void PushConsumerImpl::onVerifyMessage(MessageConstSharedPtr message, std::funct
cmd.mutable_status()->set_message("Unexpected exception raised");
}
} else {
- cmd.mutable_status()->set_code(rmq::Code::VERIFY_MESSAGE_FORBIDDEN);
+ cmd.mutable_status()->set_code(rmq::Code::VERIFY_FIFO_MESSAGE_UNSUPPORTED);
cmd.mutable_status()->set_message("Unsupported Operation For FIFO Message");
}
} else {
[rocketmq-clients] 03/18: Complete error handling for Heartbeat
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit 4a9426ea6484ac1f147d0e3c1b43de60db88bcce
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Thu Jun 30 15:50:59 2022 +0800
Complete error handling for Heartbeat
---
cpp/src/main/cpp/client/ClientManagerImpl.cpp | 53 ++++++++++++++++++++++-----
1 file changed, 43 insertions(+), 10 deletions(-)
diff --git a/cpp/src/main/cpp/client/ClientManagerImpl.cpp b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
index 05a670b..e4be88b 100644
--- a/cpp/src/main/cpp/client/ClientManagerImpl.cpp
+++ b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
@@ -16,6 +16,8 @@
*/
#include "ClientManagerImpl.h"
+#include <apache/rocketmq/v2/definition.pb.h>
+
#include <atomic>
#include <cassert>
#include <chrono>
@@ -229,26 +231,57 @@ void ClientManagerImpl::heartbeat(const std::string& target_host, const Metadata
switch (status.code()) {
case rmq::Code::OK: {
cb(ec, invocation_context->response);
- } break;
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_CONSUMER_GROUP: {
+ SPDLOG_ERROR("IllegalConsumerGroup: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::IllegalConsumerGroup;
+ break;
+ }
+
+ case rmq::Code::TOO_MANY_REQUESTS: {
+ SPDLOG_WARN("TooManyRequest: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::TooManyRequest;
+ cb(ec, invocation_context->response);
+ break;
+ }
+
case rmq::Code::UNAUTHORIZED: {
- SPDLOG_WARN("Unauthorized: {}, host={}", status.message(), invocation_context->remote_address);
+ SPDLOG_WARN("Unauthorized: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::Unauthorized;
cb(ec, invocation_context->response);
- } break;
- case rmq::Code::FORBIDDEN: {
- SPDLOG_WARN("Forbidden: {}, host={}", status.message(), invocation_context->remote_address);
- ec = ErrorCode::Forbidden;
+ break;
+ }
+
+ case rmq::Code::UNRECOGNIZED_CLIENT_TYPE: {
+ SPDLOG_ERROR("UnsupportedClientType: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::UnsupportedClientType;
cb(ec, invocation_context->response);
- } break;
+ break;
+ }
+
+ case rmq::Code::CLIENT_ID_REQUIRED: {
+ SPDLOG_ERROR("ClientIdRequired: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::ClientIdRequired;
+ cb(ec, invocation_context->response);
+ break;
+ }
+
case rmq::Code::INTERNAL_SERVER_ERROR: {
SPDLOG_WARN("InternalServerError: {}, host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::InternalServerError;
cb(ec, invocation_context->response);
- } break;
+ break;
+ }
+
default: {
- SPDLOG_WARN("NotImplemented: Please upgrade SDK to latest release. Message={}, host={}", status.message(),
+ SPDLOG_WARN("NotSupported: Please upgrade SDK to latest release. Message={}, host={}", status.message(),
invocation_context->remote_address);
- } break;
+ ec = ErrorCode::NotSupported;
+ cb(ec, invocation_context->response);
+ break;
+ }
}
};
[rocketmq-clients] 11/18: Translate ClientIdRequired to InternalClientError
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit fcf63ce82caa5ebfea51fa7d10b8e655bfa82fcb
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Thu Jun 30 19:28:35 2022 +0800
Translate ClientIdRequired to InternalClientError
---
cpp/src/main/cpp/client/ClientManagerImpl.cpp | 14 ++++++--------
cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp | 2 +-
2 files changed, 7 insertions(+), 9 deletions(-)
diff --git a/cpp/src/main/cpp/client/ClientManagerImpl.cpp b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
index 0f4c7f5..1f9eec6 100644
--- a/cpp/src/main/cpp/client/ClientManagerImpl.cpp
+++ b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
@@ -263,7 +263,7 @@ void ClientManagerImpl::heartbeat(const std::string& target_host, const Metadata
case rmq::Code::CLIENT_ID_REQUIRED: {
SPDLOG_ERROR("ClientIdRequired: {}. Host={}", status.message(), invocation_context->remote_address);
- ec = ErrorCode::ClientIdRequired;
+ ec = ErrorCode::InternalClientError;
cb(ec, invocation_context->response);
break;
}
@@ -633,7 +633,7 @@ void ClientManagerImpl::resolveRoute(const std::string& target_host, const Metad
case rmq::Code::CLIENT_ID_REQUIRED: {
SPDLOG_ERROR("ClientIdRequired: {}. Host={}", status.message(), invocation_context->remote_address);
- ec = ErrorCode::ClientIdRequired;
+ ec = ErrorCode::InternalClientError;
cb(ec, nullptr);
break;
}
@@ -707,7 +707,7 @@ void ClientManagerImpl::queryAssignment(
case rmq::Code::CLIENT_ID_REQUIRED: {
SPDLOG_WARN("ClientIdRequired: {}. Host={}", status.message(), invocation_context->remote_address);
- ec = ErrorCode::ClientIdRequired;
+ ec = ErrorCode::InternalClientError;
break;
}
@@ -1039,7 +1039,7 @@ void ClientManagerImpl::ack(const std::string& target, const Metadata& metadata,
case rmq::Code::CLIENT_ID_REQUIRED: {
SPDLOG_WARN("ClientIdRequired: {}, host={}", status.message(), invocation_context->remote_address);
- ec = ErrorCode::ClientIdRequired;
+ ec = ErrorCode::InternalClientError;
break;
}
@@ -1251,7 +1251,7 @@ void ClientManagerImpl::endTransaction(
case rmq::Code::CLIENT_ID_REQUIRED: {
SPDLOG_WARN("ClientIdRequired: {}, host={}", status.message(), peer_address);
- ec = ErrorCode::ClientIdRequired;
+ ec = ErrorCode::InternalClientError;
break;
}
@@ -1352,9 +1352,7 @@ void ClientManagerImpl::forwardMessageToDeadLetterQueue(const std::string& targe
case rmq::Code::CLIENT_ID_REQUIRED: {
SPDLOG_WARN("IllegalTopic: {}. Host={}", status.message(), peer_address);
-
- // TODO: translate to client internal error?
- ec = ErrorCode::InternalServerError;
+ ec = ErrorCode::InternalClientError;
break;
}
diff --git a/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp b/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp
index 0684f51..af99874 100644
--- a/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp
+++ b/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp
@@ -82,7 +82,7 @@ void ReceiveMessageStreamReader::OnReadDone(bool ok) {
}
case rmq::Code::CLIENT_ID_REQUIRED: {
- ec_ = ErrorCode::ClientIdRequired;
+ ec_ = ErrorCode::InternalClientError;
break;
}
[rocketmq-clients] 14/18: Add mirror for rules_proto
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit cde80b0d81cd57cf24367379e27498bc399a8a08
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Fri Jul 1 14:07:14 2022 +0800
Add mirror for rules_proto
---
cpp/bazel/rocketmq_deps.bzl | 11 +++++++++++
1 file changed, 11 insertions(+)
diff --git a/cpp/bazel/rocketmq_deps.bzl b/cpp/bazel/rocketmq_deps.bzl
index b92d20d..106b7a9 100644
--- a/cpp/bazel/rocketmq_deps.bzl
+++ b/cpp/bazel/rocketmq_deps.bzl
@@ -187,4 +187,15 @@ def rocketmq_deps():
"https://mirror.bazel.build/github.com/bazelbuild/rules_go/releases/download/v0.33.0/rules_go-v0.33.0.zip",
"https://github.com/bazelbuild/rules_go/releases/download/v0.33.0/rules_go-v0.33.0.zip",
],
+ )
+
+ maybe(
+ http_archive,
+ name = "rules_proto",
+ sha256 = "e017528fd1c91c5a33f15493e3a398181a9e821a804eb7ff5acdd1d2d6c2b18d",
+ strip_prefix = "rules_proto-4.0.0-3.20.0",
+ urls = [
+ "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/rules_proto/rules_proto-4.0.0.tar.gz",
+ "https://github.com/bazelbuild/rules_proto/archive/refs/tags/4.0.0-3.20.0.tar.gz",
+ ],
)
\ No newline at end of file
[rocketmq-clients] 04/18: Finish error handling of SendMessage
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit efdd9e10b866ae0d4badb90a50d6419b28cae210
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Thu Jun 30 16:53:08 2022 +0800
Finish error handling of SendMessage
---
cpp/api/rocketmq/ErrorCode.h | 2 +-
cpp/proto/apache/rocketmq/v2/definition.proto | 55 +++++++----
cpp/src/main/cpp/base/ErrorCategory.cpp | 22 ++++-
cpp/src/main/cpp/client/ClientManagerImpl.cpp | 109 +++++++++++++++++++--
.../main/cpp/client/ReceiveMessageStreamReader.cpp | 2 +-
5 files changed, 158 insertions(+), 32 deletions(-)
diff --git a/cpp/api/rocketmq/ErrorCode.h b/cpp/api/rocketmq/ErrorCode.h
index 14ec504..e911533 100644
--- a/cpp/api/rocketmq/ErrorCode.h
+++ b/cpp/api/rocketmq/ErrorCode.h
@@ -151,7 +151,7 @@ enum class ErrorCode : int {
* amount of time.
*
*/
- TooManyRequest = 42900,
+ TooManyRequests = 42900,
/**
* @brief The server is unwilling to process the request because either an
diff --git a/cpp/proto/apache/rocketmq/v2/definition.proto b/cpp/proto/apache/rocketmq/v2/definition.proto
index ab1fd09..67520fe 100644
--- a/cpp/proto/apache/rocketmq/v2/definition.proto
+++ b/cpp/proto/apache/rocketmq/v2/definition.proto
@@ -316,31 +316,30 @@ enum Code {
ILLEGAL_MESSAGE_GROUP = 40006;
// Format of message property key is illegal.
ILLEGAL_MESSAGE_PROPERTY_KEY = 40007;
- // Message properties total size exceeds the threshold.
- MESSAGE_PROPERTIES_TOO_LARGE = 40008;
- // Message body size exceeds the threshold.
- MESSAGE_BODY_TOO_LARGE = 40009;
// Transaction id is invalid.
- INVALID_TRANSACTION_ID = 40010;
+ INVALID_TRANSACTION_ID = 40008;
// Format of message id is illegal.
- ILLEGAL_MESSAGE_ID = 40011;
+ ILLEGAL_MESSAGE_ID = 40009;
// Format of filter expression is illegal.
- ILLEGAL_FILTER_EXPRESSION = 40012;
+ ILLEGAL_FILTER_EXPRESSION = 40010;
// Receipt handle of message is invalid.
- INVALID_RECEIPT_HANDLE = 40013;
- // Message property is not match the message type.
- MESSAGE_PROPERTY_DOES_NOT_MATCH_MESSAGE_TYPE = 40014;
+ INVALID_RECEIPT_HANDLE = 40011;
+ // Message property conflicts with its type.
+ MESSAGE_PROPERTY_CONFLICT_WITH_TYPE = 40012;
// Client type could not be recognized.
- UNRECOGNIZED_CLIENT_TYPE = 40015;
+ UNRECOGNIZED_CLIENT_TYPE = 40013;
// Message is corrupted.
- MESSAGE_CORRUPTED = 40016;
+ MESSAGE_CORRUPTED = 40014;
// Request is rejected due to missing of x-mq-client-id header.
- CLIENT_ID_REQUIRED = 40017;
+ CLIENT_ID_REQUIRED = 40015;
// Generic code indicates that the client request lacks valid authentication
// credentials for the requested resource.
UNAUTHORIZED = 40100;
+ // Generic code indicates that the account is suspended due to overdue of payment.
+ PAYMENT_REQUIRED = 40200;
+
// Generic code for the case that user does not have the permission to operate.
FORBIDDEN = 40300;
@@ -353,10 +352,29 @@ enum Code {
// Consumer group resource does not exist.
CONSUMER_GROUP_NOT_FOUND = 40403;
+ // Generic code representing client side timeout when connecting to, reading data from, or write data to server.
+ REQUEST_TIMEOUT = 40800;
+
+ // Generic code represents that the request entity is larger than limits defined by server.
+ PAYLOAD_TOO_LARGE = 41300;
+ // Message body size exceeds the threshold.
+ MESSAGE_BODY_TOO_LARGE = 41301;
+
+ // Generic code for use cases where pre-conditions are not met.
+ // For example, if a producer instance is used to publish messages without prior start() invocation,
+ // this error code will be raised.
+ PRECONDITION_FAILED = 42800;
+
// Generic code indicates that too many requests are made in short period of duration.
// Requests are throttled.
TOO_MANY_REQUESTS = 42900;
+ // Generic code for the case that the server is unwilling to process the request because its header fields are too large.
+ // The request may be resubmitted after reducing the size of the request header fields.
+ REQUEST_HEADER_FIELDS_TOO_LARGE = 43100;
+ // Message properties total size exceeds the threshold.
+ MESSAGE_PROPERTIES_TOO_LARGE = 43101;
+
// Generic code indicates that server/client encountered an unexpected
// condition that prevented it from fulfilling the request.
INTERNAL_ERROR = 50000;
@@ -377,17 +395,16 @@ enum Code {
// functionality required to fulfill the request.
NOT_IMPLEMENTED = 50100;
- // Generic code for timeout.
- TIMEOUT = 50400;
+ // Generic code represents that the server, which acts as a gateway or proxy,
+ // does not get an satisfied response in time from its upstream servers.
+ // See https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/504
+ PROXY_TIMEOUT = 50400;
// Message persistence timeout.
MASTER_PERSISTENCE_TIMEOUT = 50401;
// Slave persistence timeout.
SLAVE_PERSISTENCE_TIMEOUT = 50402;
- // Code indicates that the server, while acting as a gateway or proxy,
- // did not get a response in time from the upstream server that
- // it needed in order to complete the request.
- PROXY_TIMEOUT = 50403;
+ // Generic code for unsupported operation.
UNSUPPORTED = 50500;
// Operation is not allowed in current version.
VERSION_UNSUPPORTED = 50501;
diff --git a/cpp/src/main/cpp/base/ErrorCategory.cpp b/cpp/src/main/cpp/base/ErrorCategory.cpp
index 4f1b290..5ca2528 100644
--- a/cpp/src/main/cpp/base/ErrorCategory.cpp
+++ b/cpp/src/main/cpp/base/ErrorCategory.cpp
@@ -16,6 +16,8 @@
*/
#include "rocketmq/ErrorCategory.h"
+#include "rocketmq/ErrorCode.h"
+
ROCKETMQ_NAMESPACE_BEGIN
std::string ErrorCategory::message(int code) const {
@@ -65,7 +67,7 @@ std::string ErrorCategory::message(int code) const {
case ErrorCode::PreconditionRequired:
return "State of dependent procedure is not right";
- case ErrorCode::TooManyRequest:
+ case ErrorCode::TooManyRequests:
return "Quota exchausted. The user has sent too many requests in a given "
"amount of time.";
@@ -107,6 +109,24 @@ std::string ErrorCategory::message(int code) const {
case ErrorCode::InsufficientStorage:
return "The server is unable to store the representation needed to "
"complete the request.";
+
+ case ErrorCode::MultipleResults:
+ return "Multiple results are available";
+
+ case ErrorCode::IllegalAccessPoint:
+ return "Access point is either malformed or invalid";
+
+ case ErrorCode::IllegalTopic: {
+ return "Topic is illegal either due to length is too long or invalid character is included";
+ }
+
+ case ErrorCode::IllegalConsumerGroup: {
+ return "ConsumerGroup is illegal due to its length is too long or invalid character is included";
+ }
+
+ case ErrorCode::IllegalMessageTag: {
+ return "Format of message tag is illegal.";
+ }
}
return "";
diff --git a/cpp/src/main/cpp/client/ClientManagerImpl.cpp b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
index e4be88b..b6e8a2e 100644
--- a/cpp/src/main/cpp/client/ClientManagerImpl.cpp
+++ b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
@@ -242,7 +242,7 @@ void ClientManagerImpl::heartbeat(const std::string& target_host, const Metadata
case rmq::Code::TOO_MANY_REQUESTS: {
SPDLOG_WARN("TooManyRequest: {}. Host={}", status.message(), invocation_context->remote_address);
- ec = ErrorCode::TooManyRequest;
+ ec = ErrorCode::TooManyRequests;
cb(ec, invocation_context->response);
break;
}
@@ -352,33 +352,122 @@ bool ClientManagerImpl::send(const std::string& target_host, const Metadata& met
send_receipt.message_id = invocation_context->response.entries().begin()->message_id();
break;
}
+
+ case rmq::Code::ILLEGAL_TOPIC: {
+ SPDLOG_ERROR("IllegalTopic: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::IllegalTopic;
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_MESSAGE_TAG: {
+ SPDLOG_ERROR("IllegalMessageTag: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::IllegalMessageTag;
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_MESSAGE_KEY: {
+ SPDLOG_ERROR("IllegalMessageKey: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::IllegalMessageKey;
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_MESSAGE_GROUP: {
+ SPDLOG_ERROR("IllegalMessageGroup: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::IllegalMessageGroup;
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_MESSAGE_PROPERTY_KEY: {
+ SPDLOG_ERROR("IllegalMessageProperty: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::IllegalMessageProperty;
+ break;
+ }
+
+ case rmq::Code::MESSAGE_PROPERTIES_TOO_LARGE: {
+ SPDLOG_ERROR("MessagePropertiesTooLarge: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::MessagePropertiesTooLarge;
+ break;
+ }
+
+ case rmq::Code::MESSAGE_BODY_TOO_LARGE: {
+ SPDLOG_ERROR("MessageBodyTooLarge: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::MessageBodyTooLarge;
+ break;
+ }
+
case rmq::Code::TOPIC_NOT_FOUND: {
- SPDLOG_WARN("TopicNotFound: {}", status.message());
+ SPDLOG_WARN("TopicNotFound: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::TopicNotFound;
+ break;
+ }
+
+ case rmq::Code::NOT_FOUND: {
+ SPDLOG_WARN("NotFound: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::NotFound;
break;
}
+
case rmq::Code::UNAUTHORIZED: {
- SPDLOG_WARN("Unauthenticated: {}", status.message());
+ SPDLOG_WARN("Unauthenticated: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::Unauthorized;
break;
}
+
case rmq::Code::FORBIDDEN: {
- SPDLOG_WARN("Forbidden: {}", status.message());
+ SPDLOG_WARN("Forbidden: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::Forbidden;
- cb(ec, send_receipt);
break;
}
+
+ case rmq::Code::MESSAGE_CORRUPTED: {
+ SPDLOG_WARN("MessageCorrupted: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::MessageCorrupted;
+ break;
+ }
+
+ case rmq::Code::TOO_MANY_REQUESTS: {
+ SPDLOG_WARN("TooManyRequest: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::TooManyRequests;
+ break;
+ }
+
case rmq::Code::INTERNAL_SERVER_ERROR: {
- SPDLOG_WARN("InternalServerError: {}", status.message());
+ SPDLOG_WARN("InternalServerError: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::InternalServerError;
break;
}
+
+ case rmq::Code::HA_NOT_AVAILABLE: {
+ SPDLOG_WARN("InternalServerError: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::InternalServerError;
+ break;
+ }
+
+ case rmq::Code::PROXY_TIMEOUT: {
+ SPDLOG_WARN("GatewayTimeout: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::GatewayTimeout;
+ break;
+ }
+
+ case rmq::Code::MASTER_PERSISTENCE_TIMEOUT: {
+ SPDLOG_WARN("GatewayTimeout: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::GatewayTimeout;
+ break;
+ }
+
+ case rmq::Code::SLAVE_PERSISTENCE_TIMEOUT: {
+ SPDLOG_WARN("GatewayTimeout: {}. Host={}", status.message(), invocation_context->remote_address);
+ ec = ErrorCode::GatewayTimeout;
+ break;
+ }
+
default: {
- SPDLOG_WARN("Unsupported status code. Check and upgrade SDK to the latest");
- ec = ErrorCode::NotImplemented;
+ SPDLOG_WARN("NotSupported: Check and upgrade SDK to the latest. Host={}", invocation_context->remote_address);
+ ec = ErrorCode::NotSupported;
break;
}
}
+
cb(ec, send_receipt);
};
@@ -537,7 +626,7 @@ void ClientManagerImpl::resolveRoute(const std::string& target_host, const Metad
case rmq::Code::TOO_MANY_REQUESTS: {
SPDLOG_WARN("TooManyRequest: {}. Host={}", status.message(), invocation_context->remote_address);
- ec = ErrorCode::TooManyRequest;
+ ec = ErrorCode::TooManyRequests;
cb(ec, nullptr);
break;
}
@@ -1058,7 +1147,7 @@ void ClientManagerImpl::forwardMessageToDeadLetterQueue(const std::string& targe
ec = ErrorCode::ServiceUnavailable;
}
case rmq::Code::TOO_MANY_REQUESTS: {
- ec = ErrorCode::TooManyRequest;
+ ec = ErrorCode::TooManyRequests;
}
default: {
ec = ErrorCode::NotImplemented;
diff --git a/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp b/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp
index 4667de4..3ee8f68 100644
--- a/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp
+++ b/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp
@@ -73,7 +73,7 @@ void ReceiveMessageStreamReader::OnReadDone(bool ok) {
break;
}
case rmq::Code::TOO_MANY_REQUESTS: {
- ec_ = ErrorCode::TooManyRequest;
+ ec_ = ErrorCode::TooManyRequests;
break;
}
case rmq::Code::MESSAGE_NOT_FOUND: {
[rocketmq-clients] 12/18: Add dependent CDN through OSS
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit 390a87ed917daaeaa400136d380f030485c38caf
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Fri Jul 1 13:32:24 2022 +0800
Add dependent CDN through OSS
---
cpp/bazel/rocketmq_deps.bzl | 13 +++++++++++--
1 file changed, 11 insertions(+), 2 deletions(-)
diff --git a/cpp/bazel/rocketmq_deps.bzl b/cpp/bazel/rocketmq_deps.bzl
index 39f3642..e8196ea 100644
--- a/cpp/bazel/rocketmq_deps.bzl
+++ b/cpp/bazel/rocketmq_deps.bzl
@@ -26,7 +26,8 @@ def rocketmq_deps():
sha256 = "b4870bf121ff7795ba20d20bcdd8627b8e088f2d1dab299a031c1034eddc93d5",
strip_prefix = "googletest-release-1.11.0",
urls = [
- "https://github.com/google/googletest/archive/refs/tags/release-1.11.0.tar.gz",
+ "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/googletest/googletest-release-1.11.0.tar.gz",
+ "https://github.com/google/googletest/archive/refs/tags/release-1.11.0.tar.gz",
],
)
@@ -36,6 +37,7 @@ def rocketmq_deps():
strip_prefix = "filesystem-1.5.0",
sha256 = "eb6f3b0739908ad839cde68885d70e7324db191b9fad63d9915beaa40444d9cb",
urls = [
+ "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/filesystem/filesystem-1.5.0.tar.gz",
"https://github.com/gulrak/filesystem/archive/v1.5.0.tar.gz",
],
build_file = "@org_apache_rocketmq//third_party:filesystem.BUILD",
@@ -47,6 +49,7 @@ def rocketmq_deps():
strip_prefix = "spdlog-1.9.2",
sha256 = "6fff9215f5cb81760be4cc16d033526d1080427d236e86d70bb02994f85e3d38",
urls = [
+ "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/spdlog/spdlog-1.9.2.tar.gz",
"https://github.com/gabime/spdlog/archive/refs/tags/v1.9.2.tar.gz",
],
build_file = "@org_apache_rocketmq//third_party:spdlog.BUILD",
@@ -58,6 +61,7 @@ def rocketmq_deps():
strip_prefix = "fmt-8.0.1",
sha256 = "b06ca3130158c625848f3fb7418f235155a4d389b2abc3a6245fb01cb0eb1e01",
urls = [
+ "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/fmt/fmt-8.0.1.tar.gz",
"https://github.com/fmtlib/fmt/archive/refs/tags/8.0.1.tar.gz",
],
build_file = "@org_apache_rocketmq//third_party:fmtlib.BUILD",
@@ -125,6 +129,7 @@ def rocketmq_deps():
build_file = "@org_apache_rocketmq//third_party:asio.BUILD",
strip_prefix = "asio-1.18.2",
urls = [
+ "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/asio/asio-1.18.2.tar.gz",
"https://github.com/lizhanhui/asio/archive/refs/tags/v1.18.2.tar.gz",
],
)
@@ -135,7 +140,10 @@ def rocketmq_deps():
sha256 = "0ff62e28eb0f6e563178d44b77c94dddb8702141d83dd34b83cb046399c2b1d5",
build_file = "@org_apache_rocketmq//third_party:cpp_httplib.BUILD",
strip_prefix = "cpp-httplib-0.9.4",
- urls = ["https://github.com/yhirose/cpp-httplib/archive/refs/tags/v0.9.4.tar.gz"],
+ urls = [
+ "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/cpp-httplib/cpp-httplib-0.9.4.tar.gz",
+ "https://github.com/yhirose/cpp-httplib/archive/refs/tags/v0.9.4.tar.gz",
+ ],
)
maybe(
@@ -143,6 +151,7 @@ def rocketmq_deps():
name = "com_google_googleapis",
sha256 = "e89f15d54b0ddab0cd41d18cb2299e5447db704e2b05ff141cb1769170671466",
urls = [
+ "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/googleapis/googleapis-af7fb72df59a814221b123a4d1acb3f6c3e6cc95.zip",
"https://github.com/googleapis/googleapis/archive/af7fb72df59a814221b123a4d1acb3f6c3e6cc95.zip"
],
strip_prefix = "googleapis-af7fb72df59a814221b123a4d1acb3f6c3e6cc95",
[rocketmq-clients] 17/18: Upgrade grpc to v1.46.3.2
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit 056e68c037786dc495fbdf5b65e4c563de7980d8
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Fri Jul 1 14:30:45 2022 +0800
Upgrade grpc to v1.46.3.2
---
cpp/bazel/rocketmq_deps.bzl | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/cpp/bazel/rocketmq_deps.bzl b/cpp/bazel/rocketmq_deps.bzl
index b13bc27..30ee383 100644
--- a/cpp/bazel/rocketmq_deps.bzl
+++ b/cpp/bazel/rocketmq_deps.bzl
@@ -114,11 +114,11 @@ def rocketmq_deps():
maybe(
http_archive,
name = "com_github_grpc_grpc",
- strip_prefix = "grpc-1.46.3.1",
+ strip_prefix = "grpc-1.46.3.2",
# sha256 = "67423a4cd706ce16a88d1549297023f0f9f0d695a96dd684adc21e67b021f9bc",
urls = [
- "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/grpc/grpc-1.46.3.1.tar.gz",
- "https://github.com/lizhanhui/grpc/archive/refs/tags/v1.46.3.1.tar.gz",
+ "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/grpc/grpc-1.46.3.2.tar.gz",
+ "https://github.com/lizhanhui/grpc/archive/refs/tags/v1.46.3.2.tar.gz",
],
)
[rocketmq-clients] 08/18: Add error handling for EndTransaction
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit f40067beeffe26fcbd94b470b06d1f20b0ee0e94
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Thu Jun 30 19:15:22 2022 +0800
Add error handling for EndTransaction
---
cpp/api/rocketmq/ErrorCode.h | 16 ++--
cpp/src/main/cpp/client/ClientManagerImpl.cpp | 113 +++++++++++++++++++++++---
2 files changed, 112 insertions(+), 17 deletions(-)
diff --git a/cpp/api/rocketmq/ErrorCode.h b/cpp/api/rocketmq/ErrorCode.h
index e911533..4f05ded 100644
--- a/cpp/api/rocketmq/ErrorCode.h
+++ b/cpp/api/rocketmq/ErrorCode.h
@@ -31,29 +31,35 @@ enum class ErrorCode : int {
* @brief Client state not as expected. Call Producer#start() first.
*
*/
- IllegalState = 1,
+ IllegalState = 10000,
/**
* @brief To publish FIFO messages, only synchronous API is supported.
*/
- BadRequestAsyncPubFifoMessage = 10001,
+ BadRequestAsyncPubFifoMessage = 10100,
+
+ /**
+ * @brief 102XX is used for client side error.
+ *
+ */
+ InternalClientError = 10200,
/**
* @brief Broker has processed the request but is not going to return any content.
*/
- NoContent = 204,
+ NoContent = 20400,
/**
* @brief Bad configuration. For example, negative max-attempt-times.
*
*/
- BadConfiguration = 300,
+ BadConfiguration = 30000,
/**
* @brief Generic code, representing multiple return results.
*
*/
- MultipleResults = 30000,
+ MultipleResults = 30100,
/**
* @brief The server cannot process the request due to apparent client-side
diff --git a/cpp/src/main/cpp/client/ClientManagerImpl.cpp b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
index 3f572b3..1602769 100644
--- a/cpp/src/main/cpp/client/ClientManagerImpl.cpp
+++ b/cpp/src/main/cpp/client/ClientManagerImpl.cpp
@@ -1188,27 +1188,72 @@ void ClientManagerImpl::endTransaction(
}
auto&& status = invocation_context->response.status();
+ auto&& peer_address = invocation_context->remote_address;
switch (status.code()) {
case rmq::Code::OK: {
SPDLOG_DEBUG("endTransaction completed OK. Response: {}, host={}", invocation_context->response.DebugString(),
- invocation_context->remote_address);
- } break;
+ peer_address);
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_TOPIC: {
+ SPDLOG_WARN("IllegalTopic: {}, host={}", status.message(), peer_address);
+ ec = ErrorCode::IllegalTopic;
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_CONSUMER_GROUP: {
+ SPDLOG_WARN("IllegalConsumerGroup: {}, host={}", status.message(), peer_address);
+ ec = ErrorCode::IllegalConsumerGroup;
+ break;
+ }
+
+ case rmq::Code::INVALID_TRANSACTION_ID: {
+ SPDLOG_WARN("InvalidTransactionId: {}, host={}", status.message(), peer_address);
+ ec = ErrorCode::InvalidTransactionId;
+ break;
+ }
+
+ case rmq::Code::CLIENT_ID_REQUIRED: {
+ SPDLOG_WARN("ClientIdRequired: {}, host={}", status.message(), peer_address);
+ ec = ErrorCode::ClientIdRequired;
+ break;
+ }
+
+ case rmq::Code::TOPIC_NOT_FOUND: {
+ SPDLOG_WARN("TopicNotFound: {}, host={}", status.message(), peer_address);
+ ec = ErrorCode::TopicNotFound;
+ break;
+ }
+
case rmq::Code::UNAUTHORIZED: {
- SPDLOG_WARN("Unauthorized: {}, host={}", status.message(), invocation_context->remote_address);
+ SPDLOG_WARN("Unauthorized: {}, host={}", status.message(), peer_address);
ec = ErrorCode::Unauthorized;
- } break;
+ break;
+ }
+
case rmq::Code::FORBIDDEN: {
- SPDLOG_WARN("Forbidden: {}, host={}", status.message(), invocation_context->remote_address);
+ SPDLOG_WARN("Forbidden: {}, host={}", status.message(), peer_address);
ec = ErrorCode::Forbidden;
- } break;
+ break;
+ }
+
case rmq::Code::INTERNAL_SERVER_ERROR: {
- SPDLOG_WARN("InternalServerError: {}, host={}", status.message(), invocation_context->remote_address);
+ SPDLOG_WARN("InternalServerError: {}, host={}", status.message(), peer_address);
ec = ErrorCode::InternalServerError;
- } break;
+ break;
+ }
+
+ case rmq::Code::PROXY_TIMEOUT: {
+ SPDLOG_WARN("GatewayTimeout: {}, host={}", status.message(), peer_address);
+ ec = ErrorCode::GatewayTimeout;
+ break;
+ }
+
default: {
- SPDLOG_WARN("NotImplemented: please upgrade SDK to latest release. {}, host={}", status.message(),
- invocation_context->remote_address);
- ec = ErrorCode::NotImplemented;
+ SPDLOG_WARN("NotSupported: please upgrade SDK to latest release. {}, host={}", status.message(), peer_address);
+ ec = ErrorCode::NotSupported;
+ break;
}
}
cb(ec, invocation_context->response);
@@ -1246,22 +1291,66 @@ void ClientManagerImpl::forwardMessageToDeadLetterQueue(const std::string& targe
SPDLOG_DEBUG("Received forwardToDeadLetterQueue response from server[host={}]", invocation_context->remote_address);
std::error_code ec;
- switch (invocation_context->response.status().code()) {
+ auto&& status = invocation_context->response.status();
+ auto&& peer_address = invocation_context->remote_address;
+ switch (status.code()) {
case rmq::Code::OK: {
break;
}
+ case rmq::Code::ILLEGAL_TOPIC: {
+ SPDLOG_WARN("IllegalTopic: {}. Host={}", status.message(), peer_address);
+ ec = ErrorCode::IllegalTopic;
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_CONSUMER_GROUP: {
+ SPDLOG_WARN("IllegalConsumerGroup: {}. Host={}", status.message(), peer_address);
+ ec = ErrorCode::IllegalConsumerGroup;
+ break;
+ }
+
+ case rmq::Code::INVALID_RECEIPT_HANDLE: {
+ SPDLOG_WARN("IllegalReceiptHandle: {}. Host={}", status.message(), peer_address);
+ ec = ErrorCode::InvalidReceiptHandle;
+ break;
+ }
+
+ case rmq::Code::CLIENT_ID_REQUIRED: {
+ SPDLOG_WARN("IllegalTopic: {}. Host={}", status.message(), peer_address);
+
+ // TODO: translate to client internal error?
+ ec = ErrorCode::InternalServerError;
+ break;
+ }
+
+ case rmq::Code::TOPIC_NOT_FOUND: {
+ ec = ErrorCode::TopicNotFound;
+ break;
+ }
+
case rmq::Code::INTERNAL_SERVER_ERROR: {
ec = ErrorCode::ServiceUnavailable;
+ break;
}
+
case rmq::Code::TOO_MANY_REQUESTS: {
ec = ErrorCode::TooManyRequests;
+ break;
+ }
+
+ case rmq::Code::PROXY_TIMEOUT: {
+ ec = ErrorCode::GatewayTimeout;
+ break;
}
+
default: {
ec = ErrorCode::NotImplemented;
+ break;
}
}
cb(ec);
};
+
invocation_context->callback = callback;
client->asyncForwardMessageToDeadLetterQueue(request, invocation_context);
}
[rocketmq-clients] 16/18: Upgrade grpc to v1.46.3.1
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit c5bbe3426e12b538c548a7c620fc3227117fb52b
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Fri Jul 1 14:24:43 2022 +0800
Upgrade grpc to v1.46.3.1
---
cpp/bazel/rocketmq_deps.bzl | 20 ++++++++++----------
1 file changed, 10 insertions(+), 10 deletions(-)
diff --git a/cpp/bazel/rocketmq_deps.bzl b/cpp/bazel/rocketmq_deps.bzl
index 3ecd0cb..b13bc27 100644
--- a/cpp/bazel/rocketmq_deps.bzl
+++ b/cpp/bazel/rocketmq_deps.bzl
@@ -111,16 +111,16 @@ def rocketmq_deps():
],
)
- if "com_github_grpc_grpc" not in native.existing_rules():
- http_archive(
- name = "com_github_grpc_grpc",
- strip_prefix = "grpc-1.46.0",
- sha256 = "67423a4cd706ce16a88d1549297023f0f9f0d695a96dd684adc21e67b021f9bc",
- urls = [
- "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/grpc/grpc-1.46.0.tar.gz",
- "https://github.com/grpc/grpc/archive/refs/tags/v1.46.0.tar.gz",
- ],
- )
+ maybe(
+ http_archive,
+ name = "com_github_grpc_grpc",
+ strip_prefix = "grpc-1.46.3.1",
+ # sha256 = "67423a4cd706ce16a88d1549297023f0f9f0d695a96dd684adc21e67b021f9bc",
+ urls = [
+ "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/grpc/grpc-1.46.3.1.tar.gz",
+ "https://github.com/lizhanhui/grpc/archive/refs/tags/v1.46.3.1.tar.gz",
+ ],
+ )
maybe(
http_archive,
[rocketmq-clients] 06/18: Add error handling for ReceiveMessage
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit b4a2e50fa67be0eb9777232d4b668f507608de99
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Thu Jun 30 17:21:02 2022 +0800
Add error handling for ReceiveMessage
---
.../main/cpp/client/ReceiveMessageStreamReader.cpp | 50 +++++++++++++++++++++-
1 file changed, 49 insertions(+), 1 deletion(-)
diff --git a/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp b/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp
index 3ee8f68..0684f51 100644
--- a/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp
+++ b/cpp/src/main/cpp/client/ReceiveMessageStreamReader.cpp
@@ -17,7 +17,10 @@
#include "ReceiveMessageStreamReader.h"
+#include <apache/rocketmq/v2/definition.pb.h>
+
#include "LoggerImpl.h"
+#include "rocketmq/ErrorCode.h"
ROCKETMQ_NAMESPACE_BEGIN
@@ -63,6 +66,26 @@ void ReceiveMessageStreamReader::OnReadDone(bool ok) {
case rmq::Code::OK: {
break;
}
+ case rmq::Code::ILLEGAL_TOPIC: {
+ ec_ = ErrorCode::IllegalTopic;
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_CONSUMER_GROUP: {
+ ec_ = ErrorCode::IllegalConsumerGroup;
+ break;
+ }
+
+ case rmq::Code::ILLEGAL_FILTER_EXPRESSION: {
+ ec_ = ErrorCode::IllegalFilterExpression;
+ break;
+ }
+
+ case rmq::Code::CLIENT_ID_REQUIRED: {
+ ec_ = ErrorCode::ClientIdRequired;
+ break;
+ }
+
case rmq::Code::TOPIC_NOT_FOUND: {
ec_ = ErrorCode::TopicNotFound;
break;
@@ -72,17 +95,42 @@ void ReceiveMessageStreamReader::OnReadDone(bool ok) {
ec_ = ErrorCode::ConsumerGroupNotFound;
break;
}
+
case rmq::Code::TOO_MANY_REQUESTS: {
ec_ = ErrorCode::TooManyRequests;
break;
}
+
case rmq::Code::MESSAGE_NOT_FOUND: {
ec_ = ErrorCode::NoContent;
break;
}
- default:
+
+ case rmq::Code::UNAUTHORIZED: {
+ ec_ = ErrorCode::Unauthorized;
+ break;
+ }
+
+ case rmq::Code::FORBIDDEN: {
+ ec_ = ErrorCode::Forbidden;
+ break;
+ }
+
+ case rmq::Code::INTERNAL_SERVER_ERROR: {
+ ec_ = ErrorCode::InternalServerError;
+ break;
+ }
+
+ case rmq::Code::PROXY_TIMEOUT: {
+ ec_ = ErrorCode::GatewayTimeout;
+ break;
+ }
+
+ default: {
+ ec_ = ErrorCode::NotSupported;
SPDLOG_WARN("Unsupported code={}", response_.status().code());
break;
+ }
}
break;
}
[rocketmq-clients] 13/18: Add io_bazel_rules_go mirror
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit a5aa9049266fbbfb4d1863afcac463a9a91ba84f
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Fri Jul 1 13:42:07 2022 +0800
Add io_bazel_rules_go mirror
---
cpp/bazel/rocketmq_deps.bzl | 11 +++++++++++
1 file changed, 11 insertions(+)
diff --git a/cpp/bazel/rocketmq_deps.bzl b/cpp/bazel/rocketmq_deps.bzl
index e8196ea..b92d20d 100644
--- a/cpp/bazel/rocketmq_deps.bzl
+++ b/cpp/bazel/rocketmq_deps.bzl
@@ -176,4 +176,15 @@ def rocketmq_deps():
"https://github.com/bazelbuild/rules_swift/archive/refs/tags/0.27.0.tar.gz",
],
strip_prefix = "rules_swift-0.27.0",
+ )
+
+ maybe(
+ http_archive,
+ name = "io_bazel_rules_go",
+ sha256 = "685052b498b6ddfe562ca7a97736741d87916fe536623afb7da2824c0211c369",
+ urls = [
+ "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/rules-go/rules_go-v0.33.0.zip",
+ "https://mirror.bazel.build/github.com/bazelbuild/rules_go/releases/download/v0.33.0/rules_go-v0.33.0.zip",
+ "https://github.com/bazelbuild/rules_go/releases/download/v0.33.0/rules_go-v0.33.0.zip",
+ ],
)
\ No newline at end of file
[rocketmq-clients] 18/18: Use official gRPC release v1.46.3
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit 6b31ad5c3771ff9133ecbf432f6bf5c54cc02f8d
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Fri Jul 1 15:15:20 2022 +0800
Use official gRPC release v1.46.3
---
cpp/bazel/rocketmq_deps.bzl | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/cpp/bazel/rocketmq_deps.bzl b/cpp/bazel/rocketmq_deps.bzl
index 30ee383..b3d3168 100644
--- a/cpp/bazel/rocketmq_deps.bzl
+++ b/cpp/bazel/rocketmq_deps.bzl
@@ -114,11 +114,11 @@ def rocketmq_deps():
maybe(
http_archive,
name = "com_github_grpc_grpc",
- strip_prefix = "grpc-1.46.3.2",
+ strip_prefix = "grpc-1.46.3",
# sha256 = "67423a4cd706ce16a88d1549297023f0f9f0d695a96dd684adc21e67b021f9bc",
urls = [
- "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/grpc/grpc-1.46.3.2.tar.gz",
- "https://github.com/lizhanhui/grpc/archive/refs/tags/v1.46.3.2.tar.gz",
+ "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/grpc/grpc-1.46.3.tar.gz",
+ "https://github.com/grpc/grpc/archive/refs/tags/v1.46.3.tar.gz",
],
)
[rocketmq-clients] 15/18: Fix url of bazel_proto mirror
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit 747d699e9a167609f6642483adb9bea0087b578c
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Fri Jul 1 14:08:13 2022 +0800
Fix url of bazel_proto mirror
---
cpp/bazel/rocketmq_deps.bzl | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/cpp/bazel/rocketmq_deps.bzl b/cpp/bazel/rocketmq_deps.bzl
index 106b7a9..3ecd0cb 100644
--- a/cpp/bazel/rocketmq_deps.bzl
+++ b/cpp/bazel/rocketmq_deps.bzl
@@ -195,7 +195,7 @@ def rocketmq_deps():
sha256 = "e017528fd1c91c5a33f15493e3a398181a9e821a804eb7ff5acdd1d2d6c2b18d",
strip_prefix = "rules_proto-4.0.0-3.20.0",
urls = [
- "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/rules_proto/rules_proto-4.0.0.tar.gz",
+ "https://shutian.oss-cn-hangzhou.aliyuncs.com/cdn/rules_proto/rules_proto-4.0.0-3.20.0.tar.gz",
"https://github.com/bazelbuild/rules_proto/archive/refs/tags/4.0.0-3.20.0.tar.gz",
],
)
\ No newline at end of file