This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 8fd2c1bd805 [fix](streaming-job) initialize CDC reader at CREATE JOB
and propagate remote errors (#64728)
8fd2c1bd805 is described below
commit 8fd2c1bd805a49f93afc2c57a283015d7e76beb8
Author: wudi <[email protected]>
AuthorDate: Thu Jun 25 15:49:07 2026 +0800
[fix](streaming-job) initialize CDC reader at CREATE JOB and propagate
remote errors (#64728)
### What problem does this PR solve?
For JDBC (MySQL/PostgreSQL) CDC streaming jobs in `initial`/`snapshot`
mode, the remote
CDC reader — which creates the PostgreSQL replication slot and
publication — was only
initialized lazily on the first runtime split fetch. Two problems
followed:
1. **CREATE JOB succeeded even when the source side was unusable.** If
reader
initialization failed at runtime (e.g. the source user lacks privilege
to
`CREATE PUBLICATION`), the job was already created and simply stayed
paused with no
progress and no clear failure at creation time.
2. **The real remote error was masked.** The cdc client returns failures
as a
`{code, data:"<message>"}` envelope over HTTP 200. FE force-deserialized
the response
straight into the success payload type, so on failure the error string
was coerced
into e.g. `List<SnapshotSplit>`, producing a misleading type-mismatch
that hid the
original cause.
---
.../job/offset/jdbc/JdbcSourceOffsetProvider.java | 96 ++++++++--------
.../JdbcSourceOffsetProviderAsyncSplitTest.java | 9 +-
.../JdbcSourceOffsetProviderErrorHandlingTest.java | 106 +++++++++++++++++
.../org/apache/doris/cdcclient/common/Env.java | 17 ++-
.../cdcclient/controller/ClientController.java | 2 +-
.../cdcclient/service/PipelineCoordinator.java | 6 +-
.../cdcclient/source/reader/SourceReader.java | 7 ++
.../reader/postgres/PostgresSourceReader.java | 20 ++--
.../cdcclient/itcase/CdcClientReadHarness.java | 13 ++-
.../cdcclient/itcase/CdcClientWriteHarness.java | 13 ++-
...gres_job_slot_dropped_during_incremental.groovy | 128 +++++++++++++++++++++
11 files changed, 356 insertions(+), 61 deletions(-)
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 bb8e59326ef..0a9e8df0921 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
@@ -48,6 +48,7 @@ import org.apache.doris.thrift.TStatusCode;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.gson.Gson;
import com.google.gson.annotations.SerializedName;
@@ -283,23 +284,13 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
"Failed to get end offset from backend," +
result.getStatus().getErrorMsgs(0) + ", response: "
+ result.getResponse());
}
- String response = result.getResponse();
- try {
- ResponseBody<Map<String, String>> responseObj =
objectMapper.readValue(
- response,
- new TypeReference<ResponseBody<Map<String, String>>>()
{
- }
- );
- Map<String, String> newEndOffset = responseObj.getData();
- // null→value also counts as a change: upstream may have
advanced while fetch was blocked.
- if (endBinlogOffset == null ||
!endBinlogOffset.equals(newEndOffset)) {
- hasMoreData = true;
- }
- endBinlogOffset = newEndOffset;
- } catch (JsonProcessingException e) {
- log.warn("Failed to parse end offset response: {}", response);
- throw new JobException(response);
+ Map<String, String> newEndOffset = parseCdcResponseData(
+ result.getResponse(), new TypeReference<Map<String,
String>>() {});
+ // null→value also counts as a change: upstream may have advanced
while fetch was blocked.
+ if (endBinlogOffset == null ||
!endBinlogOffset.equals(newEndOffset)) {
+ hasMoreData = true;
}
+ endBinlogOffset = newEndOffset;
} catch (TimeoutException te) {
log.warn("cdc_client RPC timeout api=/api/fetchEndOffset jobId={}
backend={}:{} timeout_sec={}",
getJobId(), backend.getHost(), backend.getBrpcPort(),
@@ -382,18 +373,9 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
"Failed to compare offset ," +
result.getStatus().getErrorMsgs(0) + ", response: "
+ result.getResponse());
}
- String response = result.getResponse();
- try {
- ResponseBody<Integer> responseObj = objectMapper.readValue(
- response,
- new TypeReference<ResponseBody<Integer>>() {
- }
- );
- return responseObj.getData() > 0;
- } catch (JsonProcessingException e) {
- log.warn("Failed to parse compare offset response: {}",
response);
- throw new JobException("Failed to parse compare offset
response: " + response);
- }
+ Integer cmp = parseCdcResponseData(
+ result.getResponse(), new TypeReference<Integer>() {});
+ return cmp != null && cmp > 0;
} catch (TimeoutException te) {
log.warn("cdc_client RPC timeout api=/api/compareOffset jobId={}
backend={}:{} timeout_sec={}",
getJobId(), backend.getHost(), backend.getBrpcPort(),
@@ -643,21 +625,23 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
// ============ Async split progress (driven by scheduler each tick)
============
/**
- * One-time setup at CREATE.
- * - initial/snapshot mode: init split progress; scheduler will drive
advanceSplits() each tick.
- * - latest mode (and other non-splitting modes): open the remote reader
(e.g. PG slot) so the
- * binlog phase can start immediately; no snapshot splitting will happen.
+ * One-time setup at CREATE. Opens the remote reader for every mode (see
initSourceReader) so
+ * source-side problems fail fast, then:
+ * - initial/snapshot mode: also init split progress; scheduler will drive
advanceSplits() each tick.
+ * - latest mode (and other non-splitting modes): nothing further; the
binlog phase can start
+ * immediately and no snapshot splitting will happen.
*/
@Override
public void initOnCreate(List<String> syncTables) throws JobException {
- if (!checkNeedSplitChunks(sourceProperties)) {
- initSourceReader();
- return;
- }
- synchronized (splitsLock) {
- this.cachedSyncTables = syncTables;
- this.committedSplitProgress = new SplitProgress();
- this.cdcSplitProgress = new SplitProgress();
+ // Open the reader for every mode so connectivity / auth / slot /
publication problems fail
+ // the CREATE JOB instead of surfacing later as a paused,
non-progressing job.
+ initSourceReader();
+ if (checkNeedSplitChunks(sourceProperties)) {
+ synchronized (splitsLock) {
+ this.cachedSyncTables = syncTables;
+ this.committedSplitProgress = new SplitProgress();
+ this.cdcSplitProgress = new SplitProgress();
+ }
}
}
@@ -872,10 +856,8 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
if (code != TStatusCode.OK) {
throw new JobException("fetchSplits backend error: " +
result.getStatus().getErrorMsgs(0));
}
- ResponseBody<List<SnapshotSplit>> resp = objectMapper.readValue(
- result.getResponse(),
- new TypeReference<ResponseBody<List<SnapshotSplit>>>() {});
- return resp.getData();
+ return parseCdcResponseData(
+ result.getResponse(), new
TypeReference<List<SnapshotSplit>>() {});
} catch (TimeoutException te) {
throw new JobException("fetchSplits RPC timeout: jobId=" +
getJobId() + " table=" + table);
} catch (Exception ex) {
@@ -883,6 +865,30 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
}
}
+ /**
+ * Decode a remote response envelope. A failure is returned as {@code
{code:1, data:"<message>"}}
+ * over HTTP 200, while success carries the typed payload in {@code data}.
Decode the envelope
+ * with a lenient {@link JsonNode} data field first so a failure throws
the raw response (which
+ * carries the real error in {@code data}) instead of a misleading
type-mismatch from forcing the
+ * success type onto an error string. Package-private for unit testing.
+ */
+ <T> T parseCdcResponseData(String response, TypeReference<T> dataType)
throws JobException {
+ ResponseBody<JsonNode> body;
+ try {
+ body = objectMapper.readValue(response, new
TypeReference<ResponseBody<JsonNode>>() {});
+ } catch (JsonProcessingException e) {
+ throw new JobException(response);
+ }
+ if (body.getCode() != RestApiStatusCode.OK.code) {
+ throw new JobException(response);
+ }
+ try {
+ return objectMapper.convertValue(body.getData(), dataType);
+ } catch (Exception e) {
+ throw new JobException(response);
+ }
+ }
+
protected boolean checkNeedSplitChunks(Map<String, String>
sourceProperties) {
String startMode = sourceProperties.get(DataSourceConfigKeys.OFFSET);
if (startMode == null) {
@@ -956,7 +962,7 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
* For example, PG slots need to be created first;
* otherwise, conflicts will occur in multi-backends scenarios.
*/
- private void initSourceReader() throws JobException {
+ protected void initSourceReader() throws JobException {
Backend backend = StreamingJobUtils.selectBackend(cloudCluster);
JobBaseConfig requestParams =
new JobBaseConfig(getJobId().toString(), sourceType.name(),
sourceProperties, getFrontendAddress());
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderAsyncSplitTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderAsyncSplitTest.java
index e6cd257ffee..a4a1230aada 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderAsyncSplitTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderAsyncSplitTest.java
@@ -63,13 +63,18 @@ public class JdbcSourceOffsetProviderAsyncSplitTest {
TestableProvider() {
super();
- // Default to initial mode so initOnCreate() takes the splitting
path
- // (latest mode would try to call initSourceReader against a real
backend).
+ // Default to initial mode so initOnCreate() takes the splitting
path.
this.sourceProperties.put(
org.apache.doris.job.cdc.DataSourceConfigKeys.OFFSET,
org.apache.doris.job.cdc.DataSourceConfigKeys.OFFSET_INITIAL);
}
+ // initOnCreate() now opens the remote reader for every mode; stub it
out so the unit test
+ // doesn't issue a real backend RPC.
+ @Override
+ protected void initSourceReader() {
+ }
+
@Override
protected List<SnapshotSplit> rpcFetchSplitsBatch(String table,
Object[] startVal, Integer splitId) {
rpcCalls.add(new RpcCall(table, startVal, splitId));
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderErrorHandlingTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderErrorHandlingTest.java
new file mode 100644
index 00000000000..1da12b3c97a
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderErrorHandlingTest.java
@@ -0,0 +1,106 @@
+// 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.offset.jdbc;
+
+import org.apache.doris.job.cdc.split.SnapshotSplit;
+import org.apache.doris.job.exception.JobException;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Covers the remote-error handling added to {@link JdbcSourceOffsetProvider}:
+ * the response-envelope parse must surface the original remote error instead
of a
+ * type-mismatch, and initOnCreate must fail fast when the remote reader
cannot be opened.
+ */
+public class JdbcSourceOffsetProviderErrorHandlingTest {
+
+ @Test
+ public void testParseSuccessEnvelopeReturnsTypedPayload() throws
JobException {
+ JdbcSourceOffsetProvider provider = new JdbcSourceOffsetProvider();
+ String response = "{\"code\":0,\"msg\":\"Success\","
+ + "\"data\":[{\"splitId\":\"db.t:0\",\"tableId\":\"db.t\"}]}";
+ List<SnapshotSplit> splits =
+ provider.parseCdcResponseData(response, new
TypeReference<List<SnapshotSplit>>() {});
+ Assert.assertEquals(1, splits.size());
+ Assert.assertEquals("db.t:0", splits.get(0).getSplitId());
+ Assert.assertEquals("db.t", splits.get(0).getTableId());
+ }
+
+ @Test
+ public void testParseFailureEnvelopeSurfacesOriginalError() {
+ JdbcSourceOffsetProvider provider = new JdbcSourceOffsetProvider();
+ String realError = "permission denied for database regression_db";
+ String response = "{\"code\":1,\"msg\":\"Internal Error\",\"data\":\""
+ realError + "\"}";
+ try {
+ provider.parseCdcResponseData(response, new
TypeReference<List<SnapshotSplit>>() {});
+ Assert.fail("a failed envelope must throw");
+ } catch (JobException e) {
+ Assert.assertTrue("the real remote error must be surfaced, got: "
+ e.getMessage(),
+ e.getMessage().contains(realError));
+ }
+ }
+
+ @Test
+ public void
testParseSuccessEnvelopeWithIncompatibleDataSurfacesRawResponse() {
+ JdbcSourceOffsetProvider provider = new JdbcSourceOffsetProvider();
+ // code=0 (remote thinks it succeeded) but data is a String where a
Map is expected.
+ String response =
"{\"code\":0,\"msg\":\"Success\",\"data\":\"not-a-map\"}";
+ try {
+ provider.parseCdcResponseData(response, new
TypeReference<Map<String, String>>() {});
+ Assert.fail("an incompatible success payload must throw");
+ } catch (JobException e) {
+ Assert.assertTrue("the raw response must be surfaced, got: " +
e.getMessage(),
+ e.getMessage().contains("not-a-map"));
+ }
+ }
+
+ @Test
+ public void testParseUnparseableResponseThrows() {
+ JdbcSourceOffsetProvider provider = new JdbcSourceOffsetProvider();
+ String response = "<html>502 Bad Gateway</html>";
+ try {
+ provider.parseCdcResponseData(response, new
TypeReference<Integer>() {});
+ Assert.fail("an unparseable response must throw");
+ } catch (JobException e) {
+ Assert.assertTrue("the raw response must be surfaced, got: " +
e.getMessage(),
+ e.getMessage().contains("502"));
+ }
+ }
+
+ @Test
+ public void testInitOnCreateFailsFastWhenReaderInitFails() {
+ JdbcSourceOffsetProvider provider = new JdbcSourceOffsetProvider() {
+ @Override
+ protected void initSourceReader() throws JobException {
+ throw new JobException("simulated reader init failure");
+ }
+ };
+ try {
+ provider.initOnCreate(Collections.singletonList("db.t"));
+ Assert.fail("CREATE JOB must fail when the remote reader cannot be
opened");
+ } catch (JobException e) {
+ Assert.assertTrue(e.getMessage().contains("simulated reader init
failure"));
+ }
+ }
+}
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/common/Env.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/common/Env.java
index b39efd01324..f6d4e5d1243 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/common/Env.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/common/Env.java
@@ -85,12 +85,19 @@ public class Env {
}
public SourceReader getReader(JobBaseConfig jobConfig) {
+ return getReader(jobConfig, false);
+ }
+
+ // createResources=true only for CREATE (/api/initReader) and standalone
TVF; rebuild paths pass
+ // false so a dropped slot is not silently recreated.
+ public SourceReader getReader(JobBaseConfig jobConfig, boolean
createResources) {
if (jobConfig.getFrontendAddress() != null &&
!jobConfig.getFrontendAddress().isEmpty()) {
this.feMasterAddress = jobConfig.getFrontendAddress();
}
DataSource ds = resolveDataSource(jobConfig.getDataSource());
Env manager = Env.getCurrentEnv();
- return manager.getOrCreateReader(jobConfig.getJobId(), ds,
jobConfig.getConfig());
+ return manager.getOrCreateReader(
+ jobConfig.getJobId(), ds, jobConfig.getConfig(),
createResources);
}
/** Return the reader only if already created, else null (never creates
one). */
@@ -220,7 +227,10 @@ public class Env {
}
private SourceReader getOrCreateReader(
- String jobId, DataSource dataSource, Map<String, String> config) {
+ String jobId,
+ DataSource dataSource,
+ Map<String, String> config,
+ boolean createResources) {
Objects.requireNonNull(jobId, "jobId is null");
Objects.requireNonNull(dataSource, "dataSource is null");
JobContext context = jobContexts.get(jobId);
@@ -240,6 +250,9 @@ public class Env {
LOG.info("Creating new reader for job {}, dataSource {}", jobId,
dataSource);
context = new JobContext(jobId, dataSource, config);
SourceReader reader = context.initializeReader();
+ if (createResources) {
+ reader.createSourceResources(jobId, dataSource, config);
+ }
jobContexts.put(jobId, context);
return reader;
} finally {
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/controller/ClientController.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/controller/ClientController.java
index b1dbe773844..cc0bac0665b 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/controller/ClientController.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/controller/ClientController.java
@@ -51,7 +51,7 @@ public class ClientController {
@RequestMapping(path = "/api/initReader", method = RequestMethod.POST)
public Object initSourceReader(@RequestBody JobBaseConfig jobConfig) {
try {
- SourceReader reader = Env.getCurrentEnv().getReader(jobConfig);
+ SourceReader reader = Env.getCurrentEnv().getReader(jobConfig,
true);
return RestResponse.success("Source reader initialized
successfully");
} catch (Exception ex) {
LOG.error("Failed to create reader, jobId={}",
jobConfig.getJobId(), ex);
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java
index 735b2b6b815..56d64546c69 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java
@@ -128,7 +128,7 @@ public class PipelineCoordinator {
LOG.info("Generated meta for job {}: {}", fetchReq.getJobId(),
meta);
}
- sourceReader = Env.getCurrentEnv().getReader(fetchReq);
+ sourceReader = Env.getCurrentEnv().getReader(fetchReq,
!isLong(fetchReq.getJobId()));
readResult = sourceReader.prepareAndSubmitSplit(fetchReq);
} catch (Exception ex) {
throw new CommonException(ex);
@@ -286,7 +286,9 @@ public class PipelineCoordinator {
/** pull data from api for test */
public RecordWithMeta fetchRecords(FetchRecordRequest fetchRecordRequest)
throws Exception {
- SourceReader sourceReader =
Env.getCurrentEnv().getReader(fetchRecordRequest);
+ SourceReader sourceReader =
+ Env.getCurrentEnv()
+ .getReader(fetchRecordRequest,
!isLong(fetchRecordRequest.getJobId()));
SplitReadResult readResult =
sourceReader.prepareAndSubmitSplit(fetchRecordRequest);
return buildRecordResponse(sourceReader, fetchRecordRequest,
readResult);
}
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/SourceReader.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/SourceReader.java
index c51572b377d..a924f33dec1 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/SourceReader.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/SourceReader.java
@@ -43,6 +43,13 @@ public interface SourceReader {
/** Initialization, called when the program starts */
void initialize(String jobId, DataSource dataSource, Map<String, String>
config);
+ /**
+ * Provision job-scoped source resources (PG slot/publication) once on
first open; rebuilds must
+ * not recreate them.
+ */
+ default void createSourceResources(
+ String jobId, DataSource dataSource, Map<String, String> config) {}
+
/** Divide the data to be read. For example: split mysql to chunks */
List<AbstractSourceSplit> getSourceSplits(FetchTableSplitsRequest config);
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReader.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReader.java
index fe37d5c392f..20b5694fd76 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReader.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReader.java
@@ -103,6 +103,19 @@ public class PostgresSourceReader extends
JdbcIncrementalSourceReader {
@Override
public void initialize(String jobId, DataSource dataSource, Map<String,
String> config) {
+ super.initialize(jobId, dataSource, config);
+ // Inject PG schema refresher so the deserializer can fetch accurate
column types on DDL
+ if (serializer instanceof PostgresDebeziumJsonDeserializer) {
+ ((PostgresDebeziumJsonDeserializer) serializer)
+ .setPgSchemaRefresher(
+ tableId -> refreshSingleTableSchema(tableId,
config, jobId));
+ }
+ }
+
+ // First open only, NOT initialize: a rebuilt reader must not recreate a
dropped slot.
+ @Override
+ public void createSourceResources(
+ String jobId, DataSource dataSource, Map<String, String> config) {
PostgresSourceConfig sourceConfig = generatePostgresConfig(config,
jobId, 0);
PostgresDialect dialect = new PostgresDialect(sourceConfig);
// Doris-owned publication: pre-create it covering all include_tables
(autocreate is
@@ -118,13 +131,6 @@ public class PostgresSourceReader extends
JdbcIncrementalSourceReader {
createSlotForGlobalStreamSplit(dialect);
}
}
- super.initialize(jobId, dataSource, config);
- // Inject PG schema refresher so the deserializer can fetch accurate
column types on DDL
- if (serializer instanceof PostgresDebeziumJsonDeserializer) {
- ((PostgresDebeziumJsonDeserializer) serializer)
- .setPgSchemaRefresher(
- tableId -> refreshSingleTableSchema(tableId,
config, jobId));
- }
}
/**
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientReadHarness.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientReadHarness.java
index d10aac75a33..1dd332bbc37 100644
---
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientReadHarness.java
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientReadHarness.java
@@ -143,12 +143,22 @@ public final class CdcClientReadHarness implements
AutoCloseable {
return new JobBaseConfig(jobId, dataSource, config, null);
}
+ /**
+ * Open the job's reader the way FE's CREATE JOB does via {@code
/api/initReader}:
+ * createResources=true so the PG slot/publication are pre-created on
first open. Idempotent —
+ * later opens reuse the cached reader. Must run before any coordinator
read, whose numeric jobId
+ * would otherwise open the reader with createResources=false and never
provision them.
+ */
+ private SourceReader openReader() {
+ return Env.getCurrentEnv().getReader(baseConfig(), true);
+ }
+
/**
* Compute every snapshot chunk for one table by advancing the FE-style
cursor (nextSplitStart /
* nextSplitId) until the final chunk (the one whose splitEnd is null) is
reached.
*/
public List<SnapshotSplit> fetchAllSnapshotSplits(String table) {
- SourceReader reader = Env.getCurrentEnv().getReader(baseConfig());
+ SourceReader reader = openReader();
List<SnapshotSplit> all = new ArrayList<>();
Object[] nextSplitStart = null;
Integer nextSplitId = null;
@@ -263,6 +273,7 @@ public final class CdcClientReadHarness implements
AutoCloseable {
req.setMeta(meta);
req.setTaskId(taskId);
+ openReader();
StreamingResponseBody body = coordinator.fetchRecordStream(req);
ByteArrayOutputStream out = new ByteArrayOutputStream();
body.writeTo(out);
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientWriteHarness.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientWriteHarness.java
index 199c1ad3a6f..c1eb902f46b 100644
---
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientWriteHarness.java
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientWriteHarness.java
@@ -159,9 +159,19 @@ final class CdcClientWriteHarness implements AutoCloseable
{
return new JobBaseConfig(jobId, dataSource, config, null);
}
+ /**
+ * Open the job's reader the way FE's CREATE JOB does via {@code
/api/initReader}:
+ * createResources=true so the PG slot/publication are pre-created on
first open. Idempotent —
+ * later opens reuse the cached reader. Must run before any coordinator
write, whose numeric jobId
+ * would otherwise open the reader with createResources=false and never
provision them.
+ */
+ private SourceReader openReader() {
+ return Env.getCurrentEnv().getReader(baseConfig(), true);
+ }
+
/** Compute every snapshot chunk for one table (mirrors FE's
/api/fetchSplits cursor). */
List<SnapshotSplit> fetchAllSnapshotSplits(String table) {
- SourceReader reader = Env.getCurrentEnv().getReader(baseConfig());
+ SourceReader reader = openReader();
List<SnapshotSplit> all = new ArrayList<>();
Object[] nextSplitStart = null;
Integer nextSplitId = null;
@@ -263,6 +273,7 @@ final class CdcClientWriteHarness implements AutoCloseable {
}
private void runWrite(Map<String, Object> meta) throws Exception {
+ openReader();
coordinator.writeRecords(buildRequest(meta));
capture();
}
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_slot_dropped_during_incremental.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_slot_dropped_during_incremental.groovy
index bbca6e22830..70bbd267b2b 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_slot_dropped_during_incremental.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_slot_dropped_during_incremental.groovy
@@ -170,5 +170,133 @@
suite("test_streaming_postgres_job_slot_dropped_during_incremental",
sql """DROP TABLE IF EXISTS ${pgDB}.${pgSchema}.${table1}"""
}
sql """drop table if exists ${currentDb}.${table1} force"""
+
+ // ===== Second scenario: same drop, but a Doris-owned slot (no
slot_name given) =====
+ // This is the case the fix actually guards: before it, the reader
rebuilt after the drop
+ // would silently recreate the Doris-owned slot at the latest LSN and
resume RUNNING (data
+ // loss). Now slot/publication are provisioned only at CREATE, so the
rebuild finds the slot
+ // gone and fails non-resumably.
+ def dorisJob = "test_streaming_pg_slot_dropped_doris_owned_job"
+ def dorisTable = "slot_dropped_doris_pg_tbl"
+ sql """DROP JOB IF EXISTS where jobname = '${dorisJob}'"""
+ sql """drop table if exists ${currentDb}.${dorisTable} force"""
+
+ connect("${pgUser}", "${pgPassword}",
"jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
+ sql """DROP TABLE IF EXISTS ${pgDB}.${pgSchema}.${dorisTable}"""
+ sql """CREATE TABLE ${pgDB}.${pgSchema}.${dorisTable} (
+ "id" int PRIMARY KEY,
+ "name" varchar(200)
+ )"""
+ }
+
+ // No slot_name/publication_name: Doris owns and auto-creates them at
CREATE (initReader).
+ sql """CREATE JOB ${dorisJob}
+ PROPERTIES ("max_interval" = "3")
+ ON STREAMING
+ FROM POSTGRES (
+ "jdbc_url" =
"jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}",
+ "driver_url" = "${driver_url}",
+ "driver_class" = "org.postgresql.Driver",
+ "user" = "${pgUser}",
+ "password" = "${pgPassword}",
+ "database" = "${pgDB}",
+ "schema" = "${pgSchema}",
+ "include_tables" = "${dorisTable}",
+ "offset" = "latest"
+ )
+ TO DATABASE ${currentDb} (
+ "table.create.properties.replication_num" = "1"
+ )
+ """
+
+ Awaitility.await().atMost(120, SECONDS).pollInterval(1,
SECONDS).until({
+ def st = sql """select status from jobs("type"="insert") where
Name='${dorisJob}'"""
+ st.size() == 1 && st.get(0).get(0) == "RUNNING"
+ })
+
+ // Phase 1: confirm steady-state incremental sync
+ connect("${pgUser}", "${pgPassword}",
"jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
+ for (int i = 1; i <= 10; i++) {
+ sql """INSERT INTO ${pgDB}.${pgSchema}.${dorisTable} VALUES
(${i}, 'name_${i}')"""
+ }
+ }
+ Awaitility.await().atMost(180, SECONDS).pollInterval(2,
SECONDS).until({
+ def cnt = sql """SELECT count(*) FROM ${currentDb}.${dorisTable}"""
+ cnt.size() == 1 && cnt.get(0).get(0) >= 10
+ })
+
+ // The Doris-owned slot is deterministic: doris_cdc_<jobId>. Derive it
from this job's Id
+ // rather than scanning LIKE 'doris_cdc_%' (which could pick another
suite's leftover or
+ // concurrent slot in the shared PG instance), then assert that exact
slot exists.
+ def dorisJobId = (sql """select Id from jobs("type"="insert") where
Name='${dorisJob}'""").get(0).get(0).toString()
+ def dorisSlot = "doris_cdc_${dorisJobId}"
+ connect("${pgUser}", "${pgPassword}",
"jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
+ def slotExists = sql """SELECT COUNT(1) FROM pg_replication_slots
WHERE slot_name = '${dorisSlot}'"""
+ assert slotExists[0][0] == 1 : "expected Doris-owned slot
${dorisSlot} to exist"
+ }
+ log.info("Doris-owned slot to drop: ${dorisSlot}")
+
+ // Phase 2: drop the Doris-owned slot out from under the running job.
Terminate whatever
+ // holds it, then drop; retry until gone (an auto-resume task may
briefly re-acquire it).
+ Awaitility.await().atMost(60, SECONDS).pollInterval(2, SECONDS).until({
+ boolean dropped = false
+ connect("${pgUser}", "${pgPassword}",
"jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
+ sql """SELECT pg_terminate_backend(active_pid) FROM
pg_replication_slots
+ WHERE slot_name = '${dorisSlot}' AND active_pid IS NOT
NULL"""
+ def cnt = sql """SELECT COUNT(1) FROM pg_replication_slots
WHERE slot_name = '${dorisSlot}'"""
+ if (cnt[0][0] == 0) {
+ dropped = true
+ } else {
+ try {
+ sql """SELECT
pg_drop_replication_slot('${dorisSlot}')"""
+ dropped = true
+ } catch (Exception e) {
+ log.info("slot still active, retry drop: " +
e.getMessage())
+ }
+ }
+ }
+ dropped
+ })
+
+ // Generate WAL so the rebuilt reader is forced to locate the
now-missing slot.
+ connect("${pgUser}", "${pgPassword}",
"jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
+ for (int i = 100; i < 110; i++) {
+ sql """INSERT INTO ${pgDB}.${pgSchema}.${dorisTable} VALUES
(${i}, 'after_drop_${i}')"""
+ }
+ }
+
+ // Phase 3: job settles in PAUSED with the slot-invalidated marker
+ Awaitility.await().atMost(240, SECONDS).pollInterval(3,
SECONDS).until({
+ def r = sql """select status, ErrorMsg from jobs("type"="insert")
where Name='${dorisJob}'"""
+ if (r.size() != 1) {
+ return false
+ }
+ def status = r.get(0).get(0)
+ def errMsg = r.get(0).get(1)
+ log.info("waiting slot-invalidated PAUSED: status=${status}
errMsg=${errMsg}")
+ status == "PAUSED" && errMsg != null &&
errMsg.toString().contains("Replication slot invalidated")
+ })
+
+ // Phase 4: must NOT auto-resume, and the dropped slot must NOT be
silently recreated
+ for (int i = 0; i < 8; i++) {
+ sleep(5000)
+ def r = sql """select status from jobs("type"="insert") where
Name='${dorisJob}'"""
+ assert r.size() == 1 && r.get(0).get(0) == "PAUSED" :
+ "job must stay PAUSED after slot invalidation, got ${r}"
+ }
+ connect("${pgUser}", "${pgPassword}",
"jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
+ def recreated = sql """SELECT COUNT(1) FROM pg_replication_slots
WHERE slot_name = '${dorisSlot}'"""
+ assert recreated[0][0] == 0 : "dropped Doris-owned slot must not
be silently recreated"
+ }
+
+ sql """DROP JOB IF EXISTS where jobname = '${dorisJob}'"""
+ connect("${pgUser}", "${pgPassword}",
"jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
+ def slotLeft = sql """SELECT COUNT(1) FROM pg_replication_slots
WHERE slot_name = '${dorisSlot}'"""
+ if (slotLeft[0][0] != 0) {
+ sql """SELECT pg_drop_replication_slot('${dorisSlot}')"""
+ }
+ sql """DROP TABLE IF EXISTS ${pgDB}.${pgSchema}.${dorisTable}"""
+ }
+ sql """drop table if exists ${currentDb}.${dorisTable} force"""
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
