This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 6f9ca8c5945 branch-4.1: [fix](cloud) Trigger periodic cloud
checkpoints through edit log rolls (#66154) (#67422)
6f9ca8c5945 is described below
commit 6f9ca8c5945da8f734eeaab19adadc7c45892936
Author: meiyi <[email protected]>
AuthorDate: Sun Sep 6 15:48:11 2026 +0800
branch-4.1: [fix](cloud) Trigger periodic cloud checkpoints through edit
log rolls (#66154) (#67422)
pick https://github.com/apache/doris/pull/66154
---
.../main/java/org/apache/doris/common/Config.java | 5 +
.../java/org/apache/doris/master/Checkpoint.java | 8 --
.../java/org/apache/doris/persist/EditLog.java | 24 +++-
.../java/org/apache/doris/persist/EditLogTest.java | 159 +++++++++++++++++++++
.../test_cloud_edit_log_roll_checkpoint.groovy | 65 +++++++++
5 files changed, 246 insertions(+), 15 deletions(-)
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index e506ac318b0..d8becec686d 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -264,6 +264,11 @@ public class Config extends ConfigBase {
"The log roll size of BDBJE. When the number of log entries
exceeds this value, the log will be rolled"})
public static int edit_log_roll_num = 50000;
+ @ConfField(mutable = true, masterOnly = true, description = {
+ "The maximum interval in seconds between edit log rolls in cloud
mode. "
+ + "A non-positive value disables time-based edit log
rolling"})
+ public static int cloud_edit_log_roll_interval_second = 3600;
+
@ConfField(mutable = true, masterOnly = true, description = {
"批量 BDBJE 日志包含的最大条目数", "The max number of log entries for batching
BDBJE"})
public static int batch_edit_log_max_item_num = 100;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/master/Checkpoint.java
b/fe/fe-core/src/main/java/org/apache/doris/master/Checkpoint.java
index b908954541d..1fe717d59ff 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/master/Checkpoint.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/master/Checkpoint.java
@@ -121,14 +121,6 @@ public class Checkpoint extends MasterDaemon {
if (imageVersion < checkPointVersion) {
LOG.info("Trigger checkpoint since last checkpoint journal id:
{} is less than "
+ "current finalized journal id: {}", imageVersion,
checkPointVersion);
- } else if (Config.isCloudMode() &&
Config.cloud_checkpoint_image_stale_threshold_seconds > 0
- && latestImageCreateTime > 0 &&
((System.currentTimeMillis() - latestImageCreateTime)
- >= Config.cloud_checkpoint_image_stale_threshold_seconds *
1000L)) {
- // No new finalized journals beyond the latest image.
- // But in cloud mode, we may still want to force a checkpoint
if the latest image file is expired.
- // This helps that image can keep the newer table version,
partition version, tablet stats.
- LOG.info("Trigger checkpoint in cloud mode because latest
image is expired. "
- + "latestImageSeq: {}, latestImageCreateTime: {}",
imageVersion, latestImageCreateTime);
} else {
return;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
b/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
index bb01746a0ab..f86581d5185 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
@@ -153,7 +153,8 @@ public class EditLog {
private EditLogOutputStream editStream = null;
private long txId = 0;
-
+ // This best-effort timer starts when EditLog is created and resets after
every roll.
+ private volatile long lastEditLogRollTimeMs = System.currentTimeMillis();
private AtomicLong numTransactions = new AtomicLong(0);
private AtomicLong totalTimeTransactions = new AtomicLong(0);
@@ -249,9 +250,10 @@ public class EditLog {
txId += batch.size();
// update statistics, etc. (optional, can be added as needed)
- if (txId >= Config.edit_log_roll_num) {
- LOG.info("txId {} is equal to or larger than edit_log_roll_num {},
will roll edit.", txId,
- Config.edit_log_roll_num);
+ if (txId >= Config.edit_log_roll_num || exceedEditLogRollInterval()) {
+ LOG.info("edit log roll condition met. txId: {}, edit log roll
num: {}, "
+ + "cloud edit log roll interval: {} seconds",
+ txId, Config.edit_log_roll_num,
Config.cloud_edit_log_roll_interval_second);
rollEditLog();
txId = 0;
}
@@ -1513,6 +1515,7 @@ public class EditLog {
*/
public void rollEditLog() {
journal.rollJournal();
+ lastEditLogRollTimeMs = System.currentTimeMillis();
}
// NOTICE: No guarantee atomicity of entries
@@ -1598,9 +1601,10 @@ public class EditLog {
// get a new transactionId
txId++;
- if (txId >= Config.edit_log_roll_num) {
- LOG.info("txId {} is equal to or larger than edit_log_roll_num {},
will roll edit.", txId,
- Config.edit_log_roll_num);
+ if (txId >= Config.edit_log_roll_num || exceedEditLogRollInterval()) {
+ LOG.info("edit log roll condition met. txId: {}, edit log roll
num: {}, "
+ + "cloud edit log roll interval: {} seconds",
+ txId, Config.edit_log_roll_num,
Config.cloud_edit_log_roll_interval_second);
rollEditLog();
txId = 0;
}
@@ -1608,6 +1612,12 @@ public class EditLog {
return logId;
}
+ private boolean exceedEditLogRollInterval() {
+ return Config.isCloudMode() &&
Config.cloud_edit_log_roll_interval_second > 0
+ && System.currentTimeMillis() - lastEditLogRollTimeMs
+ >=
TimeUnit.SECONDS.toMillis(Config.cloud_edit_log_roll_interval_second);
+ }
+
/**
* Write an operation to the edit log. Do not sync to persistent store yet.
*/
diff --git a/fe/fe-core/src/test/java/org/apache/doris/persist/EditLogTest.java
b/fe/fe-core/src/test/java/org/apache/doris/persist/EditLogTest.java
index e1c56810db8..61b4b786d8c 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/persist/EditLogTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/persist/EditLogTest.java
@@ -17,14 +17,58 @@
package org.apache.doris.persist;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.journal.bdbje.Timestamp;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
+import java.util.concurrent.TimeUnit;
public class EditLogTest {
private String meta = "editLogTestDir/";
+ private String originalEditLogType;
+ private int originalEditLogRollNum;
+ private int originalCloudEditLogRollIntervalSecond;
+ private String originalDeployMode;
+ private String originalCloudUniqueId;
+
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Before
+ public void setUpEditLogRollConfig() {
+ originalEditLogType = Config.edit_log_type;
+ originalEditLogRollNum = Config.edit_log_roll_num;
+ originalCloudEditLogRollIntervalSecond =
Config.cloud_edit_log_roll_interval_second;
+ originalDeployMode = Config.deploy_mode;
+ originalCloudUniqueId = Config.cloud_unique_id;
+
+ Config.edit_log_type = "local";
+ Config.edit_log_roll_num = Integer.MAX_VALUE;
+ Config.cloud_edit_log_roll_interval_second = 3600;
+ Config.cloud_unique_id = "";
+ }
+
+ @After
+ public void restoreEditLogRollConfig() {
+ Config.edit_log_type = originalEditLogType;
+ Config.edit_log_roll_num = originalEditLogRollNum;
+ Config.cloud_edit_log_roll_interval_second =
originalCloudEditLogRollIntervalSecond;
+ Config.deploy_mode = originalDeployMode;
+ Config.cloud_unique_id = originalCloudUniqueId;
+ }
public void mkdir() {
File dir = new File(meta);
@@ -102,4 +146,119 @@ public class EditLogTest {
public void test() {
}
+
+ @Test
+ public void testCloudModeTimeBasedEditLogRoll() throws Exception {
+ Config.deploy_mode = "cloud";
+
+ File imageDir = temporaryFolder.newFolder("time_based_roll");
+ Env env = Mockito.mock(Env.class);
+ Mockito.when(env.getImageDir()).thenReturn(imageDir.getAbsolutePath());
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+ EditLog editLog = new EditLog("test");
+ editLog.open();
+ try {
+ Deencapsulation.setField(editLog, "lastEditLogRollTimeMs",
+ System.currentTimeMillis() -
TimeUnit.HOURS.toMillis(2));
+
+ editLog.logTimestamp(new Timestamp());
+
+ Assert.assertTrue(new File(imageDir, "edits.2").exists());
+ long txId = Deencapsulation.getField(editLog, "txId");
+ Assert.assertEquals(0L, txId);
+ } finally {
+ editLog.close();
+ }
+ }
+ }
+
+ @Test
+ public void testNonCloudModeDoesNotRollEditLogByTime() throws Exception {
+ Config.deploy_mode = "share_nothing";
+
+ File imageDir = temporaryFolder.newFolder("non_cloud_time_based_roll");
+ Env env = Mockito.mock(Env.class);
+ Mockito.when(env.getImageDir()).thenReturn(imageDir.getAbsolutePath());
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+ EditLog editLog = new EditLog("test");
+ editLog.open();
+ try {
+ Deencapsulation.setField(editLog, "lastEditLogRollTimeMs",
+ System.currentTimeMillis() -
TimeUnit.HOURS.toMillis(2));
+
+ editLog.logTimestamp(new Timestamp());
+
+ Assert.assertFalse(new File(imageDir, "edits.2").exists());
+
+ Config.edit_log_roll_num = 2;
+ editLog.logTimestamp(new Timestamp());
+
+ Assert.assertTrue(new File(imageDir, "edits.3").exists());
+ } finally {
+ editLog.close();
+ }
+ }
+ }
+
+ @Test
+ public void testRollEditLogResetsCloudRollTime() throws Exception {
+ Config.deploy_mode = "cloud";
+
+ File imageDir = temporaryFolder.newFolder("reset_time_after_roll");
+ Env env = Mockito.mock(Env.class);
+ Mockito.when(env.getImageDir()).thenReturn(imageDir.getAbsolutePath());
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+ EditLog editLog = new EditLog("test");
+ editLog.open();
+ try {
+ editLog.logTimestamp(new Timestamp());
+ Deencapsulation.setField(editLog, "lastEditLogRollTimeMs",
+ System.currentTimeMillis() -
TimeUnit.HOURS.toMillis(2));
+
+ editLog.rollEditLog();
+ editLog.logTimestamp(new Timestamp());
+
+ Assert.assertTrue(new File(imageDir, "edits.2").exists());
+ Assert.assertFalse(new File(imageDir, "edits.3").exists());
+ } finally {
+ editLog.close();
+ }
+ }
+ }
+
+ @Test
+ public void testNonPositiveCloudEditLogRollIntervalDisablesTimeBasedRoll()
throws Exception {
+ Config.deploy_mode = "cloud";
+ int[] disabledIntervals = {0, -1};
+ for (int i = 0; i < disabledIntervals.length; i++) {
+ Config.cloud_edit_log_roll_interval_second = disabledIntervals[i];
+ File imageDir =
temporaryFolder.newFolder("disabled_time_based_roll_" + i);
+ Env env = Mockito.mock(Env.class);
+
Mockito.when(env.getImageDir()).thenReturn(imageDir.getAbsolutePath());
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+ EditLog editLog = new EditLog("test");
+ editLog.open();
+ try {
+ Deencapsulation.setField(editLog, "lastEditLogRollTimeMs",
+ System.currentTimeMillis() -
TimeUnit.HOURS.toMillis(2));
+
+ editLog.logTimestamp(new Timestamp());
+
+ Assert.assertFalse(new File(imageDir, "edits.2").exists());
+
+ Config.edit_log_roll_num = 2;
+ editLog.logTimestamp(new Timestamp());
+
+ Assert.assertTrue(new File(imageDir, "edits.3").exists());
+ Config.edit_log_roll_num = Integer.MAX_VALUE;
+ } finally {
+ editLog.close();
+ }
+ }
+ }
+ }
}
diff --git
a/regression-test/suites/cloud_p0/test_cloud_edit_log_roll_checkpoint.groovy
b/regression-test/suites/cloud_p0/test_cloud_edit_log_roll_checkpoint.groovy
new file mode 100644
index 00000000000..d800cbb872e
--- /dev/null
+++ b/regression-test/suites/cloud_p0/test_cloud_edit_log_roll_checkpoint.groovy
@@ -0,0 +1,65 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+
+suite("test_cloud_edit_log_roll_checkpoint", "docker") {
+ if (!isCloudMode()) {
+ return
+ }
+
+ def options = new ClusterOptions()
+ options.feNum = 1
+ options.beNum = 1
+ options.cloudMode = true
+ options.feConfigs += [
+ "cloud_edit_log_roll_interval_second=30"
+ ]
+
+ docker(options) {
+ def config = sql_return_maparray """
+ ADMIN SHOW FRONTEND CONFIG LIKE
'cloud_edit_log_roll_interval_second'
+ """
+ assertEquals(1, config.size())
+ assertEquals("30", config[0].Value)
+
+ def getCheckpointVersion = {
+ def masterFe = cluster.getMasterFe()
+ def response = parseJson(new URL(
+
"http://${masterFe.host}:${masterFe.httpPort}/api/show_meta_info?action=SHOW_HA";).text)
+ assertEquals(0, response.code)
+ return response.data.last_checkpoint_version as long
+ }
+
+ long initialCheckpointVersion = getCheckpointVersion()
+ logger.info("Initial checkpoint version: {}", initialCheckpointVersion)
+
+ long previousCheckpointVersion = initialCheckpointVersion
+ int checkpointCount = 0
+ awaitUntil(180, 5) {
+ long currentCheckpointVersion = getCheckpointVersion()
+ logger.info("Current checkpoint version: {}",
currentCheckpointVersion)
+ if (currentCheckpointVersion > previousCheckpointVersion) {
+ checkpointCount++
+ previousCheckpointVersion = currentCheckpointVersion
+ logger.info("Observed checkpoint image advancement {},
version: {}",
+ checkpointCount, currentCheckpointVersion)
+ }
+ return checkpointCount >= 2
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
