You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2026/04/15 02:10:04 UTC
(doris) branch branch-4.0 updated: branch-4.0: [fix](streaming-job) fix streaming job properties not parsed after FE restart #62298 (#62431)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 25e25c1a20a branch-4.0: [fix](streaming-job) fix streaming job properties not parsed after FE restart #62298 (#62431)
25e25c1a20a is described below
commit 25e25c1a20a313f2972a4775c494d4cd7b40e6a1
Author: github-actions[bot] <41...@users.noreply.github.com>
AuthorDate: Wed Apr 15 10:09:55 2026 +0800
branch-4.0: [fix](streaming-job) fix streaming job properties not parsed after FE restart #62298 (#62431)
Cherry-picked from #62298
Co-authored-by: wudi <wu...@selectdb.com>
---
.../insert/streaming/StreamingJobProperties.java | 22 +++-
.../insert/streaming/StreamingMultiTblTask.java | 7 +-
.../streaming/StreamingJobPropertiesTest.java | 78 +++++++++++
...treaming_mysql_job_restart_fe_with_props.groovy | 143 +++++++++++++++++++++
4 files changed, 245 insertions(+), 5 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java
index 58e236e6720..daa2cc36921 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java
@@ -24,6 +24,7 @@ import org.apache.doris.job.exception.JobException;
import org.apache.doris.qe.SessionVariable;
import org.apache.doris.qe.VariableMgr;
+import com.google.common.base.Strings;
import lombok.Data;
import org.json.simple.JSONObject;
@@ -59,10 +60,23 @@ public class StreamingJobProperties implements JobProperties {
public StreamingJobProperties(Map<String, String> jobProperties) {
this.properties = jobProperties;
- if (properties.isEmpty()) {
- this.maxIntervalSecond = DEFAULT_MAX_INTERVAL_SECOND;
- this.s3BatchFiles = DEFAULT_MAX_S3_BATCH_FILES;
- this.s3BatchBytes = DEFAULT_MAX_S3_BATCH_BYTES;
+ this.maxIntervalSecond = parseLongOrDefault(
+ properties.get(MAX_INTERVAL_SECOND_PROPERTY), DEFAULT_MAX_INTERVAL_SECOND);
+ this.s3BatchFiles = parseLongOrDefault(
+ properties.get(S3_MAX_BATCH_FILES_PROPERTY), DEFAULT_MAX_S3_BATCH_FILES);
+ this.s3BatchBytes = parseLongOrDefault(
+ properties.get(S3_MAX_BATCH_BYTES_PROPERTY), DEFAULT_MAX_S3_BATCH_BYTES);
+ }
+
+ private static long parseLongOrDefault(String valStr, long defaultVal) {
+ if (Strings.isNullOrEmpty(valStr)) {
+ return defaultVal;
+ }
+ try {
+ long val = Long.parseLong(valStr);
+ return val >= 1 ? val : defaultVal;
+ } catch (NumberFormatException e) {
+ return defaultVal;
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java
index 50330595da6..d905cca9f46 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java
@@ -306,7 +306,12 @@ public class StreamingMultiTblTask extends AbstractStreamingTask {
// It's still pending, waiting for scheduling.
return false;
}
- return (System.currentTimeMillis() - startTimeMs) > timeoutMs;
+ long elapsed = System.currentTimeMillis() - startTimeMs;
+ if (elapsed > timeoutMs) {
+ log.info("Task {} timeout detected: elapsed={}ms, timeoutMs={}ms", taskId, elapsed, timeoutMs);
+ return true;
+ }
+ return false;
}
/**
diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java
index 328059aaf1f..97a98031dde 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java
@@ -17,6 +17,7 @@
package org.apache.doris.job.extensions.insert.streaming;
+import org.apache.doris.common.AnalysisException;
import org.apache.doris.job.exception.JobException;
import org.apache.doris.job.extensions.insert.InsertTask;
import org.apache.doris.qe.ConnectContext;
@@ -29,6 +30,83 @@ import java.util.HashMap;
public class StreamingJobPropertiesTest {
+ /**
+ * Simulate FE restart: constructor is called without validate().
+ * Before the fix, maxIntervalSecond would be 0 when properties is non-empty,
+ * causing streaming tasks to timeout immediately after FE restart.
+ */
+ @Test
+ public void testConstructorParsesPropertiesWithoutValidate() {
+ // Case 1: empty properties -> should use defaults
+ StreamingJobProperties emptyProps = new StreamingJobProperties(new HashMap<>());
+ Assert.assertEquals(StreamingJobProperties.DEFAULT_MAX_INTERVAL_SECOND,
+ emptyProps.getMaxIntervalSecond());
+ Assert.assertEquals(StreamingJobProperties.DEFAULT_MAX_S3_BATCH_FILES,
+ emptyProps.getS3BatchFiles());
+ Assert.assertEquals(StreamingJobProperties.DEFAULT_MAX_S3_BATCH_BYTES,
+ emptyProps.getS3BatchBytes());
+
+ // Case 2: explicit max_interval=1 (the bug scenario)
+ // Before fix: maxIntervalSecond would be 0 because isEmpty()=false skipped defaults
+ HashMap<String, String> props = new HashMap<>();
+ props.put("max_interval", "1");
+ StreamingJobProperties customProps = new StreamingJobProperties(props);
+ Assert.assertEquals(1L, customProps.getMaxIntervalSecond());
+
+ // Case 3: explicit max_interval=5
+ HashMap<String, String> props2 = new HashMap<>();
+ props2.put("max_interval", "5");
+ StreamingJobProperties customProps2 = new StreamingJobProperties(props2);
+ Assert.assertEquals(5L, customProps2.getMaxIntervalSecond());
+ // s3 properties not set -> should use defaults
+ Assert.assertEquals(StreamingJobProperties.DEFAULT_MAX_S3_BATCH_FILES,
+ customProps2.getS3BatchFiles());
+ }
+
+ /**
+ * Constructor should be resilient to bad data (e.g. corrupted metadata),
+ * falling back to defaults instead of throwing exceptions.
+ */
+ @Test
+ public void testConstructorHandlesBadValues() {
+ // non-numeric value -> fallback to default
+ HashMap<String, String> props = new HashMap<>();
+ props.put("max_interval", "abc");
+ StreamingJobProperties p = new StreamingJobProperties(props);
+ Assert.assertEquals(StreamingJobProperties.DEFAULT_MAX_INTERVAL_SECOND,
+ p.getMaxIntervalSecond());
+
+ // zero value -> fallback to default (must be >= 1)
+ HashMap<String, String> props2 = new HashMap<>();
+ props2.put("max_interval", "0");
+ StreamingJobProperties p2 = new StreamingJobProperties(props2);
+ Assert.assertEquals(StreamingJobProperties.DEFAULT_MAX_INTERVAL_SECOND,
+ p2.getMaxIntervalSecond());
+
+ // negative value -> fallback to default
+ HashMap<String, String> props3 = new HashMap<>();
+ props3.put("max_interval", "-1");
+ StreamingJobProperties p3 = new StreamingJobProperties(props3);
+ Assert.assertEquals(StreamingJobProperties.DEFAULT_MAX_INTERVAL_SECOND,
+ p3.getMaxIntervalSecond());
+ }
+
+ /**
+ * validate() should still reject bad values with AnalysisException,
+ * keeping the strict check for job creation.
+ */
+ @Test
+ public void testValidateStillRejectsBadValues() {
+ HashMap<String, String> props = new HashMap<>();
+ props.put("max_interval", "0");
+ StreamingJobProperties p = new StreamingJobProperties(props);
+ // constructor fallback is fine
+ Assert.assertEquals(StreamingJobProperties.DEFAULT_MAX_INTERVAL_SECOND,
+ p.getMaxIntervalSecond());
+ // but validate() should throw
+ Assert.assertThrows(AnalysisException.class, p::validate);
+ }
+
@Test
public void testSessionVariables() throws JobException {
//default
diff --git a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_restart_fe_with_props.groovy b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_restart_fe_with_props.groovy
new file mode 100644
index 00000000000..9fe209b7dd9
--- /dev/null
+++ b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_restart_fe_with_props.groovy
@@ -0,0 +1,143 @@
+// 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
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_streaming_mysql_job_restart_fe_with_props", "docker,mysql,external_docker,external_docker_mysql,nondatalake") {
+ def jobName = "test_streaming_mysql_job_restart_fe_with_props"
+ def options = new ClusterOptions()
+ options.setFeNum(1)
+ options.cloudMode = null
+
+ docker(options) {
+ def currentDb = (sql "select database()")[0][0]
+ def table1 = "restart_props_user_info"
+ def mysqlDb = "test_cdc_db"
+
+ sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
+ sql """drop table if exists ${currentDb}.${table1} force"""
+
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String mysql_port = context.config.otherConfigs.get("mysql_57_port");
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String s3_endpoint = getS3Endpoint()
+ String bucket = getS3BucketName()
+ String driver_url = "https://${bucket}.${s3_endpoint}/regression/jdbc_driver/mysql-connector-j-8.4.0.jar"
+
+ connect("root", "123456", "jdbc:mysql://${externalEnvIp}:${mysql_port}") {
+ sql """CREATE DATABASE IF NOT EXISTS ${mysqlDb}"""
+ sql """DROP TABLE IF EXISTS ${mysqlDb}.${table1}"""
+ sql """CREATE TABLE ${mysqlDb}.${table1} (
+ `name` varchar(200) NOT NULL,
+ `age` int DEFAULT NULL,
+ PRIMARY KEY (`name`)
+ ) ENGINE=InnoDB"""
+ sql """INSERT INTO ${mysqlDb}.${table1} (name, age) VALUES ('A1', 1);"""
+ sql """INSERT INTO ${mysqlDb}.${table1} (name, age) VALUES ('B1', 2);"""
+ }
+
+ // Create job with explicit max_interval property.
+ // Before the fix, after FE restart the max_interval would not be parsed
+ // from properties in the constructor, causing timeoutMs=0 and every task
+ // to timeout immediately.
+ sql """CREATE JOB ${jobName}
+ PROPERTIES("max_interval" = "5")
+ ON STREAMING
+ FROM MYSQL (
+ "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}",
+ "driver_url" = "${driver_url}",
+ "driver_class" = "com.mysql.cj.jdbc.Driver",
+ "user" = "root",
+ "password" = "123456",
+ "database" = "${mysqlDb}",
+ "include_tables" = "${table1}",
+ "offset" = "initial"
+ )
+ TO DATABASE ${currentDb} (
+ "table.create.properties.replication_num" = "1"
+ )
+ """
+
+ // Wait for snapshot data to be loaded
+ try {
+ Awaitility.await().atMost(300, SECONDS)
+ .pollInterval(1, SECONDS).until(
+ {
+ def jobSucceedCount = sql """ select SucceedTaskCount from jobs("type"="insert") where Name = '${jobName}' and ExecuteType='STREAMING' """
+ log.info("jobSucceedCount: " + jobSucceedCount)
+ jobSucceedCount.size() == 1 && '2' <= jobSucceedCount.get(0).get(0)
+ }
+ )
+ } catch (Exception ex) {
+ def showjob = sql """select * from jobs("type"="insert") where Name='${jobName}'"""
+ def showtask = sql """select * from tasks("type"="insert") where JobName='${jobName}'"""
+ log.info("show job: " + showjob)
+ log.info("show task: " + showtask)
+ throw ex;
+ }
+
+ def jobInfoBeforeRestart = sql """
+ select loadStatistic, status, currentOffset from jobs("type"="insert") where Name='${jobName}'
+ """
+ log.info("jobInfoBeforeRestart: " + jobInfoBeforeRestart)
+ def loadStatBefore = parseJson(jobInfoBeforeRestart.get(0).get(0))
+ assert loadStatBefore.scannedRows == 2
+ assert jobInfoBeforeRestart.get(0).get(1) == "RUNNING"
+
+ // Restart FE
+ cluster.restartFrontends()
+ sleep(60000)
+ context.reconnectFe()
+
+ // Insert new data after restart
+ connect("root", "123456", "jdbc:mysql://${externalEnvIp}:${mysql_port}") {
+ sql """INSERT INTO ${mysqlDb}.${table1} (name, age) VALUES ('C1', 3);"""
+ }
+
+ // Verify new data gets consumed after restart.
+ // Before the fix, timeoutMs=0 caused every task to timeout immediately,
+ // so the job would be stuck in PAUSED and never consume new data.
+ try {
+ Awaitility.await().atMost(120, SECONDS)
+ .pollInterval(2, SECONDS).until(
+ {
+ def loadStat = sql """ select loadStatistic from jobs("type"="insert") where Name = '${jobName}' """
+ def stat = parseJson(loadStat.get(0).get(0))
+ log.info("scannedRows after restart: " + stat.scannedRows)
+ stat.scannedRows == 3
+ }
+ )
+ } catch (Exception ex) {
+ def showjob = sql """select * from jobs("type"="insert") where Name='${jobName}'"""
+ def showtask = sql """select * from tasks("type"="insert") where JobName='${jobName}'"""
+ log.info("show job after restart insert: " + showjob)
+ log.info("show task after restart insert: " + showtask)
+ throw ex;
+ }
+
+ def result = sql """select * from ${currentDb}.${table1} order by name"""
+ log.info("final result: " + result)
+ assert result.size() == 3
+
+ sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org