github-actions[bot] commented on code in PR #65588: URL: https://github.com/apache/doris/pull/65588#discussion_r3635395617
########## regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_restart_fe.groovy: ########## @@ -0,0 +1,150 @@ +// 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_oceanbase_job_restart_fe", + "p2,docker,oceanbase,external_docker,external_docker_oceanbase,nondatalake") { + def jobName = "test_streaming_oceanbase_job_restart_fe" + def sourceDb = "test_oceanbase_streaming_db" + def table1 = "oceanbase_streaming_restart_fe" + 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}.${table1} FORCE""" + + String enabled = context.config.otherConfigs.get("enableJdbcTest") + if (enabled != null && enabled.equalsIgnoreCase("true")) { + String oceanbaseCdcPort = context.config.otherConfigs.get("oceanbase_cdc_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"; + String sourceUrl = "jdbc:mysql://${externalEnvIp}:${oceanbaseCdcPort}" + + def waitForValue = { int id, String expected -> + Awaitility.await().atMost(300, SECONDS).pollInterval(2, SECONDS).until({ + def rows = sql "SELECT value FROM ${currentDb}.${table1} WHERE id=${id}" + rows.size() == 1 && rows[0][0] == expected + }) + } + def waitForJobAfterRestart = { + Awaitility.await().atMost(300, SECONDS).pollInterval(2, SECONDS) + .ignoreExceptions().until({ + context.reconnectFe() + def rows = sql """SELECT Status FROM jobs("type"="insert") + WHERE Name='${jobName}'""" + rows.size() == 1 && rows[0][0] == "RUNNING" + }) + } + def dumpJobState = { + log.info("jobs: " + sql("""SELECT * FROM jobs("type"="insert") WHERE Name='${jobName}'""")) + log.info("tasks: " + sql("""SELECT * FROM tasks("type"="insert") WHERE JobName='${jobName}'""")) + } + + connect("root@test", "123456", sourceUrl) { + sql """CREATE DATABASE IF NOT EXISTS ${sourceDb}""" + sql """DROP TABLE IF EXISTS ${sourceDb}.${table1}""" + sql """CREATE TABLE ${sourceDb}.${table1} ( + id INT NOT NULL, + value VARCHAR(100), + PRIMARY KEY (id) + ) ENGINE=InnoDB""" + sql """INSERT INTO ${sourceDb}.${table1} VALUES (1, 'snapshot')""" + } + + sql """CREATE JOB ${jobName} + ON STREAMING + FROM OCEANBASE ( + "jdbc_url" = "${sourceUrl}", + "driver_url" = "${driverUrl}", + "driver_class" = "com.mysql.cj.jdbc.Driver", + "user" = "root@test", + "password" = "123456", + "database" = "${sourceDb}", + "include_tables" = "${table1}", + "offset" = "initial" + ) + TO DATABASE ${currentDb} ( + "table.create.properties.replication_num" = "1" + )""" + + try { + waitForValue(1, "snapshot") + connect("root@test", "123456", sourceUrl) { + sql """INSERT INTO ${sourceDb}.${table1} VALUES (2, 'before_restart')""" + } + waitForValue(2, "before_restart") + } catch (Exception ex) { + dumpJobState() + throw ex + } + + def beforeOffset = null + Awaitility.await().atMost(300, SECONDS).pollInterval(2, SECONDS).until({ + def rows = sql """SELECT currentOffset FROM jobs("type"="insert") + WHERE Name='${jobName}'""" + if (rows.size() != 1 || rows[0][0] == null) { + return false + } + def parsed = parseJson(rows[0][0]) + if (parsed.file == null || parsed.pos == null) { + return false + } + beforeOffset = parsed + return true + }) + + cluster.restartFrontends() Review Comment: [P2] Make this test force a persisted-position resume. `restartFrontends()` leaves the BE CDC reader alive, and every source mutation happens either before the restart or after the job is RUNNING. A replay that drops the checkpoint can therefore be repopulated by the surviving reader or restart from latest and still consume row 3/update; a rewind is hidden by UNIQUE KEY upserts, while `afterOffset >= beforeOffset` accepts a jump. Add an identifiable source event during FE downtime, force reader rebind/rebuild, and assert it is delivered exactly once (or otherwise assert the exact persisted resume position). ########## docker/thirdparties/docker-compose/oceanbase/oceanbase.yaml.tpl: ########## @@ -19,16 +19,18 @@ version: "2.1" services: doris--oceanbase: - image: oceanbase/oceanbase-ce:4.2.1-lts + image: quay.io/oceanbase/obbinlog-ce:4.2.5-test Review Comment: [P2] Update the resource configuration together with this image switch. `obbinlog-ce` does not read the retained `MODE` or `OB_MEMORY_LIMIT` variables; its entrypoint starts Observer with a hard-coded `memory_limit=6G` and `system_memory=1G`. This compose service therefore no longer has the declared slim/5-GiB footprint and can exceed the capacity of regression hosts sized for the old profile. Use settings the new image actually supports (or a pinned test image/build that honors the intended Observer limit), and remove the ineffective legacy variables. ########## fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/factory/SourceReaderFactory.java: ########## @@ -38,6 +39,7 @@ public final class SourceReaderFactory { static { register(DataSource.MYSQL, MySqlSourceReader::new); register(DataSource.POSTGRES, PostgresSourceReader::new); + register(DataSource.OCEANBASE, OceanBaseSourceReader::new); Review Comment: [P2] Define ownership/cleanup for OceanBase readers before registering them. CREATE sends `/api/initReader` to an unbound round-robin BE, and each snapshot `/api/fetchSplits` call can hit another BE; both endpoints cache this factory-created reader in that BE's `Env`. Those contexts are never claimed, keep `lastAliveTime`/`maxIntervalMs` at zero so idle cleanup skips them, and DROP sends `/api/close` to only the runtime/bound/random BE. In a multi-BE cluster, repeated OceanBase create/snapshot/drop cycles therefore leave a job context plus its configuration/deserializer/cache and lazy executor-service object on every other contacted BE indefinitely. Split discovery closes its JDBC resources, so make these cached contexts throwaway and close them, or track all contacted BEs and close each; add a multi-BE lifecycle test. ########## fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/OceanBaseTestBase.java: ########## @@ -0,0 +1,150 @@ +// 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.cdcclient.itcase; + +import static java.util.concurrent.TimeUnit.SECONDS; + +import org.awaitility.Awaitility; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy; +import org.testcontainers.utility.DockerImageName; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.time.Duration; + +abstract class OceanBaseTestBase { + + protected static final int OCEANBASE_CDC_PORT = 2883; + protected static final String USER = "root@test"; + protected static final String PASSWORD = "123456"; + + private static final String EXTERNAL_HOST = System.getProperty("oceanbase.host"); + private static final int EXTERNAL_CDC_PORT = + Integer.getInteger("oceanbase.cdc.port", OCEANBASE_CDC_PORT); + + static final GenericContainer<?> OCEANBASE = + new GenericContainer<>( + DockerImageName.parse("quay.io/oceanbase/obbinlog-ce:4.2.5-test")) + .withExposedPorts(2881, OCEANBASE_CDC_PORT) + .withStartupTimeout(Duration.ofMinutes(6)) + .waitingFor( + new LogMessageWaitStrategy() + .withRegEx(".*OBBinlog is ready!.*") + .withTimes(1) + .withStartupTimeout(Duration.ofMinutes(6))); + + @BeforeAll Review Comment: [P2] Avoid booting this image once per subclass. JUnit runs the inherited `@BeforeAll`/`@AfterAll` pair for each of the three concrete OceanBase IT classes, so the same static container is started and destroyed three times in one reused Failsafe fork. The image entrypoint has at least 83 seconds of fixed serial sleeps before readiness, adding roughly three avoidable minutes to this CI job. Keep the per-test database cleanup, but use one suite/JVM-scoped OceanBase container (or consolidate these cases under one concrete fixture) and stop it once. ########## fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/OceanBaseTestBase.java: ########## @@ -0,0 +1,150 @@ +// 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.cdcclient.itcase; + +import static java.util.concurrent.TimeUnit.SECONDS; + +import org.awaitility.Awaitility; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy; +import org.testcontainers.utility.DockerImageName; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.time.Duration; + +abstract class OceanBaseTestBase { + + protected static final int OCEANBASE_CDC_PORT = 2883; + protected static final String USER = "root@test"; + protected static final String PASSWORD = "123456"; + + private static final String EXTERNAL_HOST = System.getProperty("oceanbase.host"); + private static final int EXTERNAL_CDC_PORT = + Integer.getInteger("oceanbase.cdc.port", OCEANBASE_CDC_PORT); + + static final GenericContainer<?> OCEANBASE = + new GenericContainer<>( + DockerImageName.parse("quay.io/oceanbase/obbinlog-ce:4.2.5-test")) Review Comment: [P2] Retry this known-transient image startup. The upstream Flink CDC fixture for this exact `obbinlog-ce:4.2.5-test` image added `.withStartupAttempts(3)` because Observer occasionally dies during CI initialization with `ob clog disk hang` before `OBBinlog is ready!` (https://github.com/apache/flink-cdc/commit/78baa80afa3fe036b214d9bf8010219178f774bf). Here `OCEANBASE.start()` gets one attempt, so one transient abort fails every new IT despite the six-minute timeout. Match the upstream retry and attach a log consumer so exhausted attempts are diagnosable. -- 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]
