This is an automated email from the ASF dual-hosted git repository.
luwei16 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 75ddba65b40 [enhance](cloud) proactively sync tablet meta after alter
(#61585)
75ddba65b40 is described below
commit 75ddba65b40c07a518e8cf7926e20320110d7376
Author: Luwei <[email protected]>
AuthorDate: Sat Aug 15 22:00:40 2026 +0800
[enhance](cloud) proactively sync tablet meta after alter (#61585)
FE now sends a sync_tablet_meta RPC to all alive cloud backends after
alter updates tablet meta in meta service. The request carries affected
tablet ids and is dispatched as a best-effort notification, so alter
success still depends on meta service update instead of backend
acknowledgements.
BE handles the RPC by refreshing meta only for tablets that are already
cached locally. Uncached tablets are skipped, which avoids polluting
tablet cache while still fixing stale compaction policy and related
tablet meta on active compute clusters. The RPC also returns
synced/skipped/failed counts and exposes bvar counters for
observability.
This change adds FE and BE unit tests and a cloud regression suite. The
regression covers cached and uncached multi-cluster behavior, the
negative path with proactive notify disabled, and the version-limit
scenario where a size_based table hits too many versions, is altered to
time_series, and can accept new writes immediately after alter.
---
be/src/cloud/cloud_internal_service.cpp | 77 ++++
be/src/cloud/cloud_internal_service.h | 8 +
be/src/cloud/cloud_tablet.cpp | 95 ++--
be/src/cloud/cloud_tablet_mgr.cpp | 4 +
be/src/cloud/cloud_tablet_mgr.h | 2 +
be/src/service/internal_service.cpp | 9 +
be/src/service/internal_service.h | 4 +
be/test/cloud/cloud_internal_service_test.cpp | 180 ++++++++
be/test/cloud/cloud_tablet_mgr_test.cpp | 32 ++
be/test/cloud/cloud_tablet_test.cpp | 63 ++-
.../cloud/alter/CloudSchemaChangeHandler.java | 83 ++++
.../org/apache/doris/rpc/BackendServiceClient.java | 5 +
.../org/apache/doris/rpc/BackendServiceProxy.java | 12 +
.../cloud/alter/CloudSchemaChangeHandlerTest.java | 478 +++++++++++++++++++++
.../apache/doris/rpc/BackendServiceClientTest.java | 30 ++
.../apache/doris/rpc/BackendServiceProxyTest.java | 54 +++
gensrc/proto/internal_service.proto | 12 +
.../test_cloud_sync_tablet_meta_after_alter.groovy | 284 ++++++++++++
18 files changed, 1384 insertions(+), 48 deletions(-)
diff --git a/be/src/cloud/cloud_internal_service.cpp
b/be/src/cloud/cloud_internal_service.cpp
index 7dca1d1517f..0de9a78127a 100644
--- a/be/src/cloud/cloud_internal_service.cpp
+++ b/be/src/cloud/cloud_internal_service.cpp
@@ -102,6 +102,67 @@ bvar::Adder<int64_t>
g_peer_server_fill_rejected("peer_server_fill_rejected");
bvar::LatencyRecorder g_peer_server_fill_latency("peer_server_fill_latency");
bvar::LatencyRecorder
g_cloud_internal_service_get_file_cache_meta_by_tablet_id_latency(
"cloud_internal_service_get_file_cache_meta_by_tablet_id_latency");
+bvar::Adder<int64_t> g_cloud_sync_tablet_meta_requests_total(
+ "cloud_sync_tablet_meta_requests_total");
+bvar::Adder<int64_t>
g_cloud_sync_tablet_meta_synced_total("cloud_sync_tablet_meta_synced_total");
+bvar::Adder<int64_t>
g_cloud_sync_tablet_meta_skipped_total("cloud_sync_tablet_meta_skipped_total");
+bvar::Adder<int64_t>
g_cloud_sync_tablet_meta_failed_total("cloud_sync_tablet_meta_failed_total");
+
+namespace {
+
+void submit_sync_tablet_meta(CloudStorageEngine& engine, FifoThreadPool&
work_pool,
+ const PSyncTabletMetaRequest* request,
+ PSyncTabletMetaResponse* response,
google::protobuf::Closure* done) {
+ auto start_time = std::chrono::steady_clock::now();
+ bool ret = work_pool.try_offer([engine = &engine, request, response, done,
start_time]() {
+ brpc::ClosureGuard closure_guard(done);
+ LOG(INFO) << "begin to sync tablet meta, request=" <<
request->ShortDebugString();
+ int64_t synced = 0;
+ int64_t skipped = 0;
+ int64_t failed = 0;
+ g_cloud_sync_tablet_meta_requests_total << 1;
+ for (const auto tablet_id : request->tablet_ids()) {
+ auto tablet = engine->tablet_mgr().get_tablet_if_cached(tablet_id);
+ if (!tablet) {
+ ++skipped;
+ continue;
+ }
+ auto st = tablet->sync_meta();
+ if (!st.ok()) {
+ ++failed;
+ LOG(WARNING) << "failed to sync tablet meta from cloud meta
service, tablet="
+ << tablet_id << ", err=" << st;
+ continue;
+ }
+ ++synced;
+ }
+ g_cloud_sync_tablet_meta_synced_total << synced;
+ g_cloud_sync_tablet_meta_skipped_total << skipped;
+ g_cloud_sync_tablet_meta_failed_total << failed;
+ response->set_synced_tablets(synced);
+ response->set_skipped_tablets(skipped);
+ response->set_failed_tablets(failed);
+ Status::OK().to_protobuf(response->mutable_status());
+ auto cost_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+ std::chrono::steady_clock::now() - start_time)
+ .count();
+ LOG(INFO) << "finish to sync tablet meta, request=" <<
request->ShortDebugString()
+ << ", response=" << response->ShortDebugString() << ",
cost_ms=" << cost_ms;
+ });
+ if (!ret) {
+ brpc::ClosureGuard closure_guard(done);
+ Status::InternalError("failed to offer sync_tablet_meta request to
work pool")
+ .to_protobuf(response->mutable_status());
+ auto cost_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+ std::chrono::steady_clock::now() - start_time)
+ .count();
+ LOG(WARNING) << "failed to offer sync_tablet_meta request to work
pool, request="
+ << request->ShortDebugString() << ", response=" <<
response->ShortDebugString()
+ << ", cost_ms=" << cost_ms;
+ }
+}
+
+} // namespace
// Concurrency guard for server-side S3 pull-through fills.
static std::atomic<int32_t> g_active_server_fills {0};
@@ -114,6 +175,22 @@
CloudInternalServiceImpl::CloudInternalServiceImpl(CloudStorageEngine& engine, E
CloudInternalServiceImpl::~CloudInternalServiceImpl() = default;
+void
CloudInternalServiceImpl::sync_tablet_meta(google::protobuf::RpcController*
controller,
+ const PSyncTabletMetaRequest*
request,
+ PSyncTabletMetaResponse*
response,
+ google::protobuf::Closure*
done) {
+ submit_sync_tablet_meta(_engine, _light_work_pool, request, response,
done);
+}
+
+#ifdef BE_TEST
+void test_submit_sync_tablet_meta(CloudStorageEngine& engine, FifoThreadPool&
work_pool,
+ const PSyncTabletMetaRequest* request,
+ PSyncTabletMetaResponse* response,
+ google::protobuf::Closure* done) {
+ submit_sync_tablet_meta(engine, work_pool, request, response, done);
+}
+#endif
+
void
CloudInternalServiceImpl::alter_vault_sync(google::protobuf::RpcController*
controller,
const
doris::PAlterVaultSyncRequest* request,
PAlterVaultSyncResponse*
response,
diff --git a/be/src/cloud/cloud_internal_service.h
b/be/src/cloud/cloud_internal_service.h
index 75fad0ba0ce..524ce14d275 100644
--- a/be/src/cloud/cloud_internal_service.h
+++ b/be/src/cloud/cloud_internal_service.h
@@ -35,6 +35,10 @@ Status test_handle_peer_file_cache_block_request(const
PFetchPeerDataRequest* re
brpc::Controller* cntl);
bool test_try_reject_if_queue_timed_out(std::chrono::steady_clock::time_point
enqueue_ts,
PFetchPeerDataResponse* response);
+void test_submit_sync_tablet_meta(CloudStorageEngine& engine, FifoThreadPool&
work_pool,
+ const PSyncTabletMetaRequest* request,
+ PSyncTabletMetaResponse* response,
+ google::protobuf::Closure* done);
#endif
class CloudInternalServiceImpl final : public PInternalService {
@@ -48,6 +52,10 @@ public:
PAlterVaultSyncResponse* response,
google::protobuf::Closure* done) override;
+ void sync_tablet_meta(google::protobuf::RpcController* controller,
+ const PSyncTabletMetaRequest* request,
PSyncTabletMetaResponse* response,
+ google::protobuf::Closure* done) override;
+
// Get messages (filename, offset, size) about the tablet data in cache
void get_file_cache_meta_by_tablet_id(google::protobuf::RpcController*
controller,
const PGetFileCacheMetaRequest*
request,
diff --git a/be/src/cloud/cloud_tablet.cpp b/be/src/cloud/cloud_tablet.cpp
index 3550d0f3db2..98bf2fdfeca 100644
--- a/be/src/cloud/cloud_tablet.cpp
+++ b/be/src/cloud/cloud_tablet.cpp
@@ -964,6 +964,12 @@ void CloudTablet::get_compaction_status(std::string*
json_result) {
// get snapshot version path json_doc
_timestamped_version_tracker.get_stale_version_path_json_doc(path_arr);
+ rapidjson::Value compaction_policy_value;
+ auto compaction_policy = _tablet_meta->compaction_policy();
+ compaction_policy_value.SetString(compaction_policy.c_str(),
+
cast_set<uint32_t>(compaction_policy.length()),
+ root.GetAllocator());
+ root.AddMember("compaction policy", compaction_policy_value,
root.GetAllocator());
root.AddMember("cumulative point", _cumulative_point.load(),
root.GetAllocator());
rapidjson::Value cumu_value;
std::string format_str =
ToStringFromUnixMillis(_last_cumu_compaction_failure_millis.load());
@@ -1501,9 +1507,9 @@ void CloudTablet::agg_delete_bitmap_for_compaction(
}
Status CloudTablet::sync_meta() {
- if (!config::enable_file_cache) {
- return Status::OK();
- }
+ // Keep tablet metadata refresh serialized with rowset sync and apply
local metadata updates
+ // under _meta_lock. The lock order must stay _sync_meta_lock ->
_meta_lock.
+ std::unique_lock lock(_sync_meta_lock);
TabletMetaSharedPtr tablet_meta;
auto st = _engine.meta_mgr().get_tablet_meta(tablet_id(), &tablet_meta);
@@ -1515,57 +1521,64 @@ Status CloudTablet::sync_meta() {
}
auto new_compaction_policy = tablet_meta->compaction_policy();
- if (_tablet_meta->compaction_policy() != new_compaction_policy) {
- _tablet_meta->set_compaction_policy(new_compaction_policy);
- }
+ auto new_ttl_seconds = tablet_meta->ttl_seconds();
auto new_time_series_compaction_goal_size_mbytes =
tablet_meta->time_series_compaction_goal_size_mbytes();
- if (_tablet_meta->time_series_compaction_goal_size_mbytes() !=
- new_time_series_compaction_goal_size_mbytes) {
- _tablet_meta->set_time_series_compaction_goal_size_mbytes(
- new_time_series_compaction_goal_size_mbytes);
- }
auto new_time_series_compaction_file_count_threshold =
tablet_meta->time_series_compaction_file_count_threshold();
- if (_tablet_meta->time_series_compaction_file_count_threshold() !=
- new_time_series_compaction_file_count_threshold) {
- _tablet_meta->set_time_series_compaction_file_count_threshold(
- new_time_series_compaction_file_count_threshold);
- }
auto new_time_series_compaction_time_threshold_seconds =
tablet_meta->time_series_compaction_time_threshold_seconds();
- if (_tablet_meta->time_series_compaction_time_threshold_seconds() !=
- new_time_series_compaction_time_threshold_seconds) {
- _tablet_meta->set_time_series_compaction_time_threshold_seconds(
- new_time_series_compaction_time_threshold_seconds);
- }
auto new_time_series_compaction_empty_rowsets_threshold =
tablet_meta->time_series_compaction_empty_rowsets_threshold();
- if (_tablet_meta->time_series_compaction_empty_rowsets_threshold() !=
- new_time_series_compaction_empty_rowsets_threshold) {
- _tablet_meta->set_time_series_compaction_empty_rowsets_threshold(
- new_time_series_compaction_empty_rowsets_threshold);
- }
auto new_time_series_compaction_level_threshold =
tablet_meta->time_series_compaction_level_threshold();
- if (_tablet_meta->time_series_compaction_level_threshold() !=
- new_time_series_compaction_level_threshold) {
- _tablet_meta->set_time_series_compaction_level_threshold(
- new_time_series_compaction_level_threshold);
- }
- // Sync disable_auto_compaction (stored in tablet_schema)
auto new_disable_auto_compaction =
tablet_meta->tablet_schema()->disable_auto_compaction();
- if (_tablet_meta->tablet_schema()->disable_auto_compaction() !=
new_disable_auto_compaction) {
- _tablet_meta->mutable_tablet_schema()->set_disable_auto_compaction(
- new_disable_auto_compaction);
- }
- // Sync vertical_compaction_num_columns_per_group
auto new_vertical_compaction_num_columns_per_group =
tablet_meta->vertical_compaction_num_columns_per_group();
- if (_tablet_meta->vertical_compaction_num_columns_per_group() !=
- new_vertical_compaction_num_columns_per_group) {
- _tablet_meta->set_vertical_compaction_num_columns_per_group(
- new_vertical_compaction_num_columns_per_group);
+
+ {
+ std::unique_lock wlock(_meta_lock);
+ if (_tablet_meta->compaction_policy() != new_compaction_policy) {
+ _tablet_meta->set_compaction_policy(new_compaction_policy);
+ }
+ if (_tablet_meta->ttl_seconds() != new_ttl_seconds) {
+ _tablet_meta->set_ttl_seconds(new_ttl_seconds);
+ }
+ if (_tablet_meta->time_series_compaction_goal_size_mbytes() !=
+ new_time_series_compaction_goal_size_mbytes) {
+ _tablet_meta->set_time_series_compaction_goal_size_mbytes(
+ new_time_series_compaction_goal_size_mbytes);
+ }
+ if (_tablet_meta->time_series_compaction_file_count_threshold() !=
+ new_time_series_compaction_file_count_threshold) {
+ _tablet_meta->set_time_series_compaction_file_count_threshold(
+ new_time_series_compaction_file_count_threshold);
+ }
+ if (_tablet_meta->time_series_compaction_time_threshold_seconds() !=
+ new_time_series_compaction_time_threshold_seconds) {
+ _tablet_meta->set_time_series_compaction_time_threshold_seconds(
+ new_time_series_compaction_time_threshold_seconds);
+ }
+ if (_tablet_meta->time_series_compaction_empty_rowsets_threshold() !=
+ new_time_series_compaction_empty_rowsets_threshold) {
+ _tablet_meta->set_time_series_compaction_empty_rowsets_threshold(
+ new_time_series_compaction_empty_rowsets_threshold);
+ }
+ if (_tablet_meta->time_series_compaction_level_threshold() !=
+ new_time_series_compaction_level_threshold) {
+ _tablet_meta->set_time_series_compaction_level_threshold(
+ new_time_series_compaction_level_threshold);
+ }
+ if (_tablet_meta->tablet_schema()->disable_auto_compaction() !=
+ new_disable_auto_compaction) {
+ _tablet_meta->mutable_tablet_schema()->set_disable_auto_compaction(
+ new_disable_auto_compaction);
+ }
+ if (_tablet_meta->vertical_compaction_num_columns_per_group() !=
+ new_vertical_compaction_num_columns_per_group) {
+ _tablet_meta->set_vertical_compaction_num_columns_per_group(
+ new_vertical_compaction_num_columns_per_group);
+ }
}
return Status::OK();
diff --git a/be/src/cloud/cloud_tablet_mgr.cpp
b/be/src/cloud/cloud_tablet_mgr.cpp
index 3affe632996..82dd46a924d 100644
--- a/be/src/cloud/cloud_tablet_mgr.cpp
+++ b/be/src/cloud/cloud_tablet_mgr.cpp
@@ -295,6 +295,10 @@ bool CloudTabletMgr::peek_tablet_meta(int64_t tablet_id,
TabletMetaSharedPtr* ta
return true;
}
+std::shared_ptr<CloudTablet> CloudTabletMgr::get_tablet_if_cached(int64_t
tablet_id) {
+ return _tablet_map->get(tablet_id);
+}
+
void CloudTabletMgr::erase_tablet(int64_t tablet_id) {
auto tablet_id_str = std::to_string(tablet_id);
CacheKey key(tablet_id_str.data(), tablet_id_str.size());
diff --git a/be/src/cloud/cloud_tablet_mgr.h b/be/src/cloud/cloud_tablet_mgr.h
index c44f4f36f8c..5513468809b 100644
--- a/be/src/cloud/cloud_tablet_mgr.h
+++ b/be/src/cloud/cloud_tablet_mgr.h
@@ -62,6 +62,8 @@ public:
// Return true if cached tablet meta is found (without triggering RPC) and
filled.
bool peek_tablet_meta(int64_t tablet_id, TabletMetaSharedPtr* tablet_meta);
+ std::shared_ptr<CloudTablet> get_tablet_if_cached(int64_t tablet_id);
+
void erase_tablet(int64_t tablet_id);
void vacuum_stale_rowsets(const CountDownLatch& stop_latch);
diff --git a/be/src/service/internal_service.cpp
b/be/src/service/internal_service.cpp
index 08b846c888f..371b7803c19 100644
--- a/be/src/service/internal_service.cpp
+++ b/be/src/service/internal_service.cpp
@@ -2313,5 +2313,14 @@ void
PInternalService::request_cdc_client(google::protobuf::RpcController* contr
}
}
+void PInternalService::sync_tablet_meta(google::protobuf::RpcController*
controller,
+ const PSyncTabletMetaRequest* request,
+ PSyncTabletMetaResponse* response,
+ google::protobuf::Closure* done) {
+ brpc::ClosureGuard closure_guard(done);
+ Status::NotSupported("sync_tablet_meta only supports cloud mode")
+ .to_protobuf(response->mutable_status());
+}
+
#include "common/compile_check_avoid_end.h"
} // namespace doris
diff --git a/be/src/service/internal_service.h
b/be/src/service/internal_service.h
index fdbd3a83c13..66816a1b245 100644
--- a/be/src/service/internal_service.h
+++ b/be/src/service/internal_service.h
@@ -223,6 +223,10 @@ public:
const PGetBeResourceRequest* request,
PGetBeResourceResponse* response,
google::protobuf::Closure* done) override;
+ void sync_tablet_meta(google::protobuf::RpcController* controller,
+ const PSyncTabletMetaRequest* request,
PSyncTabletMetaResponse* response,
+ google::protobuf::Closure* done) override;
+
void delete_dictionary(google::protobuf::RpcController* controller,
const PDeleteDictionaryRequest* request,
PDeleteDictionaryResponse* response,
diff --git a/be/test/cloud/cloud_internal_service_test.cpp
b/be/test/cloud/cloud_internal_service_test.cpp
new file mode 100644
index 00000000000..7bcb0f90a42
--- /dev/null
+++ b/be/test/cloud/cloud_internal_service_test.cpp
@@ -0,0 +1,180 @@
+// 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 "cloud/cloud_internal_service.h"
+
+#include <chrono>
+#include <condition_variable>
+#include <functional>
+#include <mutex>
+#include <unordered_map>
+
+#include "cloud/cloud_storage_engine.h"
+#include "cloud/cloud_tablet_mgr.h"
+#include "cpp/sync_point.h"
+#include "gen_cpp/Status_types.h"
+#include "storage/tablet/tablet_meta.h"
+#include "util/uid_util.h"
+#include "util/work_thread_pool.hpp"
+
+namespace doris {
+
+namespace {
+
+class TestClosure final : public google::protobuf::Closure {
+public:
+ explicit TestClosure(std::function<void()> callback) :
_callback(std::move(callback)) {}
+
+ void Run() override {
+ if (_callback) {
+ _callback();
+ }
+ }
+
+private:
+ std::function<void()> _callback;
+};
+
+} // namespace
+
+class CloudInternalServiceTest : public testing::Test {
+public:
+ CloudInternalServiceTest() : _engine(CloudStorageEngine(EngineOptions()))
{}
+
+protected:
+ static constexpr int64_t kSyncedTabletId = 90001;
+ static constexpr int64_t kFailedTabletId = 90002;
+ static constexpr int64_t kUncachedTabletId = 90003;
+
+ void SetUp() override {
+ auto sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+
+ sp->set_call_back("CloudMetaMgr::sync_tablet_rowsets",
+ [](auto&& args) {
try_any_cast_ret<Status>(args)->second = true; });
+ sp->set_call_back("CloudMetaMgr::get_tablet_meta", [this](auto&& args)
{
+ const auto tablet_id = try_any_cast<int64_t>(args[0]);
+ auto* tablet_meta_ptr =
try_any_cast<TabletMetaSharedPtr*>(args[1]);
+ auto* ret = try_any_cast_ret<Status>(args);
+
+ auto& call_count = _tablet_meta_call_count[tablet_id];
+ ++call_count;
+ if (tablet_id == kFailedTabletId && call_count >= 2) {
+ ret->first = Status::InternalError<false>("injected sync_meta
failure");
+ ret->second = true;
+ return;
+ }
+
+ *tablet_meta_ptr =
+ create_tablet_meta(tablet_id, call_count >= 2 ?
"time_series" : "size_based");
+ ret->second = true;
+ });
+ }
+
+ void TearDown() override {
+ auto sp = SyncPoint::get_instance();
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+ }
+
+ TabletMetaSharedPtr create_tablet_meta(int64_t tablet_id, std::string
compaction_policy) {
+ TTabletSchema tablet_schema;
+ TColumn column;
+ column.__set_column_name("c1");
+ column.__set_column_type(TColumnType());
+ column.column_type.__set_type(TPrimitiveType::INT);
+ column.__set_is_key(true);
+ column.__set_aggregation_type(TAggregationType::NONE);
+ column.__set_col_unique_id(0);
+ tablet_schema.__set_columns({column});
+ tablet_schema.__set_keys_type(TKeysType::DUP_KEYS);
+
+ std::unordered_map<uint32_t, uint32_t> col_ordinal_to_unique_id = {{0,
0}};
+ auto tablet_meta = std::make_shared<TabletMeta>(
+ 1, 2, tablet_id, 15674, 4, 5, tablet_schema, 1,
col_ordinal_to_unique_id,
+ UniqueId(9, 10), TTabletType::TABLET_TYPE_DISK,
TCompressionType::LZ4F);
+ tablet_meta->set_compaction_policy(std::move(compaction_policy));
+ return tablet_meta;
+ }
+
+ CloudStorageEngine _engine;
+ std::unordered_map<int64_t, int> _tablet_meta_call_count;
+};
+
+TEST_F(CloudInternalServiceTest,
TestSyncTabletMetaCountsSyncedSkippedAndFailedTablets) {
+ auto synced_res = _engine.tablet_mgr().get_tablet(kSyncedTabletId);
+ ASSERT_TRUE(synced_res.has_value()) << synced_res.error();
+ auto failed_res = _engine.tablet_mgr().get_tablet(kFailedTabletId);
+ ASSERT_TRUE(failed_res.has_value()) << failed_res.error();
+
+ auto cached_synced_tablet = synced_res.value();
+ auto cached_failed_tablet = failed_res.value();
+ ASSERT_EQ("size_based",
cached_synced_tablet->tablet_meta()->compaction_policy());
+ ASSERT_EQ("size_based",
cached_failed_tablet->tablet_meta()->compaction_policy());
+
+ FifoThreadPool work_pool(1, 1, "sync_tablet_meta_test");
+ PSyncTabletMetaRequest request;
+ request.add_tablet_ids(kSyncedTabletId);
+ request.add_tablet_ids(kUncachedTabletId);
+ request.add_tablet_ids(kFailedTabletId);
+ PSyncTabletMetaResponse response;
+
+ std::mutex mutex;
+ std::condition_variable cv;
+ bool done = false;
+ TestClosure closure([&]() {
+ std::lock_guard lock(mutex);
+ done = true;
+ cv.notify_all();
+ });
+
+ test_submit_sync_tablet_meta(_engine, work_pool, &request, &response,
&closure);
+
+ std::unique_lock lock(mutex);
+ ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(5), [&] { return done;
}));
+
+ ASSERT_TRUE(response.has_status());
+ EXPECT_EQ(TStatusCode::OK, response.status().status_code()) <<
response.status().DebugString();
+ EXPECT_EQ(1, response.synced_tablets());
+ EXPECT_EQ(1, response.skipped_tablets());
+ EXPECT_EQ(1, response.failed_tablets());
+ EXPECT_EQ("time_series",
cached_synced_tablet->tablet_meta()->compaction_policy());
+ EXPECT_EQ("size_based",
cached_failed_tablet->tablet_meta()->compaction_policy());
+ EXPECT_EQ(nullptr,
_engine.tablet_mgr().get_tablet_if_cached(kUncachedTabletId));
+}
+
+TEST_F(CloudInternalServiceTest, TestSyncTabletMetaOfferFailure) {
+ FifoThreadPool work_pool(1, 0, "sync_tablet_meta_offer_failure_test");
+ PSyncTabletMetaRequest request;
+ request.add_tablet_ids(kSyncedTabletId);
+ PSyncTabletMetaResponse response;
+
+ bool done = false;
+ TestClosure closure([&]() { done = true; });
+
+ test_submit_sync_tablet_meta(_engine, work_pool, &request, &response,
&closure);
+
+ EXPECT_TRUE(done);
+ ASSERT_TRUE(response.has_status());
+ EXPECT_EQ(TStatusCode::INTERNAL_ERROR, response.status().status_code());
+ EXPECT_TRUE(response.status().error_msgs(0).find(
+ "failed to offer sync_tablet_meta request to work
pool") !=
+ std::string::npos);
+}
+
+} // namespace doris
diff --git a/be/test/cloud/cloud_tablet_mgr_test.cpp
b/be/test/cloud/cloud_tablet_mgr_test.cpp
index 3c9e0b26eb1..c893b72a475 100644
--- a/be/test/cloud/cloud_tablet_mgr_test.cpp
+++ b/be/test/cloud/cloud_tablet_mgr_test.cpp
@@ -178,4 +178,36 @@ TEST_F(CloudTabletMgrTest,
TestConcurrentGetTabletTabletMapConsistency) {
sp->clear_all_call_backs();
}
+TEST_F(CloudTabletMgrTest, TestGetTabletIfCachedOnlyReturnsCachedTablet) {
+ auto sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+
+ sp->set_call_back("CloudMetaMgr::get_tablet_meta", [this](auto&& args) {
+ auto* tablet_meta_ptr = try_any_cast<TabletMetaSharedPtr*>(args[1]);
+ *tablet_meta_ptr = _tablet_meta;
+ try_any_cast_ret<Status>(args)->second = true;
+ });
+ sp->set_call_back("CloudMetaMgr::sync_tablet_rowsets",
+ [](auto&& args) { try_any_cast_ret<Status>(args)->second
= true; });
+
+ CloudTabletMgr mgr(_engine);
+ const int64_t cached_tablet_id = 99999;
+ const int64_t uncached_tablet_id = 100000;
+
+ EXPECT_EQ(nullptr, mgr.get_tablet_if_cached(cached_tablet_id));
+
+ auto res = mgr.get_tablet(cached_tablet_id);
+ ASSERT_TRUE(res.has_value()) << res.error();
+
+ auto cached_tablet = mgr.get_tablet_if_cached(cached_tablet_id);
+ ASSERT_NE(nullptr, cached_tablet);
+ EXPECT_EQ(cached_tablet_id, cached_tablet->tablet_id());
+
+ EXPECT_EQ(nullptr, mgr.get_tablet_if_cached(uncached_tablet_id));
+
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+}
+
} // namespace doris
diff --git a/be/test/cloud/cloud_tablet_test.cpp
b/be/test/cloud/cloud_tablet_test.cpp
index 4c1bfbf79a3..3d3982286d5 100644
--- a/be/test/cloud/cloud_tablet_test.cpp
+++ b/be/test/cloud/cloud_tablet_test.cpp
@@ -721,6 +721,7 @@ protected:
const std::string&
compaction_policy = "size_based") {
TTabletSchema new_schema;
new_schema.__set_disable_auto_compaction(disable_auto_compaction);
+ new_schema.__set_is_in_memory(false);
TColumn col;
col.__set_column_name("test_col_" +
std::to_string(_current_tablet_id));
col.__set_column_type(TColumnType());
@@ -878,8 +879,8 @@ TEST_F(CloudTabletSyncMetaTest,
TestSyncMetaDisableAutoCompactionUnchanged) {
sp->clear_all_call_backs();
}
-// Test sync_meta is skipped when enable_file_cache is false
-TEST_F(CloudTabletSyncMetaTest, TestSyncMetaSkippedWhenFileCacheDisabled) {
+// sync_meta should keep refreshing tablet meta even when file cache is
disabled.
+TEST_F(CloudTabletSyncMetaTest, TestSyncMetaWhenFileCacheDisabled) {
// Disable file cache
config::enable_file_cache = false;
@@ -890,17 +891,22 @@ TEST_F(CloudTabletSyncMetaTest,
TestSyncMetaSkippedWhenFileCacheDisabled) {
sp->clear_all_call_backs();
sp->enable_processing();
+ auto mock_tablet_meta = createMockTabletMeta(true, "time_series");
bool callback_called = false;
sp->set_call_back("CloudMetaMgr::get_tablet_meta",
- [&callback_called](auto&& args) { callback_called =
true; });
+ [mock_tablet_meta, &callback_called](auto&& args) {
+ callback_called = true;
+ auto* tablet_meta_ptr =
try_any_cast<TabletMetaSharedPtr*>(args[1]);
+ *tablet_meta_ptr = mock_tablet_meta;
+ try_any_cast_ret<Status>(args)->second = true;
+ });
- // Call sync_meta - should return early without calling get_tablet_meta
Status st = _tablet->sync_meta();
EXPECT_TRUE(st.ok());
- EXPECT_FALSE(callback_called);
+ EXPECT_TRUE(callback_called);
- // Verify disable_auto_compaction is not changed
-
EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+
EXPECT_TRUE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+ EXPECT_EQ(_tablet->tablet_meta()->compaction_policy(), "time_series");
sp->disable_processing();
sp->clear_all_call_backs();
@@ -919,6 +925,12 @@ TEST_F(CloudTabletSyncMetaTest,
TestSyncMetaMultipleProperties) {
// Create mock tablet meta with updated properties
auto mock_tablet_meta = createMockTabletMeta(true, "time_series");
+ mock_tablet_meta->set_time_series_compaction_goal_size_mbytes(1234);
+ mock_tablet_meta->set_time_series_compaction_file_count_threshold(17);
+ mock_tablet_meta->set_time_series_compaction_time_threshold_seconds(3601);
+ mock_tablet_meta->set_time_series_compaction_empty_rowsets_threshold(9);
+ mock_tablet_meta->set_time_series_compaction_level_threshold(7);
+ mock_tablet_meta->set_vertical_compaction_num_columns_per_group(13);
// Mock get_tablet_meta to return tablet_meta with updated properties
sp->set_call_back("CloudMetaMgr::get_tablet_meta",
[mock_tablet_meta](auto&& args) {
@@ -936,10 +948,47 @@ TEST_F(CloudTabletSyncMetaTest,
TestSyncMetaMultipleProperties) {
// Verify both properties are synced
EXPECT_TRUE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
EXPECT_EQ(_tablet->tablet_meta()->compaction_policy(), "time_series");
+
EXPECT_EQ(_tablet->tablet_meta()->time_series_compaction_goal_size_mbytes(),
1234);
+
EXPECT_EQ(_tablet->tablet_meta()->time_series_compaction_file_count_threshold(),
17);
+
EXPECT_EQ(_tablet->tablet_meta()->time_series_compaction_time_threshold_seconds(),
3601);
+
EXPECT_EQ(_tablet->tablet_meta()->time_series_compaction_empty_rowsets_threshold(),
9);
+
EXPECT_EQ(_tablet->tablet_meta()->time_series_compaction_level_threshold(), 7);
+
EXPECT_EQ(_tablet->tablet_meta()->vertical_compaction_num_columns_per_group(),
13);
+
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+}
+
+TEST_F(CloudTabletSyncMetaTest, TestSyncMetaSyncsTtlWithoutChangingInMemory) {
+ EXPECT_EQ(0, _tablet->tablet_meta()->ttl_seconds());
+ EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->is_in_memory());
+
+ auto sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+
+ auto mock_tablet_meta = createMockTabletMeta(true, "time_series");
+ mock_tablet_meta->set_ttl_seconds(3600);
+ mock_tablet_meta->mutable_tablet_schema()->set_is_in_memory(true);
+
+ sp->set_call_back("CloudMetaMgr::get_tablet_meta",
[mock_tablet_meta](auto&& args) {
+ auto* tablet_meta_ptr = try_any_cast<TabletMetaSharedPtr*>(args[1]);
+ *tablet_meta_ptr = mock_tablet_meta;
+ try_any_cast_ret<Status>(args)->second = true;
+ });
+
+ Status st = _tablet->sync_meta();
+ EXPECT_TRUE(st.ok());
+
+ EXPECT_EQ(3600, _tablet->tablet_meta()->ttl_seconds());
+ EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->is_in_memory());
+
EXPECT_TRUE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+ EXPECT_EQ(_tablet->tablet_meta()->compaction_policy(), "time_series");
sp->disable_processing();
sp->clear_all_call_backs();
}
+
class CloudTabletApplyVisiblePendingTest : public testing::Test {
public:
CloudTabletApplyVisiblePendingTest() :
_engine(CloudStorageEngine(EngineOptions {})) {}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandler.java
index 46ec3ff00a2..1f9037d6454 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandler.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandler.java
@@ -34,12 +34,21 @@ import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.common.util.DynamicPartitionUtil;
import org.apache.doris.common.util.PropertyAnalyzer;
+import org.apache.doris.proto.InternalService;
+import org.apache.doris.rpc.BackendServiceProxy;
import org.apache.doris.service.FrontendOptions;
+import org.apache.doris.system.Backend;
+import org.apache.doris.thrift.TStatusCode;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -538,6 +547,80 @@ public class CloudSchemaChangeHandler extends
SchemaChangeHandler {
if (response.getStatus().getCode() != Cloud.MetaServiceCode.OK) {
throw new UserException(response.getStatus().getMsg());
}
+
+ notifyBackendsToSyncTabletMeta(tableName,
updateTabletReq.getTabletMetaInfosList().stream()
+ .map(Cloud.TabletMetaInfoPB::getTabletId)
+ .collect(Collectors.toList()));
+ }
+ }
+
+ void notifyBackendsToSyncTabletMeta(String tableName, List<Long>
tabletIds) {
+ if (tabletIds.isEmpty()) {
+ return;
+ }
+ if
(DebugPointUtil.isEnable("CloudSchemaChangeHandler.notifyBackendsToSyncTabletMeta.skip"))
{
+ LOG.info("skip sync tablet meta rpc dispatch by debug point,
tableName={}, tabletIds={}",
+ tableName, tabletIds);
+ return;
+ }
+ List<Backend> backends;
+ try {
+ backends =
Env.getCurrentSystemInfo().getAllBackendsByAllCluster().values().asList();
+ } catch (UserException e) {
+ LOG.warn("failed to get alive backends for sync tablet meta,
tableName={}, tabletIds={}",
+ tableName, tabletIds, e);
+ return;
+ }
+
+ InternalService.PSyncTabletMetaRequest request =
InternalService.PSyncTabletMetaRequest.newBuilder()
+ .addAllTabletIds(tabletIds)
+ .build();
+ long dispatchStartTimeMs = System.currentTimeMillis();
+ int sentBackends = 0;
+ int skippedBackends = 0;
+ LOG.info("start to dispatch sync tablet meta rpc, tableName={},
tabletIds={}, request={}",
+ tableName, tabletIds, request);
+ for (Backend backend : backends) {
+ if (!backend.isAlive() || backend.getBrpcPort() <= 0) {
+ skippedBackends++;
+ continue;
+ }
+ try {
+ long rpcStartTimeMs = System.currentTimeMillis();
+ ListenableFuture<InternalService.PSyncTabletMetaResponse>
future =
+
BackendServiceProxy.getInstance().syncTabletMeta(backend.getBrpcAddress(),
request);
+ sentBackends++;
+ Futures.addCallback(future, new
FutureCallback<InternalService.PSyncTabletMetaResponse>() {
+ @Override
+ public void
onSuccess(InternalService.PSyncTabletMetaResponse response) {
+ long costMs = System.currentTimeMillis() -
rpcStartTimeMs;
+ if (response == null || !response.hasStatus()
+ || response.getStatus().getStatusCode() !=
TStatusCode.OK.getValue()) {
+ LOG.warn("sync tablet meta rpc returned non-ok
response, backendId={}, tabletIds={},"
+ + " tableName={}, response={},
costMs={}",
+ backend.getId(), tabletIds, tableName,
response, costMs);
+ return;
+ }
+ LOG.info("sync tablet meta rpc finished, backendId={},
tableName={}, tabletIds={},"
+ + " response={}, costMs={}",
+ backend.getId(), tableName, tabletIds,
response, costMs);
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ LOG.warn("sync tablet meta rpc failed, backendId={},
tableName={}, tabletIds={}, costMs={}",
+ backend.getId(), tableName, tabletIds,
+ System.currentTimeMillis() - rpcStartTimeMs,
t);
+ }
+ }, MoreExecutors.directExecutor());
+ } catch (Exception e) {
+ LOG.warn("failed to dispatch sync tablet meta rpc,
backendId={}, tableName={}, tabletIds={}",
+ backend.getId(), tableName, tabletIds, e);
+ }
}
+ LOG.info("finish dispatching sync tablet meta rpc, tableName={},
tabletIds={}, backendNum={},"
+ + " sentBackends={}, skippedBackends={}, costMs={}",
+ tableName, tabletIds, backends.size(), sentBackends,
skippedBackends,
+ System.currentTimeMillis() - dispatchStartTimeMs);
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java
b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java
index 13489e1894f..d61c1c3e00d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java
@@ -189,6 +189,11 @@ public class BackendServiceClient {
return stub.alterVaultSync(request);
}
+ public ListenableFuture<InternalService.PSyncTabletMetaResponse>
syncTabletMeta(
+ InternalService.PSyncTabletMetaRequest request) {
+ return stub.syncTabletMeta(request);
+ }
+
public Future<InternalService.PGetBeResourceResponse>
getBeResource(InternalService.PGetBeResourceRequest request,
int timeoutSec) {
return stub.withDeadlineAfter(timeoutSec,
TimeUnit.SECONDS).getBeResource(request);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java
b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java
index 15f35d37117..70e96650622 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java
@@ -548,6 +548,18 @@ public class BackendServiceProxy {
}
}
+ public ListenableFuture<InternalService.PSyncTabletMetaResponse>
syncTabletMeta(TNetworkAddress address,
+ InternalService.PSyncTabletMetaRequest request) throws
RpcException {
+ try {
+ final BackendServiceClient client = getProxy(address);
+ return client.syncTabletMeta(request);
+ } catch (Throwable e) {
+ LOG.warn("failed to sync tablet meta from address={}:{}",
address.getHostname(),
+ address.getPort(), e);
+ throw new RpcException(address.getHostname(), e.getMessage());
+ }
+ }
+
public Future<InternalService.PFetchRemoteSchemaResponse>
fetchRemoteTabletSchemaAsync(
TNetworkAddress address, InternalService.PFetchRemoteSchemaRequest
request) throws RpcException {
try {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandlerTest.java
new file mode 100644
index 00000000000..54d55797443
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandlerTest.java
@@ -0,0 +1,478 @@
+// 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.
+
+package org.apache.doris.cloud.alter;
+
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.MaterializedIndex;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Partition;
+import org.apache.doris.catalog.Table;
+import org.apache.doris.cloud.proto.Cloud;
+import org.apache.doris.cloud.rpc.MetaServiceProxy;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.DebugPointUtil;
+import org.apache.doris.proto.InternalService;
+import org.apache.doris.rpc.BackendServiceProxy;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TStatusCode;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.Futures;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class CloudSchemaChangeHandlerTest {
+ private int originalCloudTxnTabletBatchSize;
+ private String originalCloudUniqueId;
+ private String originalMetaServiceEndpoint;
+ private boolean originalEnableDebugPoints;
+
+ @Before
+ public void setUp() {
+ originalCloudTxnTabletBatchSize = Config.cloud_txn_tablet_batch_size;
+ originalCloudUniqueId = Config.cloud_unique_id;
+ originalMetaServiceEndpoint = Config.meta_service_endpoint;
+ originalEnableDebugPoints = Config.enable_debug_points;
+ Config.cloud_txn_tablet_batch_size = 2;
+ Config.cloud_unique_id = "cloud-test";
+ Config.meta_service_endpoint = "127.0.0.1:20121";
+ Config.enable_debug_points = false;
+ DebugPointUtil.clearDebugPoints();
+ }
+
+ @After
+ public void tearDown() {
+ Config.cloud_txn_tablet_batch_size = originalCloudTxnTabletBatchSize;
+ Config.cloud_unique_id = originalCloudUniqueId;
+ Config.meta_service_endpoint = originalMetaServiceEndpoint;
+ Config.enable_debug_points = originalEnableDebugPoints;
+ DebugPointUtil.clearDebugPoints();
+ }
+
+ @Test
+ public void testUpdateTablePropertiesNotifiesAllAliveBackendsInBatches()
throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Database db = createMockDatabaseWithThreeTablets();
+
+ Backend aliveBackend1 = new Backend(1L, "be-1", 9050);
+ aliveBackend1.setAlive(true);
+ aliveBackend1.setBrpcPort(8060);
+ Backend aliveBackend2 = new Backend(2L, "be-2", 9050);
+ aliveBackend2.setAlive(true);
+ aliveBackend2.setBrpcPort(8061);
+ Backend deadBackend = new Backend(3L, "be-dead", 9050);
+ deadBackend.setAlive(false);
+ deadBackend.setBrpcPort(8062);
+
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of(
+ aliveBackend1.getId(), aliveBackend1,
+ aliveBackend2.getId(), aliveBackend2,
+ deadBackend.getId(), deadBackend));
+
+ Env env = Mockito.mock(Env.class);
+
+ MetaServiceProxy metaServiceProxy =
Mockito.mock(MetaServiceProxy.class);
+
Mockito.when(metaServiceProxy.updateTablet(Mockito.any())).thenReturn(okUpdateTabletResponse());
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.any(),
Mockito.any()))
+
.thenReturn(Futures.immediateFuture(okSyncTabletMetaResponse()));
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<MetaServiceProxy> metaProxyMock =
Mockito.mockStatic(MetaServiceProxy.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+ envMock.when(Env::getCurrentEnv).thenReturn(env);
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
metaProxyMock.when(MetaServiceProxy::getInstance).thenReturn(metaServiceProxy);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ Map<String, String> properties = new HashMap<>();
+ properties.put("compaction_policy", "time_series");
+ handler.updateTableProperties(db, "tbl", properties);
+ }
+
+ ArgumentCaptor<Cloud.UpdateTabletRequest> updateCaptor =
+ ArgumentCaptor.forClass(Cloud.UpdateTabletRequest.class);
+ Mockito.verify(metaServiceProxy,
Mockito.times(2)).updateTablet(updateCaptor.capture());
+ List<Cloud.UpdateTabletRequest> updateRequests =
updateCaptor.getAllValues();
+ Assert.assertEquals(Arrays.asList(101L, 102L),
+ updateRequests.get(0).getTabletMetaInfosList().stream()
+
.map(Cloud.TabletMetaInfoPB::getTabletId).collect(Collectors.toList()));
+ Assert.assertEquals(Arrays.asList(103L),
+ updateRequests.get(1).getTabletMetaInfosList().stream()
+
.map(Cloud.TabletMetaInfoPB::getTabletId).collect(Collectors.toList()));
+
+ ArgumentCaptor<TNetworkAddress> addressCaptor =
ArgumentCaptor.forClass(TNetworkAddress.class);
+ ArgumentCaptor<InternalService.PSyncTabletMetaRequest> syncCaptor =
+
ArgumentCaptor.forClass(InternalService.PSyncTabletMetaRequest.class);
+ Mockito.verify(backendServiceProxy, Mockito.times(4))
+ .syncTabletMeta(addressCaptor.capture(), syncCaptor.capture());
+
+ List<TNetworkAddress> addresses = addressCaptor.getAllValues();
+ Assert.assertFalse(addresses.stream().anyMatch(addr ->
"be-dead".equals(addr.getHostname())));
+ Assert.assertEquals(2L, addresses.stream().filter(addr ->
"be-1".equals(addr.getHostname())).count());
+ Assert.assertEquals(2L, addresses.stream().filter(addr ->
"be-2".equals(addr.getHostname())).count());
+
+ List<InternalService.PSyncTabletMetaRequest> syncRequests =
syncCaptor.getAllValues();
+ Assert.assertEquals(Arrays.asList(101L, 102L),
syncRequests.get(0).getTabletIdsList());
+ Assert.assertEquals(Arrays.asList(101L, 102L),
syncRequests.get(1).getTabletIdsList());
+ Assert.assertEquals(Arrays.asList(103L),
syncRequests.get(2).getTabletIdsList());
+ Assert.assertEquals(Arrays.asList(103L),
syncRequests.get(3).getTabletIdsList());
+ }
+
+ @Test
+ public void testUpdateTablePropertiesThrowsWhenUpdateTabletFails() throws
Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Database db = createMockDatabaseWithThreeTablets();
+
+ MetaServiceProxy metaServiceProxy =
Mockito.mock(MetaServiceProxy.class);
+ Mockito.when(metaServiceProxy.updateTablet(Mockito.any()))
+ .thenThrow(new RuntimeException("update failed"));
+
+ try (MockedStatic<MetaServiceProxy> metaProxyMock =
Mockito.mockStatic(MetaServiceProxy.class)) {
+
metaProxyMock.when(MetaServiceProxy::getInstance).thenReturn(metaServiceProxy);
+
+ Map<String, String> properties = new HashMap<>();
+ properties.put("compaction_policy", "time_series");
+ UserException exception = Assert.assertThrows(UserException.class,
+ () -> handler.updateTableProperties(db, "tbl",
properties));
+ Assert.assertTrue(exception.getMessage().contains("update
failed"));
+ }
+ }
+
+ @Test
+ public void testUpdateTablePropertiesThrowsWhenUpdateTabletResponseNotOk()
throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Database db = createMockDatabaseWithThreeTablets();
+
+ MetaServiceProxy metaServiceProxy =
Mockito.mock(MetaServiceProxy.class);
+ Mockito.when(metaServiceProxy.updateTablet(Mockito.any()))
+ .thenReturn(Cloud.UpdateTabletResponse.newBuilder()
+ .setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
+
.setCode(Cloud.MetaServiceCode.INVALID_ARGUMENT)
+ .setMsg("meta service rejected"))
+ .build());
+
+ try (MockedStatic<MetaServiceProxy> metaProxyMock =
Mockito.mockStatic(MetaServiceProxy.class)) {
+
metaProxyMock.when(MetaServiceProxy::getInstance).thenReturn(metaServiceProxy);
+
+ Map<String, String> properties = new HashMap<>();
+ properties.put("compaction_policy", "time_series");
+ UserException exception = Assert.assertThrows(UserException.class,
+ () -> handler.updateTableProperties(db, "tbl",
properties));
+ Assert.assertTrue(exception.getMessage().contains("meta service
rejected"));
+ }
+ }
+
+ @Test
+ public void testNotifyBackendsToSyncTabletMetaSwallowsDispatchFailure()
throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Backend aliveBackend1 = new Backend(1L, "be-1", 9050);
+ aliveBackend1.setAlive(true);
+ aliveBackend1.setBrpcPort(8060);
+ Backend aliveBackend2 = new Backend(2L, "be-2", 9050);
+ aliveBackend2.setAlive(true);
+ aliveBackend2.setBrpcPort(8061);
+
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of(
+ aliveBackend1.getId(), aliveBackend1,
+ aliveBackend2.getId(), aliveBackend2));
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.argThat(
+ addr -> addr != null &&
"be-1".equals(addr.getHostname())), Mockito.any()))
+
.thenReturn(Futures.immediateFuture(okSyncTabletMetaResponse()));
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.argThat(
+ addr -> addr != null &&
"be-2".equals(addr.getHostname())), Mockito.any()))
+ .thenThrow(new org.apache.doris.rpc.RpcException("be-2",
"dispatch failed"));
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ Mockito.verify(backendServiceProxy, Mockito.times(2))
+ .syncTabletMeta(Mockito.any(), Mockito.any());
+ }
+
+ @Test
+ public void testNotifyBackendsToSyncTabletMetaReturnsForEmptyTabletList() {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+
+ try (MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList());
+ }
+
+ Mockito.verifyNoInteractions(backendServiceProxy);
+ }
+
+ @Test
+ public void
testNotifyBackendsToSyncTabletMetaReturnsWhenDebugPointEnabled() {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Config.enable_debug_points = true;
+
DebugPointUtil.addDebugPoint("CloudSchemaChangeHandler.notifyBackendsToSyncTabletMeta.skip");
+
+ try (MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ Mockito.verifyNoInteractions(backendServiceProxy);
+ }
+
+ @Test
+ public void testNotifyBackendsToSyncTabletMetaSkipsUnavailableBackends()
throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Backend aliveBackend = new Backend(1L, "be-1", 9050);
+ aliveBackend.setAlive(true);
+ aliveBackend.setBrpcPort(8060);
+ Backend deadBackend = new Backend(2L, "be-dead", 9050);
+ deadBackend.setAlive(false);
+ deadBackend.setBrpcPort(8061);
+ Backend noBrpcBackend = new Backend(3L, "be-no-brpc", 9050);
+ noBrpcBackend.setAlive(true);
+ noBrpcBackend.setBrpcPort(0);
+
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of(
+ aliveBackend.getId(), aliveBackend,
+ deadBackend.getId(), deadBackend,
+ noBrpcBackend.getId(), noBrpcBackend));
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.any(),
Mockito.any()))
+
.thenReturn(Futures.immediateFuture(okSyncTabletMetaResponse()));
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ ArgumentCaptor<InternalService.PSyncTabletMetaRequest> syncCaptor =
+
ArgumentCaptor.forClass(InternalService.PSyncTabletMetaRequest.class);
+ Mockito.verify(backendServiceProxy, Mockito.times(1))
+ .syncTabletMeta(Mockito.argThat(addr ->
"be-1".equals(addr.getHostname())), syncCaptor.capture());
+ Assert.assertEquals(Arrays.asList(101L, 102L),
syncCaptor.getValue().getTabletIdsList());
+ }
+
+ @Test
+ public void testNotifyBackendsToSyncTabletMetaSwallowsAsyncFailure()
throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Backend aliveBackend = new Backend(1L, "be-1", 9050);
+ aliveBackend.setAlive(true);
+ aliveBackend.setBrpcPort(8060);
+
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of(
+ aliveBackend.getId(), aliveBackend));
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.any(),
Mockito.any()))
+ .thenReturn(Futures.immediateFailedFuture(new
RuntimeException("async failed")));
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ Mockito.verify(backendServiceProxy, Mockito.times(1))
+ .syncTabletMeta(Mockito.any(), Mockito.any());
+ }
+
+ @Test
+ public void testNotifyBackendsToSyncTabletMetaSwallowsNullResponse()
throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Backend aliveBackend = new Backend(1L, "be-1", 9050);
+ aliveBackend.setAlive(true);
+ aliveBackend.setBrpcPort(8060);
+
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of(
+ aliveBackend.getId(), aliveBackend));
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.any(),
Mockito.any()))
+ .thenReturn(Futures.immediateFuture(null));
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ Mockito.verify(backendServiceProxy, Mockito.times(1))
+ .syncTabletMeta(Mockito.any(), Mockito.any());
+ }
+
+ @Test
+ public void
testNotifyBackendsToSyncTabletMetaSwallowsResponseWithoutStatus() throws
Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Backend aliveBackend = new Backend(1L, "be-1", 9050);
+ aliveBackend.setAlive(true);
+ aliveBackend.setBrpcPort(8060);
+
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of(
+ aliveBackend.getId(), aliveBackend));
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.any(),
Mockito.any()))
+
.thenReturn(Futures.immediateFuture(InternalService.PSyncTabletMetaResponse.newBuilder()
+ .setSyncedTablets(2)
+ .build()));
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ Mockito.verify(backendServiceProxy, Mockito.times(1))
+ .syncTabletMeta(Mockito.any(), Mockito.any());
+ }
+
+ @Test
+ public void testNotifyBackendsToSyncTabletMetaSwallowsNonOkResponse()
throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Backend aliveBackend1 = new Backend(1L, "be-1", 9050);
+ aliveBackend1.setAlive(true);
+ aliveBackend1.setBrpcPort(8060);
+ Backend aliveBackend2 = new Backend(2L, "be-2", 9050);
+ aliveBackend2.setAlive(true);
+ aliveBackend2.setBrpcPort(8061);
+
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of(
+ aliveBackend1.getId(), aliveBackend1,
+ aliveBackend2.getId(), aliveBackend2));
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.any(),
Mockito.any()))
+
.thenReturn(Futures.immediateFuture(InternalService.PSyncTabletMetaResponse.newBuilder()
+
.setStatus(org.apache.doris.proto.Types.PStatus.newBuilder()
+
.setStatusCode(TStatusCode.CANCELLED.getValue()))
+ .setFailedTablets(2)
+ .build()));
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ Mockito.verify(backendServiceProxy, Mockito.times(2))
+ .syncTabletMeta(Mockito.any(), Mockito.any());
+ }
+
+ @Test
+ public void
testNotifyBackendsToSyncTabletMetaReturnsWhenBackendDiscoveryFails() throws
Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+ Mockito.when(systemInfoService.getAllBackendsByAllCluster())
+ .thenThrow(new
org.apache.doris.common.AnalysisException("backend discovery failed"));
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ Mockito.verifyNoInteractions(backendServiceProxy);
+ }
+
+ private Cloud.UpdateTabletResponse okUpdateTabletResponse() {
+ return Cloud.UpdateTabletResponse.newBuilder()
+
.setStatus(Cloud.MetaServiceResponseStatus.newBuilder().setCode(Cloud.MetaServiceCode.OK))
+ .build();
+ }
+
+ private Database createMockDatabaseWithThreeTablets() throws Exception {
+ Database db = Mockito.mock(Database.class);
+ OlapTable table = Mockito.mock(OlapTable.class);
+ Partition partition = Mockito.mock(Partition.class);
+ MaterializedIndex index = Mockito.mock(MaterializedIndex.class);
+ org.apache.doris.catalog.Tablet tablet1 =
Mockito.mock(org.apache.doris.catalog.Tablet.class);
+ org.apache.doris.catalog.Tablet tablet2 =
Mockito.mock(org.apache.doris.catalog.Tablet.class);
+ org.apache.doris.catalog.Tablet tablet3 =
Mockito.mock(org.apache.doris.catalog.Tablet.class);
+
+ Mockito.when(db.getTableOrMetaException("tbl",
Table.TableType.OLAP)).thenReturn(table);
+ Mockito.when(table.getName()).thenReturn("tbl");
+ Mockito.when(table.getCompactionPolicy()).thenReturn("size_based");
+ Mockito.when(table.getKeysType()).thenReturn(KeysType.DUP_KEYS);
+
Mockito.when(table.getPartitions()).thenReturn(Arrays.asList(partition));
+ Mockito.when(table.getPartition("p1")).thenReturn(partition);
+ Mockito.when(partition.getName()).thenReturn("p1");
+
Mockito.when(partition.getMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE,
true))
+ .thenReturn(Arrays.asList(index));
+ Mockito.when(index.getTablets()).thenReturn(Arrays.asList(tablet1,
tablet2, tablet3));
+ Mockito.when(tablet1.getId()).thenReturn(101L);
+ Mockito.when(tablet2.getId()).thenReturn(102L);
+ Mockito.when(tablet3.getId()).thenReturn(103L);
+ return db;
+ }
+
+ private InternalService.PSyncTabletMetaResponse okSyncTabletMetaResponse()
{
+ return InternalService.PSyncTabletMetaResponse.newBuilder()
+ .setStatus(org.apache.doris.proto.Types.PStatus.newBuilder()
+ .setStatusCode(TStatusCode.OK.getValue()))
+ .setSyncedTablets(1)
+ .build();
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceClientTest.java
b/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceClientTest.java
index 36eb030a19e..8a16137f771 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceClientTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceClientTest.java
@@ -19,13 +19,18 @@ package org.apache.doris.rpc;
import org.apache.doris.common.Config;
import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.proto.InternalService;
+import org.apache.doris.proto.PBackendServiceGrpc;
import org.apache.doris.thrift.TNetworkAddress;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
import io.grpc.ManagedChannel;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.Mockito;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -228,4 +233,29 @@ public class BackendServiceClientTest {
client1.shutdown();
client2.shutdown();
}
+
+ @Test
+ public void testSyncTabletMeta() {
+ TNetworkAddress address = new TNetworkAddress("localhost", 9060);
+ BackendServiceClient client = new BackendServiceClient(address,
"127.0.0.1", executor);
+
+ PBackendServiceGrpc.PBackendServiceFutureStub stub =
+
Mockito.mock(PBackendServiceGrpc.PBackendServiceFutureStub.class);
+ Deencapsulation.setField(client, "stub", stub);
+
+ InternalService.PSyncTabletMetaRequest request =
InternalService.PSyncTabletMetaRequest.newBuilder()
+ .addTabletIds(10001L)
+ .build();
+ ListenableFuture<InternalService.PSyncTabletMetaResponse>
expectedFuture = Futures.immediateFuture(
+ InternalService.PSyncTabletMetaResponse.newBuilder()
+ .setSyncedTablets(1)
+ .build());
+ Mockito.when(stub.syncTabletMeta(request)).thenReturn(expectedFuture);
+
+ ListenableFuture<InternalService.PSyncTabletMetaResponse> actualFuture
= client.syncTabletMeta(request);
+
+ Assert.assertSame(expectedFuture, actualFuture);
+ Mockito.verify(stub).syncTabletMeta(request);
+ client.shutdown();
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceProxyTest.java
b/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceProxyTest.java
index 2a897f0e400..427ff457034 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceProxyTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceProxyTest.java
@@ -21,8 +21,11 @@ import org.apache.doris.catalog.Env;
import org.apache.doris.common.Config;
import org.apache.doris.common.DNSCache;
import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.proto.InternalService;
import org.apache.doris.thrift.TNetworkAddress;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -30,6 +33,7 @@ import org.junit.Test;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import java.lang.reflect.Constructor;
import java.net.UnknownHostException;
import java.util.Map;
@@ -300,4 +304,54 @@ public class BackendServiceProxyTest {
Assert.assertTrue(serviceMap.containsKey(address1));
Assert.assertTrue(serviceMap.containsKey(address2));
}
+
+ @Test
+ public void testSyncTabletMeta() throws Exception {
+ String hostname = "backend-host.example.com";
+ String resolvedIp = "10.0.0.1";
+ int port = 9060;
+ TNetworkAddress address = new TNetworkAddress(hostname, port);
+ InternalService.PSyncTabletMetaRequest request =
InternalService.PSyncTabletMetaRequest.newBuilder()
+ .addTabletIds(10001L)
+ .build();
+
+ BackendServiceClient client = Mockito.mock(BackendServiceClient.class);
+ Mockito.when(client.isNormalState()).thenReturn(true);
+ Mockito.when(client.isUsingLatestChannelConfig()).thenReturn(true);
+ ListenableFuture<InternalService.PSyncTabletMetaResponse>
expectedFuture = Futures.immediateFuture(
+ InternalService.PSyncTabletMetaResponse.newBuilder()
+ .setSyncedTablets(1)
+ .build());
+
Mockito.when(client.syncTabletMeta(request)).thenReturn(expectedFuture);
+ Mockito.when(mockDnsCache.get(hostname)).thenReturn(resolvedIp);
+
+ Map<TNetworkAddress, Object> serviceMap =
Deencapsulation.getField(proxy, "serviceMap");
+ serviceMap.put(address, newBackendServiceClientExtIp(resolvedIp,
client));
+
+ ListenableFuture<InternalService.PSyncTabletMetaResponse> actualFuture
= proxy.syncTabletMeta(address, request);
+
+ Assert.assertSame(expectedFuture, actualFuture);
+ Mockito.verify(client).syncTabletMeta(request);
+ }
+
+ @Test
+ public void testSyncTabletMetaThrowsRpcException() {
+ String hostname = "non-existent-host.example.com";
+ TNetworkAddress address = new TNetworkAddress(hostname, 9060);
+ InternalService.PSyncTabletMetaRequest request =
InternalService.PSyncTabletMetaRequest.newBuilder()
+ .addTabletIds(10001L)
+ .build();
+ Mockito.when(mockDnsCache.get(hostname)).thenReturn("");
+
+ RpcException exception = Assert.assertThrows(RpcException.class, () ->
proxy.syncTabletMeta(address, request));
+ Assert.assertTrue(exception.getMessage().contains(hostname));
+ }
+
+ private Object newBackendServiceClientExtIp(String realIp,
BackendServiceClient client) throws Exception {
+ Class<?> clazz =
Class.forName("org.apache.doris.rpc.BackendServiceProxy$BackendServiceClientExtIp");
+ Constructor<?> constructor =
clazz.getDeclaredConstructor(BackendServiceProxy.class,
+ String.class, BackendServiceClient.class);
+ constructor.setAccessible(true);
+ return constructor.newInstance(proxy, realIp, client);
+ }
}
diff --git a/gensrc/proto/internal_service.proto
b/gensrc/proto/internal_service.proto
index 92a53936a89..54c7af9171b 100644
--- a/gensrc/proto/internal_service.proto
+++ b/gensrc/proto/internal_service.proto
@@ -1101,6 +1101,17 @@ message PAlterVaultSyncRequest {
message PAlterVaultSyncResponse {
}
+message PSyncTabletMetaRequest {
+ repeated int64 tablet_ids = 1;
+}
+
+message PSyncTabletMetaResponse {
+ optional PStatus status = 1;
+ optional int64 synced_tablets = 2;
+ optional int64 skipped_tablets = 3;
+ optional int64 failed_tablets = 4;
+}
+
message PGetBeResourceRequest {
}
@@ -1266,6 +1277,7 @@ service PBackendService {
rpc fetch_remote_tablet_schema(PFetchRemoteSchemaRequest) returns
(PFetchRemoteSchemaResponse);
rpc test_jdbc_connection(PJdbcTestConnectionRequest) returns
(PJdbcTestConnectionResult);
rpc alter_vault_sync(PAlterVaultSyncRequest) returns
(PAlterVaultSyncResponse);
+ rpc sync_tablet_meta(PSyncTabletMetaRequest) returns
(PSyncTabletMetaResponse);
rpc get_be_resource(PGetBeResourceRequest) returns
(PGetBeResourceResponse);
rpc delete_dictionary(PDeleteDictionaryRequest) returns
(PDeleteDictionaryResponse);
rpc commit_refresh_dictionary(PCommitRefreshDictionaryRequest) returns
(PCommitRefreshDictionaryResponse);
diff --git
a/regression-test/suites/fault_injection_p0/cloud/test_cloud_sync_tablet_meta_after_alter.groovy
b/regression-test/suites/fault_injection_p0/cloud/test_cloud_sync_tablet_meta_after_alter.groovy
new file mode 100644
index 00000000000..17bdeae2d12
--- /dev/null
+++
b/regression-test/suites/fault_injection_p0/cloud/test_cloud_sync_tablet_meta_after_alter.groovy
@@ -0,0 +1,284 @@
+// 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.
+
+suite("test_cloud_sync_tablet_meta_after_alter", "nonConcurrent") {
+ if (!isCloudMode()) {
+ return
+ }
+ if (!getFeConfig("enable_debug_points").equalsIgnoreCase("true")) {
+ logger.info("enable_debug_points=false, skip")
+ return
+ }
+ if (context.config.multiClusterBes == null ||
context.config.multiClusterBes.trim().isEmpty()) {
+ logger.info("multiClusterBes is empty, skip")
+ return
+ }
+
+ GetDebugPoint().clearDebugPointsForAllFEs()
+ GetDebugPoint().clearDebugPointsForAllBEs()
+
+ def cluster0 = "sync_meta_cluster0"
+ def cluster1 = "sync_meta_cluster1"
+ def clusterId0 = "sync_meta_cluster_id0"
+ def clusterId1 = "sync_meta_cluster_id1"
+ def dbName = "reg_sync_tablet_meta_db"
+ def tableName = "test_cloud_sync_tablet_meta_after_alter"
+ def versionLimitTableName =
"test_cloud_sync_tablet_meta_version_limit_after_alter"
+
+ List<String> ipList = []
+ List<String> hbPortList = []
+ List<String> beUniqueIdList = []
+
+ String[] bes = context.config.multiClusterBes.split(',')
+ for (String values : bes) {
+ if (ipList.size() == 2) {
+ break
+ }
+ String[] beInfo = values.split(':')
+ ipList.add(beInfo[0])
+ hbPortList.add(beInfo[1])
+ beUniqueIdList.add(beInfo[3])
+ }
+ assertTrue(ipList.size() >= 2, "need at least two backends in
multiClusterBes")
+
+ def getBackendForCluster = { String clusterName ->
+ def backend = sql_return_maparray("show backends").find {
it.Tag.contains(clusterName) }
+ assert backend != null: "backend for cluster ${clusterName} not found"
+ return backend
+ }
+
+ def getBvar = { backend, String name ->
+ def url = "http://${backend.Host}:${backend.BrpcPort}/vars/${name}";
+ def (code, out, err) = curl("GET", url)
+ assertEquals(0, code)
+ def matcher = (out =~
/(?m)^${java.util.regex.Pattern.quote(name)}\\s*:\\s*(\\d+)$/)
+ assertTrue(matcher.find(), "failed to parse bvar ${name} from ${url},
out=${out}, err=${err}")
+ return matcher.group(1).toLong()
+ }
+
+ def waitForBvarIncrease = { backend, String name, long baseline, long
delta ->
+ for (int retry = 0; retry < 20; retry++) {
+ def current = getBvar(backend, name)
+ logger.info("wait bvar ${name} on
${backend.Host}:${backend.BrpcPort}, baseline=${baseline}, current=${current}")
+ if (current >= baseline + delta) {
+ return current
+ }
+ sleep(500)
+ }
+ return getBvar(backend, name)
+ }
+
+ def getCompactionPolicy = { String currentTableName ->
+ def tablets = sql_return_maparray("""show tablets from
${currentTableName};""")
+ assertEquals(1, tablets.size())
+ def (code, out, err) = curl("GET", tablets[0].CompactionStatus)
+ assertEquals(0, code)
+ def tabletJson = parseJson(out.trim())
+ return tabletJson["compaction policy"]
+ }
+
+ def useCluster = { String clusterName ->
+ sql """use @${clusterName}"""
+ sql """use ${dbName}"""
+ }
+
+ def cleanupClusters = {
+ for (uniqueId in beUniqueIdList) {
+ def resp = get_cluster.call(uniqueId)
+ for (cluster in resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id)
+ }
+ }
+ }
+ wait_cluster_change()
+ }
+
+ setBeConfigTemporary([
+ tablet_sync_interval_s: 18000,
+ schedule_sync_tablets_interval_s: 18000
+ ]) {
+ try {
+ cleanupClusters.call()
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
cluster0, clusterId0)
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
cluster1, clusterId1)
+ wait_cluster_change()
+
+ sql """use @${cluster0}"""
+ sql """create database if not exists ${dbName}"""
+ sql """use ${dbName}"""
+ sql """drop table if exists ${tableName} force"""
+ sql """drop table if exists ${versionLimitTableName} force"""
+ sql """
+ CREATE TABLE ${tableName} (
+ id INT NOT NULL,
+ value INT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "disable_auto_compaction" = "true"
+ )
+ """
+ sql """insert into ${tableName} values (1, 10), (2, 20), (3, 30)"""
+
+ def be0 = getBackendForCluster(cluster0)
+ def be1 = getBackendForCluster(cluster1)
+
+ // Case 1: warm only cluster0, alter property, and verify cluster1
is notified but skipped.
+ useCluster.call(cluster0)
+ qt_case1_cluster0_warmup """select count(*) from ${tableName}"""
+ assertEquals("size_based", getCompactionPolicy(tableName))
+
+ def case1Synced0 = getBvar(be0,
"cloud_sync_tablet_meta_synced_total")
+ def case1Skipped1 = getBvar(be1,
"cloud_sync_tablet_meta_skipped_total")
+
+ sql """alter table ${tableName} set ("compaction_policy" =
"time_series")"""
+
+ assertTrue(waitForBvarIncrease(be0,
"cloud_sync_tablet_meta_synced_total", case1Synced0, 1) >= case1Synced0 + 1)
+ assertTrue(waitForBvarIncrease(be1,
"cloud_sync_tablet_meta_skipped_total", case1Skipped1, 1) >= case1Skipped1 + 1)
+
+ useCluster.call(cluster0)
+ qt_case1_cluster0_after_alter """select count(*) from
${tableName}"""
+ assertEquals("time_series", getCompactionPolicy(tableName))
+
+ useCluster.call(cluster1)
+ qt_case1_cluster1_first_access """select count(*) from
${tableName}"""
+ assertEquals("time_series", getCompactionPolicy(tableName))
+
+ // Case 2: warm both clusters, alter property again, and verify
both clusters sync immediately.
+ useCluster.call(cluster0)
+ qt_case2_cluster0_warmup """select count(*) from ${tableName}"""
+ useCluster.call(cluster1)
+ qt_case2_cluster1_warmup """select count(*) from ${tableName}"""
+
+ def case2Synced0 = getBvar(be0,
"cloud_sync_tablet_meta_synced_total")
+ def case2Synced1 = getBvar(be1,
"cloud_sync_tablet_meta_synced_total")
+
+ useCluster.call(cluster0)
+ sql """alter table ${tableName} set ("compaction_policy" =
"size_based")"""
+
+ assertTrue(waitForBvarIncrease(be0,
"cloud_sync_tablet_meta_synced_total", case2Synced0, 1) >= case2Synced0 + 1)
+ assertTrue(waitForBvarIncrease(be1,
"cloud_sync_tablet_meta_synced_total", case2Synced1, 1) >= case2Synced1 + 1)
+
+ useCluster.call(cluster0)
+ assertEquals("size_based", getCompactionPolicy(tableName))
+ useCluster.call(cluster1)
+ assertEquals("size_based", getCompactionPolicy(tableName))
+
+ // Case 3: hit the size_based version limit first, then alter to
time_series and verify writes succeed immediately.
+ useCluster.call(cluster0)
+ sql """drop table if exists ${versionLimitTableName} force"""
+ sql """
+ CREATE TABLE ${versionLimitTableName} (
+ id BIGINT NOT NULL,
+ value BIGINT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "disable_auto_compaction" = "true"
+ )
+ """
+ qt_case3_version_limit_warmup """select count(*) from
${versionLimitTableName}"""
+ assertEquals("size_based",
getCompactionPolicy(versionLimitTableName))
+
+ int successfulBeforeAlter = 0
+ String versionLimitFailure = null
+ for (int i = 0; i < 2400; i++) {
+ try {
+ sql """insert into ${versionLimitTableName} values (${i},
${i})"""
+ successfulBeforeAlter++
+ if ((i + 1) % 200 == 0) {
+ logger.info("inserted {} versions into {}", i + 1,
versionLimitTableName)
+ }
+ } catch (Exception e) {
+ versionLimitFailure = e.getMessage()
+ logger.info("hit version limit after {} successful
inserts, msg={}",
+ successfulBeforeAlter, versionLimitFailure)
+ break
+ }
+ }
+ assertTrue(versionLimitFailure != null, "expected version limit
failure before alter")
+ assertTrue(versionLimitFailure.contains("-235")
+ || versionLimitFailure.toLowerCase().contains("too many
versions")
+ || versionLimitFailure.toLowerCase().contains("version
count")
+ || versionLimitFailure.toLowerCase().contains("exceed
limit"),
+ "unexpected failure message: ${versionLimitFailure}")
+
+ def versionLimitTablets = sql_return_maparray("""show tablets from
${versionLimitTableName};""")
+ assertEquals(1, versionLimitTablets.size())
+ long versionCountBeforeAlter =
versionLimitTablets[0].VersionCount.toLong()
+ assertTrue(versionCountBeforeAlter >= 2000,
+ "expected version count near max limit,
actual=${versionCountBeforeAlter}")
+
+ sql """alter table ${versionLimitTableName} set
("compaction_policy" = "time_series")"""
+ assertEquals("time_series",
getCompactionPolicy(versionLimitTableName))
+
+ int successfulAfterAlter = 0
+ for (int i = 0; i < 50; i++) {
+ long rowId = 100000L + i
+ sql """insert into ${versionLimitTableName} values (${rowId},
${rowId})"""
+ successfulAfterAlter++
+ }
+ assertEquals(50, successfulAfterAlter)
+ def versionLimitTabletsAfterAlter = sql_return_maparray("""show
tablets from ${versionLimitTableName};""")
+ assertEquals(1, versionLimitTabletsAfterAlter.size())
+ assertTrue(versionLimitTabletsAfterAlter[0].VersionCount.toLong()
> versionCountBeforeAlter,
+ "expected version count to keep increasing immediately
after alter")
+ qt_case3_version_limit_after_alter """select count(*) from
${versionLimitTableName}"""
+
+ // Case 4: disable FE proactive notification and verify cached
tablets remain stale.
+
GetDebugPoint().enableDebugPointForAllFEs("CloudSchemaChangeHandler.notifyBackendsToSyncTabletMeta.skip")
+ def case4Request0 = getBvar(be0,
"cloud_sync_tablet_meta_requests_total")
+ def case4Request1 = getBvar(be1,
"cloud_sync_tablet_meta_requests_total")
+ def case4Synced0 = getBvar(be0,
"cloud_sync_tablet_meta_synced_total")
+ def case4Synced1 = getBvar(be1,
"cloud_sync_tablet_meta_synced_total")
+ def case4Skipped0 = getBvar(be0,
"cloud_sync_tablet_meta_skipped_total")
+ def case4Skipped1 = getBvar(be1,
"cloud_sync_tablet_meta_skipped_total")
+
+ useCluster.call(cluster0)
+ sql """alter table ${tableName} set ("compaction_policy" =
"time_series")"""
+ sleep(2000)
+
+ assertEquals(case4Request0, getBvar(be0,
"cloud_sync_tablet_meta_requests_total"))
+ assertEquals(case4Request1, getBvar(be1,
"cloud_sync_tablet_meta_requests_total"))
+ assertEquals(case4Synced0, getBvar(be0,
"cloud_sync_tablet_meta_synced_total"))
+ assertEquals(case4Synced1, getBvar(be1,
"cloud_sync_tablet_meta_synced_total"))
+ assertEquals(case4Skipped0, getBvar(be0,
"cloud_sync_tablet_meta_skipped_total"))
+ assertEquals(case4Skipped1, getBvar(be1,
"cloud_sync_tablet_meta_skipped_total"))
+
+ useCluster.call(cluster0)
+ assertEquals("size_based", getCompactionPolicy(tableName))
+ useCluster.call(cluster1)
+ assertEquals("size_based", getCompactionPolicy(tableName))
+ } finally {
+ GetDebugPoint().clearDebugPointsForAllFEs()
+ GetDebugPoint().clearDebugPointsForAllBEs()
+ try {
+ sql """use @${cluster0}"""
+ sql """drop table if exists ${dbName}.${tableName} force"""
+ sql """drop table if exists ${dbName}.${versionLimitTableName}
force"""
+ } catch (Exception e) {
+ logger.info("drop table in finally failed: ${e.getMessage()}")
+ }
+ cleanupClusters.call()
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
