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 232c06ff3040d4..981c2975fc7429 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 @@ -1199,6 +1199,10 @@ public class Config extends ConfigBase { @ConfField(mutable = true, masterOnly = true) public static int streaming_cdc_heavy_rpc_timeout_sec = 600; + // Max byte length of a PG database name for a CDC job; raise only for a larger NAMEDATALEN build. + @ConfField(mutable = true, masterOnly = true) + public static int streaming_pg_max_identifier_length = 63; + @ConfField(mutable = true, masterOnly = true) public static int streaming_cdc_fetch_splits_batch_size = 100; diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/PostgresResourceValidator.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/PostgresResourceValidator.java index d35cda9fe41c42..ae6277b3028b79 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/PostgresResourceValidator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/PostgresResourceValidator.java @@ -17,6 +17,7 @@ package org.apache.doris.job.extensions.insert.streaming; +import org.apache.doris.common.Config; import org.apache.doris.datasource.jdbc.client.JdbcClient; import org.apache.doris.job.cdc.DataSourceConfigKeys; import org.apache.doris.job.common.DataSourceType; @@ -25,6 +26,7 @@ import org.apache.commons.lang3.StringUtils; +import java.nio.charset.StandardCharsets; import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; @@ -44,6 +46,8 @@ public class PostgresResourceValidator { public static void validate(Map sourceProperties, String jobId, List tableNames) throws JobException { + // PG truncates an over-long db name, so the slot lookup never matches it; reject up front. + checkDatabaseNameLength(sourceProperties.get(DataSourceConfigKeys.DATABASE)); String slotName = resolveSlotName(sourceProperties, jobId); String publicationName = resolvePublicationName(sourceProperties, jobId); // Pattern-match ownership: name equals the default = Doris-owned (auto); otherwise user. @@ -123,6 +127,19 @@ private static String resolvePublicationName(Map config, String return StringUtils.isNotBlank(name) ? name : DataSourceConfigKeys.defaultPublicationName(jobId); } + private static void checkDatabaseNameLength(String database) throws JobException { + if (StringUtils.isBlank(database)) { + return; + } + // PG measures the identifier limit in bytes (NAMEDATALEN-1), so compare encoded bytes. + int bytes = database.getBytes(StandardCharsets.UTF_8).length; + if (bytes > Config.streaming_pg_max_identifier_length) { + throw new JobException("database name '" + database + "' is " + bytes + " bytes, exceeding " + + Config.streaming_pg_max_identifier_length + "; PostgreSQL truncates it and the" + + " replication-slot lookup would fail."); + } + } + private static boolean publicationExists(Connection conn, String publicationName) throws Exception { try (PreparedStatement ps = conn.prepareStatement("SELECT 1 FROM pg_publication WHERE pubname = ?")) { ps.setString(1, publicationName); diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java index 09090ba3ad184b..be14f79fa34485 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java @@ -709,12 +709,12 @@ protected void fetchMeta() throws JobException { || !InternalErrorCode.MANUAL_PAUSE_ERR.equals(this.getFailureReason().getCode())) { // When a job is manually paused, it does not need to be set again, // otherwise, it may be woken up by auto resume. + // Pause before setting the reason: updateJobStatus's writeLock orders this after any + // task-success callback that clears failureReason, so a success can't wipe the reason. + this.updateJobStatus(JobStatus.PAUSED); this.setFailureReason( new FailureReason(InternalErrorCode.GET_REMOTE_DATA_ERROR, "Failed to fetch meta, " + ex.getMessage())); - // If fetching meta fails, the job is paused - // and auto resume will automatically wake it up. - this.updateJobStatus(JobStatus.PAUSED); if (MetricRepo.isInit) { MetricRepo.COUNTER_STREAMING_JOB_GET_META_FAIL_COUNT.increase(1L); diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java index 9b7d364895bd06..3cfe67345640f0 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java @@ -924,13 +924,6 @@ protected long extractEventTimeMs(Map offsetMap) { return -1; } - @Override - public void onTaskCommitted(long scannedRows, long loadBytes) { - if (scannedRows == 0 && loadBytes == 0) { - hasMoreData = false; - } - } - @Override public boolean hasReachedEnd() { if (!isSnapshotOnlyMode()) { diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/PostgresResourceValidatorTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/PostgresResourceValidatorTest.java new file mode 100644 index 00000000000000..27e3b04b2cd904 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/PostgresResourceValidatorTest.java @@ -0,0 +1,45 @@ +// 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.job.extensions.insert.streaming; + +import org.apache.doris.job.cdc.DataSourceConfigKeys; +import org.apache.doris.job.exception.JobException; + +import org.apache.commons.lang3.StringUtils; +import org.junit.Assert; +import org.junit.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +public class PostgresResourceValidatorTest { + + // A 22-char CJK database name is length()==22 but 66 bytes in UTF-8; PG truncates it to 63 bytes. + // The byte-based check must reject it before connecting (validate fails on the very first line). + @Test + public void testRejectMultibyteOverLongDatabaseName() { + String dbName = StringUtils.repeat("εΊ“", 22); + Assert.assertEquals(22, dbName.length()); + Map props = new HashMap<>(); + props.put(DataSourceConfigKeys.DATABASE, dbName); + JobException e = Assert.assertThrows(JobException.class, + () -> PostgresResourceValidator.validate(props, "1", Collections.emptyList())); + Assert.assertTrue(e.getMessage(), e.getMessage().contains("bytes")); + } +} diff --git a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_special_offset.groovy b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_special_offset.groovy index c4214145151ba0..d9cf5886bcc8d9 100644 --- a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_special_offset.groovy +++ b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_special_offset.groovy @@ -143,6 +143,37 @@ suite("test_streaming_postgres_job_special_offset", "p0,external,pg,external_doc def jobStatus = sql """select status from jobs("type"="insert") where Name='${jobName}'""" return jobStatus[0][0] == "PAUSED" }) + + // PAUSE cancels the FE task but does NOT stop the in-flight cdc_client reader started before + // the pause: it keeps polling up to max_interval (default 10s) and would stream-load whatever + // rows we insert next, defeating the ALTER-offset reposition (this is the historical source of + // flakiness). No new reader is dispatched while PAUSED, so drain the lingering one + // deterministically: insert a disposable probe row, then wait until the row count stays stable + // for a window longer than max_interval. An active reader would have loaded the probe within + // max_interval, so a stable window proves no reader is consuming anymore. Finally delete the + // probe from both sides so it never pollutes the before/after-mark assertions or the .out. + def probeId = 10 + connect("${pgUser}", "${pgPassword}", "jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") { + sql """INSERT INTO ${pgDB}.${pgSchema}.${table1} VALUES (${probeId}, 'drain_probe')""" + } + int stablePolls = 0 + long lastCnt = -1L + Awaitility.await().atMost(120, SECONDS).pollInterval(3, SECONDS).until({ + long c = sql("""SELECT count(*) FROM ${currentDb}.${table1}""")[0][0] as long + if (c == lastCnt) { + stablePolls++ + } else { + stablePolls = 0 + lastCnt = c + } + // 5 * 3s = 15s stable > max_interval (10s) => the lingering reader has fully drained. + return stablePolls >= 5 + }) + connect("${pgUser}", "${pgPassword}", "jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") { + sql """DELETE FROM ${pgDB}.${pgSchema}.${table1} WHERE id = ${probeId}""" + } + sql """DELETE FROM ${currentDb}.${table1} WHERE id = ${probeId}""" + def alterLsn = "" connect("${pgUser}", "${pgPassword}", "jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") { // insert data BEFORE the LSN mark