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 a27bdb7acb2 [fix](streaming job) Harden alter and commit handling
(#68270)
a27bdb7acb2 is described below
commit a27bdb7acb201eaea8a189327f7d0a8ce983ebb5
Author: wudi <[email protected]>
AuthorDate: Wed Oct 14 10:26:22 2026 +0800
[fix](streaming job) Harden alter and commit handling (#68270)
### What problem does this PR solve?
Problem Summary:
This PR fixes streaming job lifecycle, ALTER, and commit handling:
- ALTER could partially mutate job state when source-property conversion
or validation failed, and changing an unrelated property reset the
current data-quality sampling window.
- CDC ALTER accepted offsets that did not match the source type,
allowing invalid PostgreSQL LSN or MySQL/OceanBase binlog positions to
fail later at runtime.
- Pausing or canceling a streaming job could leave internal scheduler
tasks registered and prevent subsequent scheduling.
- Journal replay could update the job's backend binding without
synchronizing the offset provider.
- TVF transaction commit retries could fetch a different or empty offset
after the first fetch consumed the task's result.
The fix validates ALTER inputs before mutating source state, resets
cloud offset before changing local offset state, preserves the sampling
window unless `max_interval` changes, and validates source-specific CDC
offset locations including GTID-only recovery. Internal scheduler tasks
are canceled and cleared without affecting user-visible task statistics.
Replay synchronizes the offset provider's backend binding, and each TVF
task caches its valid commit offset so transaction retries reuse the
same attachment data.
PostgreSQL stream recovery retains the missing-slot check and does not
recreate a missing slot. It does not reject recovery solely because
`restart_lsn` exceeds the saved stream offset: replication feedback can
advance slot positions during low traffic without advancing the saved
record offset. This comparison cannot reliably identify a recreated
slot. Detecting external recreation of a slot with the same name is
outside this change's guarantees.
---
.../doris/job/executor/DispatchTaskHandler.java | 7 +-
.../insert/streaming/StreamingInsertJob.java | 148 ++++++++++++---------
.../insert/streaming/StreamingInsertTask.java | 17 +++
.../org/apache/doris/job/manager/JobManager.java | 1 -
.../doris/job/offset/SourceOffsetProvider.java | 4 +-
.../job/offset/jdbc/JdbcSourceOffsetProvider.java | 42 +++++-
.../offset/jdbc/JdbcTvfSourceOffsetProvider.java | 7 +-
.../streaming/StreamingInsertJobLagTest.java | 10 +-
.../StreamingInsertJobOffsetPersistenceTest.java | 114 +++++++++++++++-
.../StreamingInsertJobStatusTransitionTest.java | 11 ++
.../streaming/StreamingInsertTaskAuditTest.java | 19 +++
.../jdbc/JdbcSourceOffsetProviderOffsetTest.java | 71 ++++++++++
.../source/reader/JdbcIncrementalSourceReader.java | 6 +-
.../source/reader/mysql/MySqlSourceReader.java | 6 +
.../reader/postgres/PostgresSourceReader.java | 41 +-----
.../source/reader/mysql/MySqlSourceReaderTest.java | 40 ++++++
16 files changed, 429 insertions(+), 115 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/executor/DispatchTaskHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/job/executor/DispatchTaskHandler.java
index 1db0cd50928..85223772779 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/executor/DispatchTaskHandler.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/executor/DispatchTaskHandler.java
@@ -65,7 +65,12 @@ public class DispatchTaskHandler<T extends AbstractJob>
implements WorkHandler<T
JobType jobType = event.getJob().getJobType();
for (AbstractTask task : tasks) {
if (!disruptorMap.get(jobType).addTask(task)) {
- task.cancel(true);
+ try {
+ task.cancel(true);
+ } finally {
+ // A rejected task cannot run and must not block
subsequent scheduling.
+ event.getJob().getRunningTasks().remove(task);
+ }
continue;
}
log.info("dispatch timer job success, job id is {}, task
id is {}",
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 780ef176e09..b5ede2873c0 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
@@ -433,6 +433,7 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
&& S3TableValuedFunction.NAME.equalsIgnoreCase(tvfType)) {
Offset offset =
validateOffset(jobProperties.getOffsetProperty());
this.offsetProvider.updateOffset(offset);
+ this.offsetProviderPersist = offsetProvider.getPersistInfo();
}
} catch (Exception ex) {
log.warn("init streaming insert job failed, sql: {}",
getExecuteSql(), ex);
@@ -486,6 +487,9 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
*/
public void validateAlterOffset(String offset) throws AnalysisException {
try {
+ if (tvfType != null) {
+ offsetProvider.ensureInitialized(getJobId(),
getOriginTvfProps());
+ }
offsetProvider.validateAlterOffset(offset);
} catch (Exception ex) {
throw new AnalysisException(ex.getMessage());
@@ -510,45 +514,74 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
}
public void alterJob(AlterJobCommand alterJobCommand) throws
AnalysisException, JobException {
- List<String> logParts = new ArrayList<>();
- // update sql
- if (StringUtils.isNotEmpty(alterJobCommand.getSql())) {
- setExecuteSql(alterJobCommand.getSql());
- initLogicalPlan(true);
- // refresh cached TVF props so fetchMeta and
createStreamingInsertTask
- // pick up the new credentials (e.g. aksk) from the updated SQL
- this.originTvfProps = getCurrentTvf().getProperties().getMap();
- String encryptedSql = generateEncryptedSql();
- logParts.add("sql: " + encryptedSql);
- }
+ writeLock();
+ try {
+ if (!JobStatus.PAUSED.equals(getJobStatus())) {
+ throw new AnalysisException("Only PAUSED job can be altered");
+ }
+ List<String> logParts = new ArrayList<>();
+
+ validateComputeGroupProperty(alterJobCommand.getProperties());
+
+ // Prepare the merged source properties before mutating the job.
+ Map<String, String> mergedSourceProperties = null;
+ Map<String, String> newConvertedSourceProperties = null;
+ if (!alterJobCommand.getSourceProperties().isEmpty()) {
+ mergedSourceProperties = new HashMap<>(this.sourceProperties);
+
mergedSourceProperties.putAll(alterJobCommand.getSourceProperties());
+ if
(alterJobCommand.getSourceProperties().containsKey(DataSourceConfigKeys.SERVER_ID)
+ || alterJobCommand.getSourceProperties()
+
.containsKey(DataSourceConfigKeys.SNAPSHOT_PARALLELISM)) {
+
DataSourceConfigValidator.validateServerIdConfig(mergedSourceProperties);
+ }
+ newConvertedSourceProperties =
buildConvertedSourceProperties(mergedSourceProperties);
+ }
- validateComputeGroupProperty(alterJobCommand.getProperties());
+ // update sql
+ if (StringUtils.isNotEmpty(alterJobCommand.getSql())) {
+ setExecuteSql(alterJobCommand.getSql());
+ initLogicalPlan(true);
+ // refresh cached TVF props so fetchMeta and
createStreamingInsertTask
+ // pick up the new credentials (e.g. aksk) from the updated SQL
+ this.originTvfProps = getCurrentTvf().getProperties().getMap();
+ String encryptedSql = generateEncryptedSql();
+ logParts.add("sql: " + encryptedSql);
+ }
- // update properties
- if (!alterJobCommand.getProperties().isEmpty()) {
- modifyPropertiesInternal(alterJobCommand.getProperties());
- logParts.add("properties: " + alterJobCommand.getProperties());
- }
+ // update properties
+ if (!alterJobCommand.getProperties().isEmpty()) {
+ String offsetProperty =
alterJobCommand.getProperties().get(StreamingJobProperties.OFFSET_PROPERTY);
+ if (StringUtils.isNotEmpty(offsetProperty)) {
+ Offset offset = validateOffset(offsetProperty);
+ if (Config.isCloudMode()) {
+ resetCloudProgress(offset);
+ }
+ offsetProvider.updateOffset(offset);
+ offsetProvider.resetLag();
+ this.offsetProviderPersist =
offsetProvider.getPersistInfo();
+ log.info("Alter streaming job {} offset to {}",
getJobId(), offsetProperty);
+ }
+ modifyPropertiesInternal(alterJobCommand.getProperties());
+ logParts.add("properties: " + alterJobCommand.getProperties());
+ }
- // update source properties
- if (!alterJobCommand.getSourceProperties().isEmpty()) {
- // Convert on a merged copy first; validateSource() only checks
ssl_rootcert is
- // non-empty, so cert-file lookup can still fail here. Commit to
fields only on success.
- Map<String, String> mergedSourceProperties = new
HashMap<>(this.sourceProperties);
-
mergedSourceProperties.putAll(alterJobCommand.getSourceProperties());
- Map<String, String> newConvertedSourceProperties =
- buildConvertedSourceProperties(mergedSourceProperties);
- this.sourceProperties = mergedSourceProperties;
- this.convertedSourceProperties = newConvertedSourceProperties;
- logParts.add("source properties: " +
alterJobCommand.getSourceProperties());
- }
+ // update source properties
+ if (mergedSourceProperties != null) {
+ this.sourceProperties = mergedSourceProperties;
+ this.convertedSourceProperties = newConvertedSourceProperties;
+ logParts.add("source properties: " +
alterJobCommand.getSourceProperties());
+ }
- // update target properties
- if (!alterJobCommand.getTargetProperties().isEmpty()) {
-
this.targetProperties.putAll(alterJobCommand.getTargetProperties());
- logParts.add("target properties: " +
alterJobCommand.getTargetProperties());
+ // update target properties
+ if (!alterJobCommand.getTargetProperties().isEmpty()) {
+
this.targetProperties.putAll(alterJobCommand.getTargetProperties());
+ logParts.add("target properties: " +
alterJobCommand.getTargetProperties());
+ }
+ log.info("Alter streaming job {}, {}", getJobId(), String.join(",
", logParts));
+ logUpdateOperation();
+ } finally {
+ writeUnlock();
}
- log.info("Alter streaming job {}, {}", getJobId(), String.join(", ",
logParts));
}
@Override
@@ -1087,6 +1120,19 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
}
try {
modifyPropertiesInternal(replayJob.getProperties());
+ if (replayJob.getOffsetProviderPersist() != null) {
+ setOffsetProviderPersist(replayJob.getOffsetProviderPersist());
+ if (tvfType != null && !Config.isCloudMode()) {
+
offsetProvider.restoreFromPersistInfo(offsetProviderPersist);
+ }
+ } else {
+ // Legacy journals lack provider state; keep local offset
replay without resetting MS.
+ String offset =
replayJob.getProperties().get(StreamingJobProperties.OFFSET_PROPERTY);
+ if (StringUtils.isNotEmpty(offset)) {
+ offsetProvider.updateOffset(validateOffset(offset));
+ this.offsetProviderPersist =
offsetProvider.getPersistInfo();
+ }
+ }
// When the pause state is restarted, it also needs to be updated
if (Config.isCloudMode()) {
replayOnCloudMode();
@@ -1095,9 +1141,6 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
// should not happen
log.error("replay modify streaming insert job properties failed,
job id: {}", getJobId(), e);
}
- if (replayJob.getOffsetProviderPersist() != null) {
- setOffsetProviderPersist(replayJob.getOffsetProviderPersist());
- }
if (replayJob.getNonTxnJobStatistic() != null) {
setNonTxnJobStatistic(replayJob.getNonTxnJobStatistic());
}
@@ -1119,24 +1162,11 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
setLastTaskSuccessTime(replayJob.getLastTaskSuccessTime());
setStartTimeMs(replayJob.getStartTimeMs());
this.boundBackendId = replayJob.boundBackendId;
+ offsetProvider.setBoundBackendId(boundBackendId);
}
- /**
- * When updating offset, you need to reset the currentOffset
- */
- private void modifyPropertiesInternal(Map<String, String> inputProperties)
throws AnalysisException, JobException {
- StreamingJobProperties inputStreamProps = new
StreamingJobProperties(inputProperties);
- if (StringUtils.isNotEmpty(inputStreamProps.getOffsetProperty())) {
- Offset offset =
validateOffset(inputStreamProps.getOffsetProperty());
- this.offsetProvider.updateOffset(offset);
- this.offsetProvider.resetLag();
- this.offsetProviderPersist = offsetProvider.getPersistInfo();
- log.info("modifyPropertiesInternal: offset updated to {}, job {}",
- inputStreamProps.getOffsetProperty(), getJobId());
- if (Config.isCloudMode()) {
- resetCloudProgress(offset);
- }
- }
+ /** Updates configuration without executing offset reset instructions
retained in replayed properties. */
+ private void modifyPropertiesInternal(Map<String, String> inputProperties)
{
if
(inputProperties.containsKey(StreamingJobProperties.COMPUTE_GROUP_PROPERTY)) {
this.cloudCluster =
inputProperties.get(StreamingJobProperties.COMPUTE_GROUP_PROPERTY);
offsetProvider.setCloudCluster(this.cloudCluster);
@@ -1406,15 +1436,7 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
LoadJob loadJob = loadJobs.get(0);
LoadStatistic loadStatistic = loadJob.getLoadStatistic();
- String offsetJson = offsetProvider.getCommitOffsetJson(
- runningStreamTask.getRunningOffset(),
- runningStreamTask.getTaskId(),
- runningStreamTask.getScanBackendIds());
-
- if (StringUtils.isBlank(offsetJson)) {
- throw new TransactionException("Cannot find offset for
attachment, load job id is "
- + runningStreamTask.getTaskId());
- }
+ String offsetJson = ((StreamingInsertTask)
runningStreamTask).getCommitOffsetJson();
txnState.setTxnCommitAttachment(new
StreamingTaskTxnCommitAttachment(
getJobId(),
runningStreamTask.getTaskId(),
@@ -1729,8 +1751,6 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
if (offsetProvider != null) {
// when fe restart, offsetProvider.jobId/sourceProperties may be
null
offsetProvider.ensureInitialized(getJobId(), getProviderProps());
- // replayOnUpdated skips the transient provider; resync routing BE.
- offsetProvider.setBoundBackendId(boundBackendId);
offsetProvider.replayIfNeed(this);
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java
index ec923146c93..a2642376937 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java
@@ -30,6 +30,7 @@ import org.apache.doris.job.common.TaskStatus;
import org.apache.doris.job.exception.JobException;
import org.apache.doris.job.extensions.insert.InsertTask;
import org.apache.doris.job.offset.SourceOffsetProvider;
+import org.apache.doris.job.offset.s3.S3EventSourceOffsetProvider;
import org.apache.doris.load.loadv2.LoadJob;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.glue.LogicalPlanAdapter;
@@ -47,6 +48,7 @@ import org.apache.doris.tablefunction.S3TableValuedFunction;
import org.apache.doris.thrift.TCell;
import org.apache.doris.thrift.TRow;
import org.apache.doris.thrift.TStatusCode;
+import org.apache.doris.transaction.TransactionException;
import com.google.common.base.Preconditions;
import lombok.Getter;
@@ -72,6 +74,7 @@ public class StreamingInsertTask extends
AbstractStreamingTask {
private StreamingJobProperties jobProperties;
private Map<String, String> originTvfProps;
private String cloudCluster;
+ private String commitOffsetJson;
private String auditSql;
private final boolean auditEnabled;
SourceOffsetProvider offsetProvider;
@@ -98,6 +101,7 @@ public class StreamingInsertTask extends
AbstractStreamingTask {
@Override
public void before() throws Exception {
+ commitOffsetJson = null;
auditSql = null;
if (getIsCanceled().get()) {
log.info("streaming insert task has been canceled, task id is {}",
getTaskId());
@@ -198,6 +202,18 @@ public class StreamingInsertTask extends
AbstractStreamingTask {
return Collections.emptyList();
}
+ public String getCommitOffsetJson() throws TransactionException {
+ // Reuse offsets across retries, but revalidate notification batch
receipts on every attempt.
+ if (commitOffsetJson == null || offsetProvider instanceof
S3EventSourceOffsetProvider) {
+ String offsetJson =
offsetProvider.getCommitOffsetJson(runningOffset, getTaskId(),
getScanBackendIds());
+ if (StringUtils.isBlank(offsetJson)) {
+ throw new TransactionException("Cannot find offset for
attachment, load job id is " + getTaskId());
+ }
+ commitOffsetJson = offsetJson;
+ }
+ return commitOffsetJson;
+ }
+
@Override
public boolean onSuccess() throws JobException {
if (getIsCanceled().get()) {
@@ -237,6 +253,7 @@ public class StreamingInsertTask extends
AbstractStreamingTask {
@Override
public void closeOrReleaseResources() {
+ commitOffsetJson = null;
if (null != stmtExecutor) {
stmtExecutor = null;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java
b/fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java
index 8c55856da57..d36b472ceb5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java
@@ -267,7 +267,6 @@ public class JobManager<T extends AbstractJob<?, C>, C>
implements Writable {
} else {
throw new JobException("Unsupported job type for ALTER:" +
job.getJobType());
}
- job.logUpdateOperation();
} finally {
writeUnlock();
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
index 0fbd5755c03..624e6a431dd 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
@@ -151,9 +151,7 @@ public interface SourceOffsetProvider {
}
/**
- * Restore offset from persisted string during image load
(gsonPostProcess).
- * Called immediately after the provider is created so that even PAUSED
jobs
- * have the correct offset state.
+ * Restore offset from an image or job update journal, including for
PAUSED jobs.
*/
default void restoreFromPersistInfo(String persistInfo) {
}
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 0eb327af571..94cfd4fbbac 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
@@ -529,13 +529,51 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
}
@Override
- public void validateAlterOffset(String offset) throws Exception {
- if (!DataSourceConfigValidator.isJsonOffset(offset)) {
+ public void validateAlterOffset(String offset) throws AnalysisException {
+ JsonNode offsetNode;
+ try {
+ offsetNode = objectMapper.readTree(offset);
+ } catch (Exception e) {
+ offsetNode = null;
+ }
+ if (offsetNode == null || !offsetNode.isObject()) {
throw new AnalysisException(
"ALTER JOB for CDC only supports JSON specific offset, "
+ "e.g. '{\"file\":\"binlog.000001\",\"pos\":\"154\"}' for
MySQL "
+ "or '{\"lsn\":\"12345678\"}' for PostgreSQL");
}
+
+ if ((sourceType == DataSourceType.MYSQL || sourceType ==
DataSourceType.OCEANBASE)
+ && offsetNode.has("kind") &&
!"SPECIFIC".equals(offsetNode.path("kind").asText())) {
+ throw new AnalysisException("ALTER JOB offset kind must be
SPECIFIC when specified");
+ }
+
+ boolean valid = switch (sourceType) {
+ case POSTGRES -> isNonNegativeLong(offsetNode.get("lsn"));
+ case MYSQL, OCEANBASE -> offsetNode.has("file") &&
offsetNode.has("pos")
+ ? hasText(offsetNode.path("file")) &&
isNonNegativeLong(offsetNode.get("pos"))
+ : hasText(offsetNode.path("gtids"));
+ default -> throw new AnalysisException(
+ "Unsupported CDC source type for ALTER JOB offset: " +
sourceType);
+ };
+ if (!valid) {
+ throw new AnalysisException("Invalid offset for " + sourceType);
+ }
+ }
+
+ private static boolean hasText(JsonNode value) {
+ return value.isTextual() && !value.asText().trim().isEmpty();
+ }
+
+ private static boolean isNonNegativeLong(JsonNode value) {
+ if (value == null || (!value.isTextual() &&
!value.isIntegralNumber())) {
+ return false;
+ }
+ try {
+ return Long.parseLong(value.asText()) >= 0;
+ } catch (NumberFormatException e) {
+ return false;
+ }
}
/**
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
index 9d0c88bbdcc..84cc3a94a53 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
@@ -437,8 +437,7 @@ public class JdbcTvfSourceOffsetProvider extends
JdbcSourceOffsetProvider {
}
/**
- * Restore chw/bop/ts from the image-persisted JSON. Called by
gsonPostProcess on FE startup
- * before any journal replay; recovers state lost when pre-checkpoint
journal is GC'd.
+ * Restore chw/bop/ts from image or job update journal JSON; preserves
cumulative snapshot progress.
*/
@Override
public void restoreFromPersistInfo(String persistInfo) {
@@ -450,6 +449,10 @@ public class JdbcTvfSourceOffsetProvider extends
JdbcSourceOffsetProvider {
this.chunkHighWatermarkMap = tmp.getChunkHighWatermarkMap();
this.binlogOffsetPersist = tmp.getBinlogOffsetPersist();
this.tableSchemas = tmp.getTableSchemas();
+ if (MapUtils.isNotEmpty(binlogOffsetPersist)) {
+ this.currentOffset = new JdbcOffset(
+ Collections.singletonList(new
BinlogSplit(binlogOffsetPersist)));
+ }
log.info("Restored TVF offset provider from persist: chw={},
bop={}, ts.len={}",
chunkHighWatermarkMap == null ? 0 :
chunkHighWatermarkMap.size(),
binlogOffsetPersist == null ? 0 :
binlogOffsetPersist.size(),
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobLagTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobLagTest.java
index 67ffc5f0170..a87b6881bf5 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobLagTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobLagTest.java
@@ -21,12 +21,15 @@ import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.job.base.JobExecutionConfiguration;
import org.apache.doris.job.base.TimerDefinition;
import org.apache.doris.job.cdc.split.BinlogSplit;
+import org.apache.doris.job.common.JobStatus;
import org.apache.doris.job.offset.jdbc.JdbcOffset;
import org.apache.doris.job.offset.jdbc.JdbcSourceOffsetProvider;
import org.apache.doris.job.offset.s3.S3EventSourceOffsetProvider;
+import org.apache.doris.nereids.trees.plans.commands.AlterJobCommand;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
import java.util.Collections;
import java.util.HashMap;
@@ -68,7 +71,9 @@ public class StreamingInsertJobLagTest {
@Test
public void testExplicitOffsetChangeInvalidatesLastObservedLag() throws
Exception {
- StreamingInsertJob job =
Deencapsulation.newInstance(StreamingInsertJob.class);
+ StreamingInsertJob job = Mockito.spy(new StreamingInsertJob());
+ Mockito.doNothing().when(job).logUpdateOperation();
+ job.setJobStatus(JobStatus.PAUSED);
Map<String, String> properties = new HashMap<>();
properties.put(StreamingJobProperties.MAX_INTERVAL_SECOND_PROPERTY,
"10");
Deencapsulation.setField(job, "properties", properties);
@@ -84,7 +89,8 @@ public class StreamingInsertJobLagTest {
Map<String, String> alterProperties = new HashMap<>();
alterProperties.put(StreamingJobProperties.OFFSET_PROPERTY,
"{\"lsn\":\"200\"}");
- Deencapsulation.invoke(job, "modifyPropertiesInternal",
alterProperties);
+ job.alterJob(new AlterJobCommand("test_job", alterProperties, null,
null, null,
+ Collections.emptyMap(), Collections.emptyMap()));
Assertions.assertEquals(-1, provider.getLagBytes());
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java
index f8e9bda0e6f..6fedf56cc8b 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java
@@ -18,8 +18,10 @@
package org.apache.doris.job.extensions.insert.streaming;
import org.apache.doris.catalog.Env;
+import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
import org.apache.doris.job.cdc.request.CommitOffsetRequest;
import org.apache.doris.job.cdc.split.SnapshotSplit;
import org.apache.doris.job.common.JobStatus;
@@ -29,10 +31,15 @@ import org.apache.doris.job.manager.JobManager;
import org.apache.doris.job.manager.StreamingTaskManager;
import org.apache.doris.job.offset.SourceOffsetProvider;
import org.apache.doris.job.offset.jdbc.JdbcSourceOffsetProvider;
+import org.apache.doris.job.offset.jdbc.JdbcTvfSourceOffsetProvider;
import org.apache.doris.job.offset.s3.S3Offset;
import org.apache.doris.job.offset.s3.S3SourceOffsetProvider;
+import org.apache.doris.nereids.trees.plans.commands.AlterJobCommand;
import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.tablefunction.CdcStreamTableValuedFunction;
+import org.apache.doris.tablefunction.S3TableValuedFunction;
import org.apache.doris.transaction.GlobalTransactionMgrIface;
+import org.apache.doris.transaction.TransactionState;
import org.apache.doris.transaction.TxnStateCallbackFactory;
import org.junit.jupiter.api.Assertions;
@@ -47,6 +54,63 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
public class StreamingInsertJobOffsetPersistenceTest {
+ @Test
+ public void testAlterChecksStatusAndPersistsUnderJobLock() throws
Exception {
+ TestStreamingInsertJob job = Mockito.spy(newJob(new
JdbcSourceOffsetProvider(), 1017L));
+ job.setJobStatus(JobStatus.PAUSED);
+ ReentrantReadWriteLock jobLock = Deencapsulation.getField(job, "lock");
+ Mockito.doAnswer(invocation -> {
+ Assertions.assertTrue(jobLock.isWriteLockedByCurrentThread());
+ return invocation.callRealMethod();
+ }).when(job).getJobStatus();
+ Mockito.doAnswer(invocation -> {
+ Assertions.assertTrue(jobLock.isWriteLockedByCurrentThread());
+ Assertions.assertEquals("1",
job.getTargetProperties().get("replication_num"));
+ return invocation.callRealMethod();
+ }).when(job).logUpdateOperation();
+ HashMap<String, String> targetProperties = new HashMap<>();
+ targetProperties.put("replication_num", "1");
+ AlterJobCommand command = new AlterJobCommand("test_job",
Collections.emptyMap(), null,
+ "mysql", null, Collections.emptyMap(), targetProperties);
+
+ job.alterJob(command);
+
+ Assertions.assertEquals(1, job.journalCount);
+ Assertions.assertFalse(jobLock.isWriteLocked());
+
+ // A resume between command validation and execution must reject ALTER
without mutation.
+ job.setJobStatus(JobStatus.PENDING);
+ targetProperties.put("replication_num", "3");
+ AnalysisException exception =
Assertions.assertThrows(AnalysisException.class, () -> job.alterJob(command));
+ Assertions.assertEquals("Only PAUSED job can be altered",
exception.getDetailMessage());
+ Assertions.assertEquals("1",
job.getTargetProperties().get("replication_num"));
+ Assertions.assertEquals(1, job.journalCount);
+ Assertions.assertFalse(jobLock.isWriteLocked());
+ }
+
+ @Test
+ public void testPausedTvfJobValidatesAlterOffsetAfterRestore() throws
Exception {
+ StreamingInsertJob job = new StreamingInsertJob();
+ Deencapsulation.setField(job, "jobId", 9001L);
+ job.setJobStatus(JobStatus.PAUSED);
+ Deencapsulation.setField(job, "tvfType",
CdcStreamTableValuedFunction.NAME);
+ HashMap<String, String> tvfProperties = new HashMap<>();
+ tvfProperties.put(DataSourceConfigKeys.TYPE, "postgres");
+ tvfProperties.put(DataSourceConfigKeys.JDBC_URL,
"jdbc:postgresql://127.0.0.1:5432/source_db");
+ tvfProperties.put(DataSourceConfigKeys.TABLE, "source_table");
+ Deencapsulation.setField(job, "originTvfProps", tvfProperties);
+
+ job.gsonPostProcess();
+ JdbcTvfSourceOffsetProvider provider = Deencapsulation.getField(job,
"offsetProvider");
+ Assertions.assertNull(Deencapsulation.getField(provider,
"sourceType"));
+
+ job.validateAlterOffset("{\"lsn\":\"300\"}");
+
+ Assertions.assertEquals(JobStatus.PAUSED, job.getJobStatus());
+ Assertions.assertThrows(AnalysisException.class,
+ () ->
job.validateAlterOffset("{\"file\":\"mysql-bin.000001\",\"pos\":\"154\"}"));
+ }
+
@Test
public void testFirstSnapshotCommitPersistsImmediately() throws Exception {
JdbcSourceOffsetProvider provider = new JdbcSourceOffsetProvider();
@@ -133,7 +197,7 @@ public class StreamingInsertJobOffsetPersistenceTest {
HashMap<String, String> properties = new HashMap<>();
properties.put(StreamingJobProperties.OFFSET_PROPERTY,
"{\"lsn\":\"300\"}");
- Deencapsulation.invoke(job, "modifyPropertiesInternal", properties);
+ alterOffset(job,
properties.get(StreamingJobProperties.OFFSET_PROPERTY));
Assertions.assertTrue(job.getOffsetProviderPersist().contains("300"));
Assertions.assertTrue(provider.getFinishedSplits().isEmpty());
@@ -297,6 +361,50 @@ public class StreamingInsertJobOffsetPersistenceTest {
Assertions.assertEquals(1234L, job.getStartTimeMs());
}
+ @Test
+ public void
testReplayUpdatedRestoresProgressWithoutReapplyingAlterOffset() throws
Exception {
+ S3SourceOffsetProvider provider = new S3SourceOffsetProvider();
+ StreamingInsertJob job = new StreamingInsertJob();
+ Deencapsulation.setField(job, "jobId", 9001L);
+ Deencapsulation.setField(job, "offsetProvider", provider);
+ Deencapsulation.setField(job, "properties", new HashMap<String,
String>());
+ Deencapsulation.setField(job, "jobProperties", new
StreamingJobProperties(job.getProperties()));
+ Deencapsulation.setField(job, "tvfType", S3TableValuedFunction.NAME);
+ alterOffset(job, "{\"fileName\":\"100.csv\"}");
+
+ TransactionState txnState = Mockito.mock(TransactionState.class);
+ Mockito.when(txnState.getTxnCommitAttachment()).thenReturn(new
StreamingTaskTxnCommitAttachment(
+ 9001L, 1017L, 0L, 0L, 0L, 0L, 0L,
"{\"endFile\":\"500.csv\"}"));
+ job.replayOnCommitted(txnState);
+ job.setJobStatus(JobStatus.PAUSED);
+ StreamingInsertJob pauseRecord =
GsonUtils.GSON.fromJson(GsonUtils.GSON.toJson(job), StreamingInsertJob.class);
+
+ job.replayOnUpdated(pauseRecord);
+
+ Assertions.assertEquals("{\"fileName\":\"500.csv\"}",
provider.getShowCurrentOffset());
+
Assertions.assertTrue(job.getOffsetProviderPersist().contains("500.csv"));
+ StreamingInsertJob restored =
GsonUtils.GSON.fromJson(GsonUtils.GSON.toJson(job), StreamingInsertJob.class);
+
Assertions.assertTrue(restored.getOffsetProviderPersist().contains("500.csv"));
+ SourceOffsetProvider restoredProvider =
Deencapsulation.getField(restored, "offsetProvider");
+ Assertions.assertEquals("{\"fileName\":\"500.csv\"}",
restoredProvider.getShowCurrentOffset());
+
+ alterOffset(pauseRecord, "{\"fileName\":\"100.csv\"}");
+ job.replayOnUpdated(pauseRecord);
+ Assertions.assertEquals("{\"fileName\":\"100.csv\"}",
provider.getShowCurrentOffset());
+ }
+
+ private static void alterOffset(StreamingInsertJob job, String offset)
throws Exception {
+ job.setJobStatus(JobStatus.PAUSED);
+ AlterJobCommand command = new AlterJobCommand("test_job",
+
Collections.singletonMap(StreamingJobProperties.OFFSET_PROPERTY, offset),
+ null, null, null, Collections.emptyMap(),
Collections.emptyMap());
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+ Env env = Mockito.mock(Env.class, Mockito.RETURNS_DEEP_STUBS);
+ envMockedStatic.when(Env::getCurrentEnv).thenReturn(env);
+ job.alterJob(command);
+ }
+ }
+
private static TestStreamingInsertJob newJob(SourceOffsetProvider
provider, long taskId) {
TestStreamingInsertJob job = new TestStreamingInsertJob();
Deencapsulation.setField(job, "lock", new
ReentrantReadWriteLock(true));
@@ -304,7 +412,9 @@ public class StreamingInsertJobOffsetPersistenceTest {
Deencapsulation.setField(job, "jobName", "test_job");
Deencapsulation.setField(job, "jobStatus", JobStatus.RUNNING);
Deencapsulation.setField(job, "offsetProvider", provider);
- Deencapsulation.setField(job, "properties", new HashMap<String,
String>());
+ HashMap<String, String> properties = new HashMap<>();
+ Deencapsulation.setField(job, "properties", properties);
+ Deencapsulation.setField(job, "jobProperties", new
StreamingJobProperties(properties));
Deencapsulation.setField(job, "targetProperties", new HashMap<String,
String>());
Deencapsulation.setField(job, "runningStreamTask", new
NoopStreamingMultiTblTask(taskId));
return job;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java
index e15de71ac3a..5d8909d6202 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java
@@ -73,6 +73,17 @@ public class StreamingInsertJobStatusTransitionTest {
Assertions.assertEquals(JobStatus.STOPPED, job.getJobStatus());
}
+ @Test
+ public void testFailedCommitCallbackReleasesWriteLock() throws Exception {
+ StreamingInsertJob job = newJob(JobStatus.RUNNING);
+ ReentrantReadWriteLock lock = Deencapsulation.getField(job, "lock");
+ lock.writeLock().lock();
+
+ job.afterCommitted(null, false);
+
+ Assertions.assertEquals(0, lock.getWriteHoldCount());
+ }
+
@Test
public void testFinishedStatusCannotBeOverwritten() {
StreamingInsertJob job = newJob(JobStatus.FINISHED);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java
index f171296fb6e..bb132542380 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java
@@ -30,6 +30,7 @@ import org.apache.doris.job.offset.Offset;
import org.apache.doris.job.offset.SourceOffsetProvider;
import org.apache.doris.job.offset.jdbc.JdbcTvfSourceOffsetProvider;
import org.apache.doris.job.offset.s3.S3EventOffset;
+import org.apache.doris.job.offset.s3.S3EventSourceOffsetProvider;
import org.apache.doris.job.offset.s3.S3Offset;
import org.apache.doris.job.offset.s3.S3SourceOffsetProvider;
import org.apache.doris.nereids.StatementContext;
@@ -105,6 +106,24 @@ public class StreamingInsertTaskAuditTest {
Assertions.assertFalse(s3Task.isNoRetry());
}
+ @Test
+ public void testCommitRetryRevalidatesNotificationReceipts() throws
Exception {
+ S3EventSourceOffsetProvider provider =
Mockito.mock(S3EventSourceOffsetProvider.class);
+ Mockito.when(provider.getSourceType()).thenReturn("s3");
+ S3EventOffset offset = new
S3EventOffset(Collections.singletonList("input/a.csv"));
+ String offsetJson = offset.toSerializedJson();
+ Mockito.when(provider.getCommitOffsetJson(offset, 2L,
Collections.emptyList()))
+ .thenReturn(offsetJson)
+ .thenThrow(new IllegalStateException("S3 EVENT batch receipts
expired before commit"));
+ StreamingInsertTask task = new StreamingInsertTask(1L, 2L, "",
provider, "test_db", null,
+ Collections.emptyMap(), UserIdentity.ROOT, null);
+ Deencapsulation.setField(task, "runningOffset", offset);
+
+ Assertions.assertEquals(offsetJson, task.getCommitOffsetJson());
+ Assertions.assertThrows(IllegalStateException.class,
task::getCommitOffsetJson);
+ Mockito.verify(provider, Mockito.times(2)).getCommitOffsetJson(offset,
2L, Collections.emptyList());
+ }
+
@Test
public void testS3RunSubmitsAuditEvent() throws Exception {
AuditEvent auditEvent = runS3Task(null);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java
index ed80084ef5f..8adac456c01 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java
@@ -17,11 +17,13 @@
package org.apache.doris.job.offset.jdbc;
+import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.job.cdc.DataSourceConfigKeys;
import org.apache.doris.job.cdc.split.BinlogSplit;
import org.apache.doris.job.cdc.split.SnapshotSplit;
+import org.apache.doris.job.common.DataSourceType;
import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob;
import org.junit.jupiter.api.Assertions;
@@ -34,6 +36,60 @@ import java.util.Map;
public class JdbcSourceOffsetProviderOffsetTest {
+ @Test
+ public void testValidateAlterOffsetAcceptsSourceSpecificPositions() throws
Exception {
+ JdbcSourceOffsetProvider postgres = provider(DataSourceType.POSTGRES);
+ postgres.validateAlterOffset("{\"lsn\":12345678,\"txId\":\"7\"}");
+ postgres.validateAlterOffset("{\"lsn\":\"12345678\"}");
+
+ JdbcSourceOffsetProvider mysql = provider(DataSourceType.MYSQL);
+
mysql.validateAlterOffset("{\"file\":\"mysql-bin.000001\",\"pos\":154,\"kind\":\"SPECIFIC\"}");
+
mysql.validateAlterOffset("{\"file\":\"mysql-bin.000001\",\"pos\":\"154\"}");
+
mysql.validateAlterOffset("{\"gtids\":\"24bc7850-2c16-11ef-a0c9-0242ac120002:1-9\"}");
+ mysql.validateAlterOffset(
+
"{\"gtids\":\"24bc7850-2c16-11ef-a0c9-0242ac120002:1-9\",\"kind\":\"SPECIFIC\"}");
+
+ JdbcSourceOffsetProvider oceanBase =
provider(DataSourceType.OCEANBASE);
+
oceanBase.validateAlterOffset("{\"file\":\"mysql-bin.000001\",\"pos\":\"154\"}");
+
oceanBase.validateAlterOffset("{\"gtids\":\"24bc7850-2c16-11ef-a0c9-0242ac120002:1-9\"}");
+ }
+
+ @Test
+ public void testValidateAlterOffsetRejectsInvalidSourceSpecificPositions()
{
+ assertInvalidAlterOffsets(provider(DataSourceType.POSTGRES),
+ "{}",
+ "{\"file\":\"mysql-bin.000001\",\"pos\":\"154\"}",
+ "{\"lsn\":null}",
+ "{\"lsn\":\"abc\"}",
+ "{\"lsn\":\"-1\"}",
+ "{\"lsn\":\"9223372036854775808\"}");
+
+ assertInvalidAlterOffsets(provider(DataSourceType.MYSQL),
+ "{}",
+ "{\"lsn\":\"12345678\"}",
+ "{\"file\":\"mysql-bin.000001\"}",
+ "{\"file\":\"\",\"pos\":\"154\"}",
+ "{\"file\":\"mysql-bin.000001\",\"pos\":null}",
+ "{\"file\":\"mysql-bin.000001\",\"pos\":\"abc\"}",
+ "{\"file\":\"mysql-bin.000001\",\"pos\":\"-1\"}",
+
"{\"file\":\"mysql-bin.000001\",\"pos\":\"9223372036854775808\"}",
+ "{\"gtids\":\"\"}",
+ "{\"gtids\":123}",
+
"{\"gtids\":\"uuid:1-10\",\"file\":\"mysql-bin.000001\",\"pos\":\"abc\"}");
+ }
+
+ @Test
+ public void testValidateAlterOffsetRejectsNonSpecificKind() {
+ for (DataSourceType type : new DataSourceType[] {DataSourceType.MYSQL,
DataSourceType.OCEANBASE}) {
+ assertInvalidAlterOffsets(provider(type),
+
"{\"file\":\"mysql-bin.000001\",\"pos\":154,\"kind\":\"LATEST\"}",
+
"{\"file\":\"mysql-bin.000001\",\"pos\":154,\"kind\":\"EARLIEST\"}",
+
"{\"file\":\"mysql-bin.000001\",\"pos\":154,\"kind\":\"TIMESTAMP\"}",
+ "{\"gtids\":\"uuid:1-10\",\"kind\":\"BOGUS\"}",
+ "{\"gtids\":\"uuid:1-10\",\"kind\":null}");
+ }
+ }
+
@Test
public void testTaskParallelismKeepsUncommittedSplitsForNextTask() {
JdbcSourceOffsetProvider provider = new JdbcSourceOffsetProvider();
@@ -192,6 +248,8 @@ public class JdbcSourceOffsetProviderOffsetTest {
new BinlogSplit(Collections.singletonMap("lsn", "200")))));
StreamingInsertJob job =
mockJobWithPersistInfo(source.getPersistInfo());
JdbcTvfSourceOffsetProvider restored = new
JdbcTvfSourceOffsetProvider();
+ restored.updateOffset(new JdbcOffset(Collections.singletonList(
+ new BinlogSplit(Collections.singletonMap("lsn", "100")))));
restored.restoreFromPersistInfo(source.getPersistInfo());
restored.replayIfNeed(job);
@@ -327,6 +385,19 @@ public class JdbcSourceOffsetProviderOffsetTest {
Assertions.assertNull(progress.getNextSplitId());
}
+ private static JdbcSourceOffsetProvider provider(DataSourceType
sourceType) {
+ JdbcSourceOffsetProvider provider = new JdbcSourceOffsetProvider();
+ provider.setSourceType(sourceType);
+ return provider;
+ }
+
+ private static void assertInvalidAlterOffsets(JdbcSourceOffsetProvider
provider, String... offsets) {
+ for (String offset : offsets) {
+ Assertions.assertThrows(AnalysisException.class,
+ () -> provider.validateAlterOffset(offset), offset);
+ }
+ }
+
private static class TestJdbcSourceOffsetProvider extends
JdbcSourceOffsetProvider {
private final int compareResult;
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
index 0592d83c54b..60be92b455e 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
@@ -444,7 +444,7 @@ public abstract class JdbcIncrementalSourceReader extends
AbstractCdcSourceReade
// Rebuild path: fail loudly if the source position is gone (e.g. slot
dropped) instead of
// silently re-locating from a lost offset.
- validateStreamSource(offsetMeta, baseReq);
+ validateStreamSource(newStreamSplit.getStartingOffset(), baseReq);
this.streamReader = getBinlogSplitReader(baseReq);
@@ -470,8 +470,8 @@ public abstract class JdbcIncrementalSourceReader extends
AbstractCdcSourceReade
}
// Source-specific check before (re)building the stream reader; default
no-op.
- protected void validateStreamSource(
- Map<String, Object> offsetMeta, JobBaseRecordRequest baseReq)
throws Exception {}
+ protected void validateStreamSource(Offset startingOffset,
JobBaseRecordRequest baseReq)
+ throws Exception {}
@Override
public Iterator<SourceRecord> pollRecords() throws Exception {
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
index 1e920969875..ed8acebd373 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
@@ -808,6 +808,12 @@ public class MySqlSourceReader extends
AbstractCdcSourceReader {
binlogSplit.getStartingOffset() == null
? new HashMap<>()
: new HashMap<>(binlogSplit.getStartingOffset());
+ if (!lastOffsetMap.containsKey(BinlogOffset.BINLOG_FILENAME_OFFSET_KEY)
+ && lastOffsetMap.containsKey(BinlogOffset.GTID_SET_KEY)) {
+
BinlogOffset.ofGtidSet(lastOffsetMap.get(BinlogOffset.GTID_SET_KEY))
+ .getOffset()
+ .forEach(lastOffsetMap::putIfAbsent);
+ }
// ALTER offset may not contain "kind", supplement it for Flink CDC
if (lastOffsetMap.containsKey(BinlogOffset.BINLOG_FILENAME_OFFSET_KEY)
&& !lastOffsetMap.containsKey(BinlogOffset.OFFSET_KIND_KEY)) {
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 e52502e5576..c6b89720676 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
@@ -540,12 +540,13 @@ public class PostgresSourceReader extends
JdbcIncrementalSourceReader {
}
}
- // Detect a replication slot that was dropped (or dropped and recreated)
out from under us while
- // the job was paused/retrying. Recreating it silently would resume from a
position whose WAL is
- // already gone -> data loss. Fail with a fixed marker so FE classifies it
as non-resumable.
+ // A missing slot cannot safely resume from the committed position; do not
recreate it.
+ // Do not compare restart_lsn with the saved offset: replication feedback
can advance slot
+ // positions beyond it during low traffic. This cannot reliably
distinguish normal advancement
+ // from a slot dropped and recreated with the same name.
@Override
- protected void validateStreamSource(
- Map<String, Object> offsetMeta, JobBaseRecordRequest baseReq)
throws Exception {
+ protected void validateStreamSource(Offset startingOffset,
JobBaseRecordRequest baseReq)
+ throws Exception {
PostgresSourceConfig sourceConfig = getSourceConfig(baseReq);
PostgresDialect dialect = new PostgresDialect(sourceConfig);
try (PostgresConnection connection = dialect.openJdbcConnection()) {
@@ -560,36 +561,6 @@ public class PostgresSourceReader extends
JdbcIncrementalSourceReader {
+ " committed position without data
loss.",
baseReq.getJobId(), dialect.getSlotName()));
}
- Lsn requestedLsn = extractRequestedLsn(offsetMeta);
- Lsn restartLsn = slotState.slotRestartLsn();
- // restart_lsn must stay <= committed position; a higher one means
the slot was
- // recreated
- // and the WAL between them was discarded, so resuming would
silently skip data.
- if (requestedLsn != null
- && requestedLsn.asLong() > 0
- && restartLsn != null
- && restartLsn.compareTo(requestedLsn) > 0) {
- throw new CdcClientException(
- String.format(
- "Replication slot invalidated for job %s: slot
%s restart_lsn %s is"
- + " ahead of the committed position %s
(slot recreated),"
- + " cannot resume without data loss.",
- baseReq.getJobId(),
- dialect.getSlotName(),
- restartLsn,
- requestedLsn));
- }
- }
- }
-
- private Lsn extractRequestedLsn(Map<String, Object> offsetMeta) {
- if (offsetMeta == null || offsetMeta.get(SourceInfo.LSN_KEY) == null) {
- return null;
- }
- try {
- return
Lsn.valueOf(Long.parseLong(String.valueOf(offsetMeta.get(SourceInfo.LSN_KEY))));
- } catch (NumberFormatException ex) {
- return null;
}
}
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
index 8905e0379e2..013ab9d2836 100644
---
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
@@ -29,8 +29,11 @@ import org.apache.doris.job.cdc.DataSourceConfigKeys;
import org.apache.doris.job.cdc.request.JobBaseConfig;
import com.github.shyiko.mysql.binlog.GtidSet;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
+import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffsetKind;
+import org.apache.flink.cdc.connectors.mysql.source.split.MySqlBinlogSplit;
import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSnapshotSplit;
import org.junit.jupiter.api.Test;
@@ -113,6 +116,43 @@ public class MySqlSourceReaderTest {
assertNotNull(off.getOffsetKind());
}
+ @Test
+ void binlogSplitUsesGtidOnlyOffsetWithoutFallingBack() throws Exception {
+ String gtid = SERVER_UUID + ":1-3:5-7";
+ MySqlSourceReader reader = new MySqlSourceReader();
+ reader.setTableSchemas(new HashMap<>());
+ JobBaseConfig config =
+ new JobBaseConfig(
+ "job-1",
+ "MYSQL",
+ Map.of(
+ DataSourceConfigKeys.JDBC_URL,
"jdbc:mysql://localhost:3306/testdb",
+ DataSourceConfigKeys.USER, "u",
+ DataSourceConfigKeys.PASSWORD, "p",
+ DataSourceConfigKeys.DATABASE, "testdb",
+ DataSourceConfigKeys.TABLE, "t_test",
+ DataSourceConfigKeys.OFFSET, "initial"),
+ null);
+ Method method =
+ MySqlSourceReader.class.getDeclaredMethod(
+ "createBinlogSplit", Map.class, JobBaseConfig.class);
+ method.setAccessible(true);
+ Tuple2<?, ?> result =
+ (Tuple2<?, ?>)
+ method.invoke(
+ reader,
+ Map.of("startingOffset", Map.of("gtids", gtid,
"event", "7", "row", "3")),
+ config);
+ BinlogOffset offset = ((MySqlBinlogSplit)
result.f0).getStartingOffset();
+
+ assertEquals(gtid, offset.getGtidSet());
+ assertEquals(BinlogOffsetKind.SPECIFIC, offset.getOffsetKind());
+ assertEquals("", offset.getFilename());
+ assertEquals(0L, offset.getPosition());
+ assertEquals(7L, offset.getRestartSkipEvents());
+ assertEquals(3L, offset.getRestartSkipRows());
+ }
+
@Test
void mysqlConfigIncludesSchemaChanges() throws Exception {
MySqlSourceConfig config = sourceConfig("initial");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
