branch-3.0: [fix](s3client) Avoild dead loop when storage not support ListObjectsV2 #50252

be/src/io/fs/s3_obj_storage_client.cpp

@@ -322,6 +322,12 @@ ObjectStorageResponse S3ObjStorageClient::list_objects(const ObjectStoragePathOp
             files->push_back(std::move(file_info));
         }
         is_trucated = outcome.GetResult().GetIsTruncated();
+        if (is_trucated && outcome.GetResult().GetNextContinuationToken().empty()) {
+            return {convert_to_obj_response(Status::InternalError(
+                    "failed to list {}, is_trucated is true, but next continuation token is empty",
+                    opts.prefix))};
+        }
+
         request.SetContinuationToken(outcome.GetResult().GetNextContinuationToken());
     } while (is_trucated);
     return ObjectStorageResponse::OK();

be/test/io/fs/s3_obj_stroage_client_mock_test.cpp

@@ -0,0 +1,121 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <aws/core/Aws.h>
+#include <aws/s3/S3Client.h>
+#include <aws/s3/model/ListObjectsV2Request.h>
+#include <aws/s3/model/ListObjectsV2Result.h>
+#include <aws/s3/model/Object.h>
+
+#include "gmock/gmock.h"
+#include "io/fs/s3_obj_storage_client.h"
+#include "util/s3_util.h"
+
+using namespace Aws::S3::Model;
+
+namespace doris::io {
+class MockS3Client : public Aws::S3::S3Client {
+public:
+    MockS3Client() {};
+
+    MOCK_METHOD(Aws::S3::Model::ListObjectsV2Outcome, ListObjectsV2,
+                (const Aws::S3::Model::ListObjectsV2Request& request), (const, override));
+};
+
+class S3ObjStorageClientMockTest : public testing::Test {
+    static void SetUpTestSuite() { S3ClientFactory::instance(); };
+    static void TearDownTestSuite() {};
+
+private:
+    static Aws::SDKOptions options;
+};
+
+Aws::SDKOptions S3ObjStorageClientMockTest::options {};
+
+TEST_F(S3ObjStorageClientMockTest, list_objects_compatibility) {
+    // If storage only supports ListObjectsV1, s3_obj_storage_client.list_objects
+    // should return an error.
+    auto mock_s3_client = std::make_shared<MockS3Client>();
+    S3ObjStorageClient s3_obj_storage_client(mock_s3_client);
+
+    std::vector<io::FileInfo> files;
+
+    ListObjectsV2Result result;
+    result.SetIsTruncated(true);
+    EXPECT_CALL(*mock_s3_client, ListObjectsV2(testing::_))
+            .WillOnce(testing::Return(ListObjectsV2Outcome(result)));
+
+    auto response = s3_obj_storage_client.list_objects(
+            {.bucket = "dummy-bucket", .prefix = "S3ObjStorageClientMockTest/list_objects_test"},
+            &files);
+
+    EXPECT_EQ(response.status.code, ErrorCode::INTERNAL_ERROR);
+    files.clear();
+}
+
+ListObjectsV2Result CreatePageResult(const std::string& nextToken,
+                                     const std::vector<std::string>& keys, bool isTruncated) {
+    ListObjectsV2Result result;
+    result.SetIsTruncated(isTruncated);
+    result.SetNextContinuationToken(nextToken);
+    for (const auto& key : keys) {
+        Object obj;
+        obj.SetKey(key);
+        result.AddContents(std::move(obj));
+    }
+    return result;
+}
+
+TEST_F(S3ObjStorageClientMockTest, list_objects_with_pagination) {
+    auto mock_s3_client = std::make_shared<MockS3Client>();
+    S3ObjStorageClient s3_obj_storage_client(mock_s3_client);
+
+    std::vector<std::vector<std::string>> pages = {
+            {"key1", "key2"}, // page1
+            {"key3", "key4"}, // page2
+            {"key5"}          // page3
+    };
+
+    EXPECT_CALL(*mock_s3_client, ListObjectsV2(testing::_))
+            .WillOnce([&](const ListObjectsV2Request& req) {
+                // page1：no ContinuationToken
+                EXPECT_FALSE(req.ContinuationTokenHasBeenSet());
+                return Aws::S3::Model::ListObjectsV2Outcome(
+                        CreatePageResult("token1", pages[0], true));
+            })
+            .WillOnce([&](const ListObjectsV2Request& req) {
+                // page2: token1
+                EXPECT_EQ(req.GetContinuationToken(), "token1");
+                return ListObjectsV2Outcome(CreatePageResult("token2", pages[1], true));
+            })
+            .WillOnce([&](const ListObjectsV2Request& req) {
+                // page3: token2
+                EXPECT_EQ(req.GetContinuationToken(), "token2");
+                return ListObjectsV2Outcome(CreatePageResult("", pages[2], false));
+            });
+
+    std::vector<io::FileInfo> files;
+    auto response = s3_obj_storage_client.list_objects(
+            {.bucket = "dummy-bucket",
+             .prefix = "S3ObjStorageClientMockTest/list_objects_with_pagination"},
+            &files);
+
+    EXPECT_EQ(response.status.code, ErrorCode::OK);
+    EXPECT_EQ(files.size(), 5);
+    files.clear();
+}
+} // namespace doris::io

