[fix](cloud) Fail closed on unknown MetaService response codes

be/src/cloud/cloud_meta_mgr.cpp

@@ -153,11 +153,56 @@ Status bthread_fork_join(std::vector<std::function<Status()>>&& 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<MetaServiceCode>(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<MetaServiceCode>(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 {

be/src/cloud/cloud_meta_mgr.h

@@ -64,7 +64,8 @@ Status bthread_fork_join(const std::vector<std::function<Status()>>& tasks, int
 Status bthread_fork_join(std::vector<std::function<Status()>>&& tasks, int concurrency,
                          std::future<Status>* 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);
 

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<int32_t>(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<int32_t>(MetaServiceCode::OK));
+    EXPECT_EQ(get_response_code(status), MetaServiceCode::OK);
+
+    status.clear_code();
+    status.set_actual_code(static_cast<int32_t>(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<int32_t>::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<int32_t>::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<int32_t>::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();

cloud/src/meta-service/meta_service_helper.h

@@ -18,6 +18,7 @@
 #pragma once
 
 #include <brpc/controller.h>
+#include <fmt/core.h>
 #include <gen_cpp/cloud.pb.h>
 #include <openssl/md5.h>
 
@@ -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<MetaServiceCode, std::string> 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<int32_t>(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) {

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<size_t>(MetaServiceCode_descriptor()->value_count()));
+              static_cast<size_t>(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

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> 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> 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();

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(

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