diff --git a/be/src/cloud/cloud_meta_mgr.cpp b/be/src/cloud/cloud_meta_mgr.cpp index ea9c7b376f332e..c3b4dd87bbe68b 100644 --- a/be/src/cloud/cloud_meta_mgr.cpp +++ b/be/src/cloud/cloud_meta_mgr.cpp @@ -153,11 +153,56 @@ Status bthread_fork_join(std::vector>&& tasks, int concu return Status::OK(); } +// Resolve the status code returned by Meta Service (MS) for BE/FE clients of different version. +// Assuming MS is always the latest version, it sends both the meta-service error code and a code that +// older clients can decode: +// +// latest MS +// +---------------------------------------+ +// | actual_code = meta-service error code | +// | code = compatible code | +// +----------------+----------------------+ +// | +// +-------------+-------------+ +// | | +// old BE/FE without old BE/FE with +// the actual_code field the actual_code field +// | | +// ignores actual_code local enum recognizes +// and reads code actual_code value? +// / \ +// yes no +// | | +// use actual code use code only when it +// is explicit and non-OK +// | +// otherwise return +// UNDEFINED_ERR +// +// After MS adds an error code, an older actual_code-aware client may not have that enum value; +// MetaServiceCode_IsValid detects this case. The non-OK fallback check is essential: +// if MS ignore or incorrectly converts the compatible code to OK, an unknown error +// must remain an error instead of becoming a false success. MetaServiceCode get_response_code(const MetaServiceResponseStatus& status) { - if (status.has_actual_code() && MetaServiceCode_IsValid(status.actual_code())) { - return static_cast(status.actual_code()); + if (status.has_actual_code()) { + // Check whether this client build contains the code in its MetaServiceCode enum. + if (MetaServiceCode_IsValid(status.actual_code())) { + return static_cast(status.actual_code()); + } + // An older client may use the compatible code, but unsupported cases must return an explicit error. + // Return the non-OK compatible code prepared by MS for older clients. + if (status.has_code() && status.code() != MetaServiceCode::OK) { + return status.code(); + } + // Never return OK when the compatible code is absent or invalid. + return MetaServiceCode::UNDEFINED_ERR; + } + // A legacy response has only code, so return its explicit value, including a real OK. + if (status.has_code()) { + return status.code(); } - return status.code(); + // A response missing both fields is invalid and must be rejected. + return MetaServiceCode::UNDEFINED_ERR; } namespace { diff --git a/be/src/cloud/cloud_meta_mgr.h b/be/src/cloud/cloud_meta_mgr.h index 9084df3d6ea42a..2108c9ae194001 100644 --- a/be/src/cloud/cloud_meta_mgr.h +++ b/be/src/cloud/cloud_meta_mgr.h @@ -64,7 +64,8 @@ Status bthread_fork_join(const std::vector>& tasks, int Status bthread_fork_join(std::vector>&& tasks, int concurrency, std::future* fut); -// Returns the exact actual_code when recognized, otherwise the legacy-compatible code. +// Returns the exact actual_code when recognized. An unknown actual_code uses an explicit non-OK +// legacy fallback and otherwise fails closed. Responses from a legacy Meta Service use code. // Exposed for unit tests. MetaServiceCode get_response_code(const MetaServiceResponseStatus& status); diff --git a/be/test/cloud/cloud_meta_mgr_test.cpp b/be/test/cloud/cloud_meta_mgr_test.cpp index ff87378348ef10..9b10f80a218660 100644 --- a/be/test/cloud/cloud_meta_mgr_test.cpp +++ b/be/test/cloud/cloud_meta_mgr_test.cpp @@ -56,17 +56,47 @@ TEST_F(CloudMetaMgrTest, response_status_uses_actual_code_when_valid) { status.set_actual_code(static_cast(MetaServiceCode::KV_TXN_CONFLICT)); EXPECT_EQ(get_response_code(status), MetaServiceCode::KV_TXN_CONFLICT); + status.set_code(MetaServiceCode::KV_TXN_CONFLICT); + status.set_actual_code(static_cast(MetaServiceCode::OK)); + EXPECT_EQ(get_response_code(status), MetaServiceCode::OK); + + status.clear_code(); + status.set_actual_code(static_cast(MetaServiceCode::MS_TOO_BUSY)); + EXPECT_EQ(get_response_code(status), MetaServiceCode::MS_TOO_BUSY); + + status.set_code(MetaServiceCode::KV_TXN_CONFLICT); status.clear_actual_code(); EXPECT_EQ(get_response_code(status), MetaServiceCode::KV_TXN_CONFLICT); } -TEST_F(CloudMetaMgrTest, response_status_falls_back_for_invalid_actual_code) { +TEST_F(CloudMetaMgrTest, response_status_falls_back_to_non_ok_code_for_invalid_actual_code) { MetaServiceResponseStatus status; status.set_code(MetaServiceCode::KV_TXN_CONFLICT); status.set_actual_code(std::numeric_limits::max()); EXPECT_EQ(get_response_code(status), MetaServiceCode::KV_TXN_CONFLICT); } +TEST_F(CloudMetaMgrTest, response_status_returns_undefined_for_invalid_actual_code_with_ok) { + MetaServiceResponseStatus status; + status.set_code(MetaServiceCode::OK); + status.set_actual_code(std::numeric_limits::max()); + EXPECT_EQ(get_response_code(status), MetaServiceCode::UNDEFINED_ERR); +} + +TEST_F(CloudMetaMgrTest, response_status_returns_undefined_for_invalid_actual_code_without_code) { + MetaServiceResponseStatus status; + status.set_actual_code(std::numeric_limits::max()); + EXPECT_EQ(get_response_code(status), MetaServiceCode::UNDEFINED_ERR); +} + +TEST_F(CloudMetaMgrTest, response_status_returns_undefined_without_any_code) { + MetaServiceResponseStatus status; + EXPECT_EQ(get_response_code(status), MetaServiceCode::UNDEFINED_ERR); + + status.set_code(MetaServiceCode::OK); + EXPECT_EQ(get_response_code(status), MetaServiceCode::OK); +} + static AbortTxnRequest get_abort_txn_request(CloudMetaMgr* meta_mgr, const StreamLoadContext& ctx) { auto* sp = SyncPoint::get_instance(); sp->clear_all_call_backs(); diff --git a/cloud/src/meta-service/meta_service_helper.h b/cloud/src/meta-service/meta_service_helper.h index 9b1fbed4648436..4a33917e82bdb0 100644 --- a/cloud/src/meta-service/meta_service_helper.h +++ b/cloud/src/meta-service/meta_service_helper.h @@ -18,6 +18,7 @@ #pragma once #include +#include #include #include @@ -42,22 +43,30 @@ #include "resource-manager/resource_manager.h" namespace doris::cloud { -inline MetaServiceCode get_legacy_code(MetaServiceCode code) { +// Converts a response code and message to values that older clients can read. +// set_response_code() stores the original code in actual_code +// Call this function only from set_response_code() or from unit tests; do not call it from other production code. +// When adding an error code that may be returned to clients, must add its conversion here. +inline std::pair resolve_response_code_and_msg(MetaServiceCode code, + std::string msg) { switch (code) { - // MS_TOO_BUSY is a overload signal. Map it to KV_TXN_CONFLICT so the BE's existing + // MS_TOO_BUSY is an overload signal. Map it to KV_TXN_CONFLICT so the BE's existing // conflict-retry path can retry the request. case MetaServiceCode::MS_TOO_BUSY: - return MetaServiceCode::KV_TXN_CONFLICT; + msg += std::string((msg.empty() ? "" : ", ")) + + "[MS_TOO_BUSY will be converted to code=KV_TXN_CONFLICT for old version clients]"; + return {MetaServiceCode::KV_TXN_CONFLICT, std::move(msg)}; default: - return code; + return {code, std::move(msg)}; } } inline void set_response_code(MetaServiceResponseStatus* status, MetaServiceCode code, std::string msg) { + auto [resolved_code, resolved_msg] = resolve_response_code_and_msg(code, std::move(msg)); status->set_actual_code(static_cast(code)); - status->set_code(get_legacy_code(code)); - status->set_msg(std::move(msg)); + status->set_code(resolved_code); + status->set_msg(std::move(resolved_msg)); } inline std::string md5(const std::string& str) { diff --git a/cloud/test/meta_service_helper_test.cpp b/cloud/test/meta_service_helper_test.cpp index 7b50792f88b631..10318e3e8d102b 100644 --- a/cloud/test/meta_service_helper_test.cpp +++ b/cloud/test/meta_service_helper_test.cpp @@ -228,7 +228,9 @@ TEST_F(MetaServiceWireCompatibilityTest, LegacyClientReadsFallbackAndIgnoresActu ASSERT_TRUE(reflection->HasField(*legacy_status, legacy_code_field_)); EXPECT_EQ(reflection->GetEnumValue(*legacy_status, legacy_code_field_), MetaServiceCode::KV_TXN_CONFLICT); - EXPECT_EQ(reflection->GetString(*legacy_status, legacy_msg_field_), "busy"); + EXPECT_EQ(reflection->GetString(*legacy_status, legacy_msg_field_), + "busy, [MS_TOO_BUSY will be converted to code=KV_TXN_CONFLICT for old version " + "clients]"); EXPECT_EQ(legacy_status_descriptor_->FindFieldByName("actual_code"), nullptr); const auto& unknown_fields = reflection->GetUnknownFields(*legacy_status); @@ -287,12 +289,18 @@ TEST(MetaServiceHelperTest, ResponseStatusUsesExactAndLegacyCodes) { set_response_code(&status, MetaServiceCode::MS_TOO_BUSY, "busy"); EXPECT_EQ(status.code(), MetaServiceCode::KV_TXN_CONFLICT); EXPECT_EQ(status.actual_code(), MetaServiceCode::MS_TOO_BUSY); - EXPECT_EQ(status.msg(), "busy"); + EXPECT_EQ(status.msg(), + "busy, [MS_TOO_BUSY will be converted to code=KV_TXN_CONFLICT for old version " + "clients]"); set_response_code(&status, MetaServiceCode::KV_TXN_CONFLICT, "conflict"); EXPECT_EQ(status.code(), MetaServiceCode::KV_TXN_CONFLICT); EXPECT_EQ(status.actual_code(), MetaServiceCode::KV_TXN_CONFLICT); EXPECT_EQ(status.msg(), "conflict"); + + set_response_code(&status, MetaServiceCode::MS_TOO_BUSY, ""); + EXPECT_EQ(status.msg(), + "[MS_TOO_BUSY will be converted to code=KV_TXN_CONFLICT for old version clients]"); } TEST(MetaServiceHelperTest, ResponseStatusCoversEveryMetaServiceCode) { @@ -395,7 +403,9 @@ TEST(MetaServiceHelperTest, ResponseStatusCoversEveryMetaServiceCode) { expect_response_status(MetaServiceCode::UNDEFINED_ERR, MetaServiceCode::UNDEFINED_ERR); EXPECT_EQ(covered_codes.size(), - static_cast(MetaServiceCode_descriptor()->value_count())); + static_cast(MetaServiceCode_descriptor()->value_count())) + << "A new MetaServiceCode was added. Add it to this test and handle it in " + "resolve_response_code_and_msg()."; } } // namespace doris::cloud diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceClient.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceClient.java index 7fbb70900f0f07..3a0f887c05c884 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceClient.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceClient.java @@ -111,17 +111,15 @@ public void onMessage(RespT response) { } @SuppressWarnings("unchecked") - // Restore the exact status code from actual_code when this FE recognizes it. - // Otherwise, keep - // the legacy-compatible value in code so responses from a newer Meta Service - // remain readable. + // Restore the exact status code from actual_code when this FE recognizes it. For an unknown + // actual_code, only an explicit non-OK legacy code is a safe fallback. private static Response restoreActualCode(Response response) { if (!(response instanceof Message)) { return response; } Message message = (Message) response; Descriptors.FieldDescriptor statusField = message.getDescriptorForType().findFieldByName("status"); - if (statusField == null || !message.hasField(statusField)) { + if (statusField == null) { return response; } Object statusObject = message.getField(statusField); @@ -130,11 +128,22 @@ private static Response restoreActualCode(Response response) { } Cloud.MetaServiceResponseStatus status = (Cloud.MetaServiceResponseStatus) statusObject; - if (!status.hasActualCode()) { - return response; + Cloud.MetaServiceCode code; + if (status.hasActualCode()) { + code = Cloud.MetaServiceCode.forNumber(status.getActualCode()); + if (code == null) { + if (status.hasCode() && status.getCode() != Cloud.MetaServiceCode.OK) { + return response; + } + code = Cloud.MetaServiceCode.UNDEFINED_ERR; + } + } else { + if (status.hasCode()) { + return response; + } + code = Cloud.MetaServiceCode.UNDEFINED_ERR; } - Cloud.MetaServiceCode code = Cloud.MetaServiceCode.forNumber(status.getActualCode()); - if (code == null || code == status.getCode()) { + if (status.hasCode() && code == status.getCode()) { return response; } Cloud.MetaServiceResponseStatus restoredStatus = status.toBuilder().setCode(code).build(); diff --git a/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceProxyTest.java b/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceProxyTest.java index 31e0887f02e24c..a9f25586bee50c 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceProxyTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceProxyTest.java @@ -229,6 +229,16 @@ public void testGetInstancePrefersKnownActualCode() throws RpcException { Assert.assertEquals(Cloud.MetaServiceCode.MS_TOO_BUSY.getNumber(), status.getActualCode()); } + @Test + public void testGetInstanceUsesKnownActualCodeWithoutFallback() throws RpcException { + Cloud.MetaServiceResponseStatus status = callGetInstanceWithStatus( + Cloud.MetaServiceResponseStatus.newBuilder() + .setActualCode(Cloud.MetaServiceCode.MS_TOO_BUSY.getNumber()) + .build()); + + Assert.assertEquals(Cloud.MetaServiceCode.MS_TOO_BUSY, status.getCode()); + } + @Test public void testGetInstanceKeepsLegacyCodeForUnknownActualCode() throws RpcException { Cloud.MetaServiceResponseStatus status = callGetInstanceWithStatus( @@ -241,6 +251,44 @@ public void testGetInstanceKeepsLegacyCodeForUnknownActualCode() throws RpcExcep Assert.assertEquals(Integer.MAX_VALUE, status.getActualCode()); } + @Test + public void testGetInstanceFailsClosedForUnknownActualCodeWithoutErrorFallback() throws RpcException { + Cloud.MetaServiceResponseStatus status = callGetInstanceWithStatus( + Cloud.MetaServiceResponseStatus.newBuilder() + .setCode(Cloud.MetaServiceCode.OK) + .setActualCode(Integer.MAX_VALUE) + .build()); + + Assert.assertEquals(Cloud.MetaServiceCode.UNDEFINED_ERR, status.getCode()); + Assert.assertEquals(Integer.MAX_VALUE, status.getActualCode()); + + status = callGetInstanceWithStatus(Cloud.MetaServiceResponseStatus.newBuilder() + .setActualCode(Integer.MAX_VALUE) + .build()); + + Assert.assertEquals(Cloud.MetaServiceCode.UNDEFINED_ERR, status.getCode()); + Assert.assertEquals(Integer.MAX_VALUE, status.getActualCode()); + } + + @Test + public void testGetInstanceFailsClosedWithoutAnyCode() throws RpcException { + Cloud.MetaServiceResponseStatus status = callGetInstanceWithStatus( + Cloud.MetaServiceResponseStatus.getDefaultInstance()); + + Assert.assertEquals(Cloud.MetaServiceCode.UNDEFINED_ERR, status.getCode()); + } + + @Test + public void testResponseFailsClosedWithoutStatus() { + Cloud.GetInstanceResponse response = Deencapsulation.invoke( + MetaServiceClient.class, + "restoreActualCode", + Cloud.GetInstanceResponse.getDefaultInstance()); + + Assert.assertTrue(response.hasStatus()); + Assert.assertEquals(Cloud.MetaServiceCode.UNDEFINED_ERR, response.getStatus().getCode()); + } + @Test public void testGetInstanceKeepsLegacyCodeWithoutActualCode() throws RpcException { Cloud.MetaServiceResponseStatus status = callGetInstanceWithStatus( diff --git a/gensrc/proto/cloud.proto b/gensrc/proto/cloud.proto index 51c522b6272cee..c58f81d89fb1bc 100644 --- a/gensrc/proto/cloud.proto +++ b/gensrc/proto/cloud.proto @@ -1467,8 +1467,9 @@ message MetaServiceResponseStatus { optional string msg = 2; // Exact client-visible status code encoded as int32, so proto2 clients do not drop unknown // enum values. Internal retry signals must be converted before the response is sent. - // New clients should use this field when the local enum descriptor recognizes the value, - // otherwise fall back to `code`. + // New clients should use this field when the local enum descriptor recognizes the value. + // Otherwise, use `code` only when it is explicitly present and non-OK, and fail closed with + // UNDEFINED_ERR when no recognizable error code is available. optional int32 actual_code = 3; } @@ -1812,7 +1813,7 @@ enum MetaServiceCode { // MetaService must write the exact client-visible code to // `MetaServiceResponseStatus.actual_code` and write only a legacy fallback code to // `MetaServiceResponseStatus.code`. Any newly added error code that may be returned to - // clients must be mapped in get_legacy_code(). + // clients must be mapped in resolve_response_code_and_msg(). OK = 0; //Meta service internal error