github-actions[bot] commented on code in PR #66238:
URL: https://github.com/apache/doris/pull/66238#discussion_r3712154500
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java:
##########
@@ -256,8 +256,13 @@ public void updateOffset(Offset offset) {
} else {
synchronized (splitsLock) {
BinlogSplit binlogSplit = (BinlogSplit)
newOffset.getSplits().get(0);
+ if (MapUtils.isEmpty(binlogSplit.getStartingOffset())) {
+ log.warn("Skip empty committed binlog offset for job {}",
getJobId());
+ return;
+ }
binlogOffsetPersist = new
HashMap<>(binlogSplit.getStartingOffset());
binlogOffsetPersist.put(SPLIT_ID, BinlogSplit.BINLOG_SPLIT_ID);
+ clearSnapshotState();
Review Comment:
This cleanup happens before `resetCloudProgress()` is known to have
succeeded. For a PAUSED cloud snapshot job, `modifyPropertiesInternal()` calls
`updateOffset()` here, which now drops the remaining/finished splits, high
watermarks, and split progress; the following meta-service reset can return
non-OK or throw. `JobManager` then skips its update journal, but the live
provider has already switched to binlog state. A same-master RESUME can
therefore skip unfinished snapshot ranges even though ALTER returned an error,
while failover restores the old state. Please stage or roll back the provider
mutation until the cloud reset succeeds, and cover an injected reset failure.
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java:
##########
@@ -256,8 +256,13 @@ public void updateOffset(Offset offset) {
} else {
synchronized (splitsLock) {
BinlogSplit binlogSplit = (BinlogSplit)
newOffset.getSplits().get(0);
+ if (MapUtils.isEmpty(binlogSplit.getStartingOffset())) {
Review Comment:
The caller cannot tell that this offset was rejected. ALTER validation
accepts any JSON object, so `{}` reaches this return, but
`modifyPropertiesInternal()` still continues through cloud reset/property
update/journaling and reports success without installing the local position.
Conversely, a non-empty source-invalid map such as PostgreSQL `{"foo":"bar"}`
falls through and clears all snapshot recovery state even though the reader
later rejects it for missing `lsn` (MySQL similarly requires file+position or
GTIDs). Please perform source-specific offset validation before mutation and
fail ALTER instead of silently returning or destructively accepting an unusable
map.
##########
regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_snapshot_finished_restart_fe.groovy:
##########
@@ -0,0 +1,160 @@
+// 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_snapshot_finished_restart_fe",
+ "docker,mysql,external_docker,external_docker_mysql,nondatalake") {
+ def jobName = "test_streaming_mysql_job_snapshot_finished_restart_fe"
+ def tableName = "snapshot_finished_restart_fe"
+ def mysqlDb = "test_cdc_db"
+ def totalRows = 5
+ def options = new ClusterOptions()
+ options.setFeNum(1)
+ options.cloudMode = null
+
+ docker(options) {
+ def currentDb = (sql "select database()")[0][0]
+
+ sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
+ sql """DROP TABLE IF EXISTS ${currentDb}.${tableName} FORCE"""
+
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String mysqlPort = context.config.otherConfigs.get("mysql_57_port")
+ String externalEnvIp =
context.config.otherConfigs.get("externalEnvIp")
+ String s3Endpoint = getS3Endpoint()
+ String bucket = getS3BucketName()
+ String driverUrl =
+
"https://${bucket}.${s3Endpoint}/regression/jdbc_driver/mysql-connector-j-8.4.0.jar";
+
+ connect("root", "123456",
"jdbc:mysql://${externalEnvIp}:${mysqlPort}") {
+ sql """CREATE DATABASE IF NOT EXISTS ${mysqlDb}"""
+ sql """DROP TABLE IF EXISTS ${mysqlDb}.${tableName}"""
+ sql """CREATE TABLE ${mysqlDb}.${tableName} (
+ `id` int NOT NULL,
+ `name` varchar(200),
+ PRIMARY KEY (`id`)
+ ) ENGINE=InnoDB"""
+ sql """INSERT INTO ${mysqlDb}.${tableName} (id, name) VALUES
+ (1, 'name_1'),
+ (2, 'name_2'),
+ (3, 'name_3'),
+ (4, 'name_4'),
+ (5, 'name_5')"""
+ }
+
+ sql """CREATE JOB ${jobName}
+ ON STREAMING
+ FROM MYSQL (
+ "jdbc_url" =
"jdbc:mysql://${externalEnvIp}:${mysqlPort}",
+ "driver_url" = "${driverUrl}",
+ "driver_class" = "com.mysql.cj.jdbc.Driver",
+ "user" = "root",
+ "password" = "123456",
+ "database" = "${mysqlDb}",
+ "include_tables" = "${tableName}",
+ "offset" = "snapshot",
+ "snapshot_split_size" = "1",
+ "snapshot_parallelism" = "1"
+ )
+ TO DATABASE ${currentDb} (
+ "table.create.properties.replication_num" = "1"
+ )
+ """
+
+ try {
+ Awaitility.await().atMost(300, SECONDS)
+ .pollInterval(2, SECONDS).until(
+ {
+ def jobStatus = sql """
+ SELECT Status
+ FROM jobs("type"="insert")
+ WHERE Name='${jobName}' AND
ExecuteType='STREAMING'
+ """
+ log.info("jobStatus before FE restart: " +
jobStatus)
+ jobStatus.size() == 1 && jobStatus.get(0).get(0)
== "FINISHED"
Review Comment:
This wait makes the only FE restart occur after the live job is already
`FINISHED`, so the test covers replay of the `onStreamTaskSuccess()` terminal
journal but never the other changed branch in
`StreamingJobSchedulerTask.handlePendingState()`. The recovery case is a
non-final/PENDING job whose provider is already at end; the scheduler marks it
`FINISHED` and the newly added journal must itself survive a later replay.
Removing that scheduler journal leaves both this regression and
`testNaturalFinishPersistsFinalState` green. Please add a scheduler-level
replay test, or restart in that recovered-PENDING window and verify the
scheduler-produced terminal update across another restart.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