cloud/src/recycler/s3_obj_client.cpp

@@ -107,6 +107,17 @@ class S3ObjListIterator final : public ObjectListIterator {
             return false;
         }
 
+        if (outcome.GetResult().GetIsTruncated() &&
+            outcome.GetResult().GetNextContinuationToken().empty()) {
+            LOG_WARNING("failed to list objects, isTruncated but no continuation token")
+                    .tag("endpoint", endpoint_)
+                    .tag("bucket", req_.GetBucket())
+                    .tag("prefix", req_.GetPrefix());
+
+            is_valid_ = false;
+            return false;
+        }
+
         has_more_ = outcome.GetResult().GetIsTruncated();
         req_.SetContinuationToken(std::move(
                 const_cast<std::string&&>(outcome.GetResult().GetNextContinuationToken())));

cloud/test/CMakeLists.txt

@@ -49,6 +49,8 @@ add_executable(fdb_injection_test fdb_injection_test.cpp)
 
 add_executable(s3_accessor_test s3_accessor_test.cpp)
 
+add_executable(s3_accessor_mock_test s3_accessor_mock_test.cpp)
+
 add_executable(hdfs_accessor_test hdfs_accessor_test.cpp)
 
 add_executable(stopwatch_test stopwatch_test.cpp)
@@ -86,6 +88,8 @@ target_link_libraries(http_encode_key_test ${TEST_LINK_LIBS})
 
 target_link_libraries(s3_accessor_test ${TEST_LINK_LIBS})
 
+target_link_libraries(s3_accessor_mock_test ${TEST_LINK_LIBS})
+
 target_link_libraries(hdfs_accessor_test ${TEST_LINK_LIBS})
 
 target_link_libraries(stopwatch_test ${TEST_LINK_LIBS})

cloud/test/s3_accessor_mock_test.cpp

@@ -15,13 +15,21 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <aws/core/Aws.h>
+#include <aws/s3/S3Client.h>
+#include <aws/s3/model/ListObjectsV2Request.h>
+#include <aws/s3/model/ListObjectsV2Result.h>
+#include <aws/s3/model/Object.h>
+#include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
 #include "common/config.h"
 #include "common/logging.h"
 #include "cpp/sync_point.h"
+#include "recycler/s3_obj_client.h"
 
 using namespace doris;
+using namespace Aws::S3::Model;
 
 int main(int argc, char** argv) {
     const std::string conf_file = "doris_cloud.conf";
@@ -40,8 +48,39 @@ int main(int argc, char** argv) {
 
 namespace doris::cloud {
 
-TEST(S3ObjClientTest, list_objects) {
-    // TODO
+class S3AccessorMockTest : public testing::Test {
+    static void SetUpTestSuite() { Aws::InitAPI(S3AccessorMockTest::options); };
+    static void TearDownTestSuite() { Aws::ShutdownAPI(options); };
+
+private:
+    static Aws::SDKOptions options;
+};
+
+Aws::SDKOptions S3AccessorMockTest::options {};
+class MockS3Client : public Aws::S3::S3Client {
+public:
+    MockS3Client() {};
+
+    MOCK_METHOD(Aws::S3::Model::ListObjectsV2Outcome, ListObjectsV2,
+                (const Aws::S3::Model::ListObjectsV2Request& request), (const, override));
+};
+
+TEST_F(S3AccessorMockTest, list_objects_compatibility) {
+    // If storage only supports ListObjectsV1, s3_obj_storage_client.list_objects
+    // should return an error.
+    auto mock_s3_client = std::make_shared<MockS3Client>();
+    S3ObjClient s3_obj_client(mock_s3_client, "dummy-endpoint");
+
+    ListObjectsV2Result result;
+    result.SetIsTruncated(true);
+    EXPECT_CALL(*mock_s3_client, ListObjectsV2(testing::_))
+            .WillOnce(testing::Return(ListObjectsV2Outcome(result)));
+
+    auto response = s3_obj_client.list_objects(
+            {.bucket = "dummy-bucket", .key = "S3AccessorMockTest/list_objects_compatibility"});
+
+    EXPECT_FALSE(response->has_next());
+    EXPECT_FALSE(response->is_valid());
 }
 
 } // namespace doris::cloud