branch-4.1: [improvement](fe) Reduce cloud version sync config (#66296)

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -3894,7 +3894,7 @@ public void handle(Field field, String value) throws Exception {
 
     @ConfField(description = {"存算分离模式下同步 table 和 partition version 的间隔. 所有 frontend 都会检查",
             "Cloud table and partition version syncer interval. All frontends will perform the checking"})
-    public static int cloud_version_syncer_interval_second = 20;
+    public static int cloud_version_syncer_interval_second = 60;
 
     @ConfField(mutable = true, description = {"存算分离模式下是否启用同步 table 和 partition version 的功能",
             "Whether to enable the function of syncing table and partition version in cloud mode"})
@@ -3909,7 +3909,10 @@ public void handle(Field field, String value) throws Exception {
 
     @ConfField(mutable = true, description = {"Get version task 包含的 table 或 partition 数目的 batch size",
             "Maximal table or partition batch size of get version task."})
-    public static int cloud_get_version_task_batch_size = 2000;
+    public static int cloud_get_version_task_batch_size = 200;
+
+    @ConfField(mutable = true, description = {"Maximum retry times for cloud version syncer get version tasks."})
+    public static int cloud_version_syncer_get_version_retry_times = 3;
 
     @ConfField(mutable = true, description = {"schema change job 失败是否重试",
             "Whether to enable retry when a schema change job fails, default is true."})

fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java

@@ -3545,6 +3545,10 @@ private static List<Long> getVisibleVersionInBatchFromMs(List<OlapTable> tables)
     }
 
     public static List<Long> getVisibleVersionFromMeta(List<Long> dbIds, List<Long> tableIds) {
+        return getVisibleVersionFromMeta(dbIds, tableIds, Config.metaServiceRpcRetryTimes());
+    }
+
+    public static List<Long> getVisibleVersionFromMeta(List<Long> dbIds, List<Long> tableIds, int maxAttempts) {
         // get version rpc
         Cloud.GetVersionRequest request = Cloud.GetVersionRequest.newBuilder()
                 .setRequestIp(FrontendOptions.getLocalHostAddressCached())
@@ -3558,7 +3562,7 @@ public static List<Long> getVisibleVersionFromMeta(List<Long> dbIds, List<Long>
                 .build();
 
         try {
-            Cloud.GetVersionResponse resp = VersionHelper.getVersionFromMeta(request);
+            Cloud.GetVersionResponse resp = VersionHelper.getVersionFromMeta(request, maxAttempts);
             if (resp.getStatus().getCode() != Cloud.MetaServiceCode.OK) {
                 throw new RpcException("get table visible version", "unexpected status " + resp.getStatus());
             }

fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudPartition.java

@@ -228,6 +228,12 @@ public static List<Long> selectNonEmptyPartitionIds(List<CloudPartition> partiti
     // Return the visible version in order of the specified partition ids
     public static List<Long> getSnapshotVisibleVersionFromMs(
             List<CloudPartition> partitions, boolean waitForPendingTxns) throws RpcException {
+        return getSnapshotVisibleVersionFromMs(
+                partitions, waitForPendingTxns, Config.metaServiceRpcRetryTimes());
+    }
+
+    public static List<Long> getSnapshotVisibleVersionFromMs(
+            List<CloudPartition> partitions, boolean waitForPendingTxns, int maxAttempts) throws RpcException {
         if (partitions.isEmpty()) {
             return new ArrayList<>();
         }
@@ -243,7 +249,7 @@ public static List<Long> getSnapshotVisibleVersionFromMs(
         }
 
         List<Long> versions = getSnapshotVisibleVersion(
-                dbIds, tableIds, partitionIds, versionUpdateTimesMs, waitForPendingTxns);
+                dbIds, tableIds, partitionIds, versionUpdateTimesMs, waitForPendingTxns, maxAttempts);
 
         // Cache visible version, see hasData() for details.
         int size = versions.size();
@@ -290,8 +296,10 @@ public static List<Long> getSnapshotVisibleVersion(List<CloudPartition> partitio
             return Collections.emptyList();
         }
 
-        long cloudPartitionVersionCacheTtlMs = ConnectContext.get() == null ? 0
-                : ConnectContext.get().getSessionVariable().cloudPartitionVersionCacheTtlMs;
+        ConnectContext ctx = ConnectContext.get();
+        long cloudPartitionVersionCacheTtlMs = ctx == null
+                ? VariableMgr.getDefaultSessionVariable().cloudPartitionVersionCacheTtlMs
+                : ctx.getSessionVariable().cloudPartitionVersionCacheTtlMs;
         if (cloudPartitionVersionCacheTtlMs <= 0) { // No cached versions will be used
             return getSnapshotVisibleVersionFromMs(partitions, false);
         }
@@ -345,7 +353,7 @@ public static List<Long> getSnapshotVisibleVersion(List<CloudPartition> partitio
     //
     // Return the visible version in order of the specified partition ids
     private static List<Long> getSnapshotVisibleVersion(List<Long> dbIds, List<Long> tableIds, List<Long> partitionIds,
-            List<Long> versionUpdateTimesMs, boolean waitForPendingTxns)
+            List<Long> versionUpdateTimesMs, boolean waitForPendingTxns, int maxAttempts)
             throws RpcException {
         assert dbIds.size() == partitionIds.size() :
                 "partition ids size: " + partitionIds.size() + " should equals to db ids size: " + dbIds.size();
@@ -367,7 +375,7 @@ private static List<Long> getSnapshotVisibleVersion(List<Long> dbIds, List<Long>
         if (LOG.isDebugEnabled()) {
             LOG.debug("getVisibleVersion use CloudPartition {}", partitionIds.toString());
         }
-        Cloud.GetVersionResponse resp = VersionHelper.getVersionFromMeta(req);
+        Cloud.GetVersionResponse resp = VersionHelper.getVersionFromMeta(req, maxAttempts);
         if (resp.getStatus().getCode() != MetaServiceCode.OK) {
             throw new RpcException("get visible version", "unexpected status " + resp.getStatus());
         }

fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudSyncVersionDaemon.java

@@ -24,6 +24,7 @@
 import org.apache.doris.catalog.Table;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.util.MasterDaemon;
+import org.apache.doris.qe.VariableMgr;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -58,6 +59,13 @@ protected void runAfterCatalogReady() {
         if (!Config.cloud_enable_version_syncer) {
             return;
         }
+        // This daemon has no ConnectContext, so use the global/default TTLs to decide whether
+        // the shared version caches need proactive refresh. Finite TTLs refresh lazily on reads,
+        // while Long.MAX_VALUE never expires and requires this daemon to keep the cache current.
+        if (VariableMgr.getDefaultSessionVariable().cloudPartitionVersionCacheTtlMs != Long.MAX_VALUE
+                && VariableMgr.getDefaultSessionVariable().cloudTableVersionCacheTtlMs != Long.MAX_VALUE) {
+            return;
+        }
         LOG.info("begin sync cloud table and partition version");
         Map<OlapTable, Long> tableVersionMap = syncTableVersions();
         if (!tableVersionMap.isEmpty()) {
@@ -121,7 +129,8 @@ private Future<Void> submitGetTableVersionTask(Map<OlapTable, Long> tableVersion
             List<Long> tableIds, List<OlapTable> tables) {
         return GET_VERSION_THREAD_POOL.submit(() -> {
             try {
-                List<Long> versions = OlapTable.getVisibleVersionFromMeta(dbIds, tableIds);
+                List<Long> versions = OlapTable.getVisibleVersionFromMeta(
+                        dbIds, tableIds, Config.cloud_version_syncer_get_version_retry_times);
                 for (int i = 0; i < tables.size(); i++) {
                     OlapTable table = tables.get(i);
                     long version = versions.get(i);
@@ -190,7 +199,8 @@ private void syncPartitionVersion(Map<OlapTable, Long> tableVersionMap) {
     private Future<Void> submitGetPartitionVersionTask(Set<Long> failedTables, List<CloudPartition> partitions) {
         return GET_VERSION_THREAD_POOL.submit(() -> {
             try {
-                CloudPartition.getSnapshotVisibleVersionFromMs(partitions, false);
+                CloudPartition.getSnapshotVisibleVersionFromMs(
+                        partitions, false, Config.cloud_version_syncer_get_version_retry_times);
             } catch (Exception e) {
                 LOG.warn("get partition version error", e);
                 Set<Long> failedTableIds = partitions.stream().map(p -> p.getTableId())

fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/VersionHelper.java

@@ -38,10 +38,15 @@ public class VersionHelper {
     // Call get_version() from meta service, and save the elapsed to summary profile.
     public static Cloud.GetVersionResponse getVersionFromMeta(Cloud.GetVersionRequest req)
             throws RpcException {
+        return getVersionFromMeta(req, Config.metaServiceRpcRetryTimes());
+    }
+
+    public static Cloud.GetVersionResponse getVersionFromMeta(Cloud.GetVersionRequest req, int maxAttempts)
+            throws RpcException {
         long startAt = System.nanoTime();
         boolean isTableVersion = req.getIsTableVersion();
         try {
-            return getVisibleVersion(req);
+            return getVisibleVersion(req, maxAttempts);
         } finally {
             SummaryProfile profile = getSummaryProfile();
             if (profile != null) {
@@ -56,8 +61,13 @@ public static Cloud.GetVersionResponse getVersionFromMeta(Cloud.GetVersionReques
     }
 
     public static Cloud.GetVersionResponse getVisibleVersion(Cloud.GetVersionRequest request) throws RpcException {
+        return getVisibleVersion(request, Config.metaServiceRpcRetryTimes());
+    }
+
+    public static Cloud.GetVersionResponse getVisibleVersion(Cloud.GetVersionRequest request, int maxAttempts)
+            throws RpcException {
         int tryTimes = 0;
-        while (tryTimes++ < Config.metaServiceRpcRetryTimes()) {
+        while (tryTimes++ < maxAttempts) {
             Cloud.GetVersionResponse resp = getVisibleVersionInternal(request,
                     Config.default_get_version_from_ms_timeout_second * 1000);
             if (resp != null) {
@@ -73,14 +83,16 @@ public static Cloud.GetVersionResponse getVisibleVersion(Cloud.GetVersionRequest
                         resp.getStatus(), tryTimes);
             }
             // sleep random millis, retry rpc failed
-            if (tryTimes > Config.metaServiceRpcRetryTimes() / 2) {
-                sleepSeveralMs(500, 1000);
-            } else {
-                sleepSeveralMs(20, 200);
+            if (tryTimes < maxAttempts) {
+                if (tryTimes > maxAttempts / 2) {
+                    sleepSeveralMs(500, 1000);
+                } else {
+                    sleepSeveralMs(20, 200);
+                }
             }
         }
 
-        LOG.warn("get version from meta service failed after retry {} times", tryTimes);
+        LOG.warn("get version from meta service failed after retry {} times", maxAttempts);
         throw new RpcException("get version from meta service", "failed after retry n times");
     }
 

fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableTest.java

@@ -637,7 +637,7 @@ public boolean isCloudMode() {
 
         new MockUp<VersionHelper>() {
             @Mock
-            public Cloud.GetVersionResponse getVersionFromMeta(Cloud.GetVersionRequest req) {
+            public Cloud.GetVersionResponse getVersionFromMeta(Cloud.GetVersionRequest req, int maxAttempts) {
                 Cloud.GetVersionResponse.Builder builder = Cloud.GetVersionResponse.newBuilder();
                 builder.setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
                         .setCode(Cloud.MetaServiceCode.OK).build());

fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudPartitionTest.java

@@ -21,13 +21,16 @@
 import org.apache.doris.cloud.rpc.VersionHelper;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.qe.VariableMgr;
 import org.apache.doris.rpc.RpcException;
 
 import mockit.Mock;
 import mockit.MockUp;
 import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.jupiter.api.Assertions;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -72,6 +75,29 @@ public void testIsCachedVersionExpired() {
 
     }
 
+    @Test
+    public void testSnapshotVisibleVersionUsesDefaultCacheTtlWithoutConnectContext() throws RpcException {
+        ConnectContext.remove();
+        SessionVariable defaultSessionVariable = VariableMgr.getDefaultSessionVariable();
+        long originalCacheTtlMs = defaultSessionVariable.cloudPartitionVersionCacheTtlMs;
+        try {
+            defaultSessionVariable.cloudPartitionVersionCacheTtlMs = Long.MAX_VALUE;
+            CloudPartition cachedPartition = createPartition(1, 2, 3);
+            cachedPartition.setCachedVisibleVersion(2, 10086L);
+
+            try (MockedStatic<VersionHelper> mockedVersionHelper = Mockito.mockStatic(VersionHelper.class)) {
+                List<Long> versions = CloudPartition.getSnapshotVisibleVersion(
+                        Arrays.asList(cachedPartition));
+
+                Assertions.assertEquals(Arrays.asList(2L), versions);
+                mockedVersionHelper.verifyNoInteractions();
+            }
+        } finally {
+            defaultSessionVariable.cloudPartitionVersionCacheTtlMs = originalCacheTtlMs;
+            ConnectContext.remove();
+        }
+    }
+
     @Test
     public void testCachedVersion() throws RpcException {
         // Create ConnectContext with SessionVariable
@@ -98,7 +124,7 @@ public void testCachedVersion() throws RpcException {
 
         new MockUp<VersionHelper>(VersionHelper.class) {
             @Mock
-            public Cloud.GetVersionResponse getVersionFromMeta(Cloud.GetVersionRequest req) {
+            public Cloud.GetVersionResponse getVersionFromMeta(Cloud.GetVersionRequest req, int maxAttempts) {
                 Cloud.GetVersionResponse.Builder builder = Cloud.GetVersionResponse.newBuilder();
                 builder.setVersion(singleVersions.get(callCount[0]));
                 builder.addAllVersions(batchVersions.get(callCount[0]));

fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/VersionHelperTest.java

@@ -0,0 +1,70 @@
+// 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.rpc;
+
+import org.apache.doris.cloud.proto.Cloud;
+import org.apache.doris.rpc.RpcException;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.concurrent.CompletableFuture;
+
+public class VersionHelperTest {
+    @Test
+    public void testGetVisibleVersionUsesSpecifiedMaxAttempts() throws RpcException {
+        Cloud.GetVersionRequest request = Cloud.GetVersionRequest.newBuilder().build();
+        Cloud.GetVersionResponse failedResponse = Cloud.GetVersionResponse.newBuilder()
+                .setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
+                        .setCode(Cloud.MetaServiceCode.KV_TXN_GET_ERR))
+                .build();
+        MetaServiceProxy proxy = Mockito.mock(MetaServiceProxy.class);
+        Mockito.when(proxy.getVisibleVersionAsync(request))
+                .thenReturn(CompletableFuture.completedFuture(failedResponse));
+
+        try (MockedStatic<MetaServiceProxy> mockedProxy = Mockito.mockStatic(MetaServiceProxy.class)) {
+            mockedProxy.when(MetaServiceProxy::getInstance).thenReturn(proxy);
+
+            Assert.assertThrows(RpcException.class, () -> VersionHelper.getVisibleVersion(request, 3));
+        }
+
+        Mockito.verify(proxy, Mockito.times(3)).getVisibleVersionAsync(request);
+    }
+
+    @Test
+    public void testGetVisibleVersionStopsOnVersionNotFound() throws RpcException {
+        Cloud.GetVersionRequest request = Cloud.GetVersionRequest.newBuilder().build();
+        Cloud.GetVersionResponse notFoundResponse = Cloud.GetVersionResponse.newBuilder()
+                .setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
+                        .setCode(Cloud.MetaServiceCode.VERSION_NOT_FOUND))
+                .build();
+        MetaServiceProxy proxy = Mockito.mock(MetaServiceProxy.class);
+        Mockito.when(proxy.getVisibleVersionAsync(request))
+                .thenReturn(CompletableFuture.completedFuture(notFoundResponse));
+
+        try (MockedStatic<MetaServiceProxy> mockedProxy = Mockito.mockStatic(MetaServiceProxy.class)) {
+            mockedProxy.when(MetaServiceProxy::getInstance).thenReturn(proxy);
+
+            Assert.assertSame(notFoundResponse, VersionHelper.getVisibleVersion(request, 3));
+        }
+
+        Mockito.verify(proxy).getVisibleVersionAsync(request);
+    }
+}