This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 128dfd9583a branch-4.1: [improvement](streaming) Improve streaming job
error messages #68241 (#68867)
128dfd9583a is described below
commit 128dfd9583abcd33631d7eb4bcf7f882bbaec9ba
Author: wudi <[email protected]>
AuthorDate: Mon Oct 12 13:57:12 2026 +0800
branch-4.1: [improvement](streaming) Improve streaming job error messages
#68241 (#68867)
Cherry-picked from #68241
Conflict resolution: preserve branch-4.1's existing CDC type conversion
and method structure in `StreamingJobUtils`, and adapt the three FE
tests to JUnit 4.
Validation: FE Checkstyle passed with 0 violations; all 3
`DorisBatchStreamLoadTest` cases passed; `git diff --check` passed. FE
unit tests were blocked by sandbox permissions when writing the Maven
cache. Streaming regression suites were not run.
---
.../doris/httpv2/rest/StreamingJobAction.java | 2 +-
.../streaming/DataSourceConfigValidator.java | 24 ++++++-
.../insert/streaming/StreamingInsertJob.java | 10 ++-
.../insert/streaming/StreamingJobProperties.java | 10 ++-
.../streaming/StreamingJobSchedulerTask.java | 3 +-
.../insert/streaming/StreamingMultiTblTask.java | 7 +-
.../job/offset/jdbc/JdbcSourceOffsetProvider.java | 20 ++++--
.../apache/doris/job/util/StreamingJobUtils.java | 11 +--
.../trees/plans/commands/AlterJobCommand.java | 2 +-
.../trees/plans/commands/info/CreateJobInfo.java | 3 +-
.../CdcStreamTableValuedFunction.java | 13 ++--
.../StreamingInsertJobCheckDataQualityTest.java | 6 ++
.../streaming/StreamingJobPropertiesTest.java | 3 +-
.../JdbcSourceOffsetProviderErrorHandlingTest.java | 18 +++--
.../doris/cdcclient/sink/DorisBatchStreamLoad.java | 11 +++
.../cdcclient/sink/DorisBatchStreamLoadTest.java | 84 ++++++++++++++++++++++
.../test_streaming_mysql_job_create_alter.groovy | 2 +-
.../cdc/test_streaming_mysql_job_dup.groovy | 2 +-
.../cdc/test_streaming_postgres_job_dup.groovy | 2 +-
.../test_streaming_insert_job_crud.groovy | 8 +--
.../test_streaming_job_max_retry.groovy | 2 +
21 files changed, 202 insertions(+), 41 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/StreamingJobAction.java
b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/StreamingJobAction.java
index 42d243b5754..7eac63ddb77 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/StreamingJobAction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/StreamingJobAction.java
@@ -113,7 +113,7 @@ public class StreamingJobAction extends RestBaseController {
private void checkAuth(HttpServletRequest request) {
String authToken = request.getHeader("token");
if (Strings.isNullOrEmpty(authToken)) {
- throw new UnauthorizedException("Miss token");
+ throw new UnauthorizedException("Missing token");
}
if (!checkClusterToken(authToken)) {
throw new UnauthorizedException("Invalid token");
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidator.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidator.java
index f75bf03c56e..f74feb336ed 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidator.java
@@ -127,7 +127,10 @@ public class DataSourceConfigValidator {
}
if (!isValidValue(key, value, dataSourceType)) {
- throw new IllegalArgumentException("Invalid value for key '" +
key + "': " + value);
+ if (DataSourceConfigKeys.SERVER_ID.equals(key)) {
+ validateServerIdConfig(input);
+ }
+ throw new IllegalArgumentException(invalidValueMessage(key,
value));
}
}
@@ -290,6 +293,25 @@ public class DataSourceConfigValidator {
return true;
}
+ public static String invalidValueMessage(String key, String value) {
+ String message = "Invalid value for key '" + key + "': " + value + ".";
+ if (DataSourceConfigKeys.SLOT_NAME.equals(key)
+ || DataSourceConfigKeys.PUBLICATION_NAME.equals(key)) {
+ return message + " Must match [a-z_][a-z0-9_]* (max 63
characters).";
+ }
+ if (DataSourceConfigKeys.SSL_MODE.equals(key)) {
+ return message + " Use disable, require, or verify-ca.";
+ }
+ if (DataSourceConfigKeys.SNAPSHOT_SPLIT_SIZE.equals(key)
+ || DataSourceConfigKeys.SNAPSHOT_PARALLELISM.equals(key)) {
+ return message + " Use a positive integer.";
+ }
+ if (DataSourceConfigKeys.SKIP_SNAPSHOT_BACKFILL.equals(key)) {
+ return message + " Use true or false.";
+ }
+ return message;
+ }
+
// Strict boolean: only "true"/"false" (case-insensitive);
Boolean.parseBoolean would
// silently coerce typos like "yes" to false.
public static boolean isValidBoolean(String value) {
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 185aa720c4a..13db794b4ca 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
@@ -1212,7 +1212,7 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
private static boolean checkPrivilege(ConnectContext ctx, String sql)
throws AnalysisException {
LogicalPlan logicalPlan = new NereidsParser().parseSingle(sql);
if (!(logicalPlan instanceof InsertIntoTableCommand)) {
- throw new AnalysisException("Only support insert command");
+ throw new AnalysisException("Streaming jobs only support INSERT
statements");
}
LogicalPlan logicalQuery = ((InsertIntoTableCommand)
logicalPlan).getLogicalQuery();
List<String> targetTable =
InsertUtils.getTargetTableQualified(logicalQuery, ctx);
@@ -1538,7 +1538,8 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
&& runningMultiTask.isTimeout(status)) {
String timeoutReason = status == null ? "" :
status.getFailReason();
if (StringUtils.isEmpty(timeoutReason)) {
- timeoutReason = "task failed cause timeout";
+ timeoutReason = "Streaming task " +
runningMultiTask.getTaskId()
+ + " timed out because no progress was reported.";
}
runningMultiTask.onFail(timeoutReason);
// renew streaming task by auto resume
@@ -1636,7 +1637,10 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
getJobId(), ratio, maxFilterRatio,
sampleWindowFilteredRows, sampleWindowScannedRows);
log.error(msg);
FailureReason failureReason = new
FailureReason(InternalErrorCode.TOO_MANY_FAILURE_ROWS_ERR,
- "too many filtered rows exceeded max_filter_ratio " +
maxFilterRatio);
+ String.format(
+ "too many filtered rows: ratio %s exceeds
load.max_filter_ratio %s. "
+ + "Fix the source data or adjust the
limit, then run RESUME JOB.",
+ ratio, maxFilterRatio));
this.setFailureReason(failureReason);
this.updateJobStatus(JobStatus.PAUSED);
throw new JobException(failureReason.getMsg());
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java
index 3db3aecaf8b..1bde1bf6205 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java
@@ -106,18 +106,22 @@ public class StreamingJobProperties implements
JobProperties {
this.maxIntervalSecond = Util.getLongPropertyOrDefault(
properties.get(StreamingJobProperties.MAX_INTERVAL_SECOND_PROPERTY),
StreamingJobProperties.DEFAULT_MAX_INTERVAL_SECOND,
(v) -> v >= 1,
- StreamingJobProperties.MAX_INTERVAL_SECOND_PROPERTY + " should
> 1");
+ StreamingJobProperties.MAX_INTERVAL_SECOND_PROPERTY + " must
be at least 1 second, but was "
+ +
properties.get(StreamingJobProperties.MAX_INTERVAL_SECOND_PROPERTY));
this.s3BatchFiles = Util.getLongPropertyOrDefault(
properties.get(StreamingJobProperties.S3_MAX_BATCH_FILES_PROPERTY),
StreamingJobProperties.DEFAULT_MAX_S3_BATCH_FILES, (v)
-> v >= 1,
- StreamingJobProperties.S3_MAX_BATCH_FILES_PROPERTY + " should
>=1 ");
+ StreamingJobProperties.S3_MAX_BATCH_FILES_PROPERTY + " must be
at least 1, but was "
+ +
properties.get(StreamingJobProperties.S3_MAX_BATCH_FILES_PROPERTY));
this.s3BatchBytes = Util.getLongPropertyOrDefault(
properties.get(StreamingJobProperties.S3_MAX_BATCH_BYTES_PROPERTY),
StreamingJobProperties.DEFAULT_MAX_S3_BATCH_BYTES, (v)
-> v >= 100 * 1024 * 1024
&& v <= (long) (1024 * 1024 * 1024) * 10,
- StreamingJobProperties.S3_MAX_BATCH_BYTES_PROPERTY + " should
between 100MB and 10GB");
+ StreamingJobProperties.S3_MAX_BATCH_BYTES_PROPERTY
+ + " must be between 100 MB and 10 GB, but was "
+ +
properties.get(StreamingJobProperties.S3_MAX_BATCH_BYTES_PROPERTY));
// validate session variables
try {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java
index 030c3bb2b8b..a017b0bb921 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobSchedulerTask.java
@@ -108,7 +108,8 @@ public class StreamingJobSchedulerTask extends AbstractTask
{
streamingInsertJob.setFailureReason(new FailureReason(
InternalErrorCode.CANNOT_RESUME_ERR,
"Auto resume failed after " + autoResumeCount
- + " attempts. Last error: " +
failureReason.getMsg()));
+ + " attempts. Last error: " +
failureReason.getMsg()
+ + ". Run RESUME JOB to retry."));
return;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java
index e652f7c9e1e..f141762bdf8 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java
@@ -163,18 +163,21 @@ public class StreamingMultiTblTask extends
AbstractStreamingTask {
log.info("Send write records request successfully,
response: {}", responseObj.getData());
return;
}
+ String errorMessage =
StringUtils.defaultIfBlank(responseObj.getData(), responseObj.getMsg());
+ throw new JobException(StringUtils.defaultIfBlank(
+ errorMessage, "cdc_client failed to start streaming
write"));
} catch (JsonProcessingException e) {
log.warn("Failed to parse write records response: {}",
response);
throw new JobException("Failed to parse write records
response: " + response);
}
- throw new JobException("Failed to send write records request ,
error message: " + response);
} catch (TimeoutException te) {
log.warn("cdc_client RPC timeout api=/api/writeRecords taskId={}
jobId={} backend={}:{} timeout_sec={}",
taskId, getJobId(), backend.getHost(),
backend.getBrpcPort(),
Config.streaming_cdc_heavy_rpc_timeout_sec);
// the request may have been dispatched and still running remotely
noRetry = true;
- throw new JobException("cdc_client RPC timeout: /api/writeRecords
taskId=" + taskId);
+ throw new JobException("cdc_client RPC timeout: /api/writeRecords
jobId="
+ + getJobId() + " taskId=" + taskId);
} catch (ExecutionException | InterruptedException ex) {
if (ex instanceof InterruptedException) {
Thread.currentThread().interrupt();
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 64752144853..16e4d109159 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
@@ -59,6 +59,7 @@ import lombok.Setter;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
+import org.apache.commons.lang3.StringUtils;
import java.util.ArrayList;
import java.util.Arrays;
@@ -960,9 +961,9 @@ 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.
+ * with a lenient {@link JsonNode} data field first so a failure surfaces
the 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;
@@ -972,6 +973,13 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
throw new JobException(response);
}
if (body.getCode() != RestApiStatusCode.OK.code) {
+ JsonNode data = body.getData();
+ if (data != null && data.isTextual() &&
StringUtils.isNotBlank(data.asText())) {
+ throw new JobException(data.asText());
+ }
+ if (StringUtils.isNotBlank(body.getMsg())) {
+ throw new JobException(body.getMsg());
+ }
throw new JobException(response);
}
try {
@@ -1102,12 +1110,12 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
if (responseObj.getCode() == RestApiStatusCode.OK.code) {
log.info("Init {} source reader successfully, response:
{}", getJobId(), responseObj.getData());
return;
- } else {
- throw new JobException("Failed to init source reader,
error: " + responseObj.getData());
}
+ String errorMessage =
StringUtils.defaultIfBlank(responseObj.getData(), responseObj.getMsg());
+ throw new JobException("Failed to init source reader, error: "
+ errorMessage);
} catch (JobException jobex) {
log.warn("Failed to init {} source reader, {}", getJobId(),
response);
- throw new JobException(jobex.getMessage());
+ throw jobex;
} catch (Exception e) {
log.warn("Failed to init {} source reader, {}", getJobId(),
response);
throw new JobException("Failed to init source reader, cause "
+ e.getMessage());
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java
b/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java
index 30cc98d3caf..af830fa77d3 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java
@@ -383,9 +383,10 @@ public class StreamingJobUtils {
JdbcClient jdbcClient = getJdbcClient(sourceType, properties);
String database = getRemoteDbName(sourceType, properties);
+ String sourceLocation = sourceType == DataSourceType.POSTGRES ?
"schema" : "database";
List<String> tablesNameList = jdbcClient.getTablesNameList(database);
if (tablesNameList.isEmpty()) {
- throw new JobException("No tables found in database " + database);
+ throw new JobException("No source tables found in " +
sourceLocation + " '" + database + "'");
}
Database targetDatabase =
Env.getCurrentEnv().getInternalCatalog().getDbNullable(targetDb);
Preconditions.checkNotNull(targetDatabase, "target database %s does
not exist", targetDb);
@@ -484,8 +485,9 @@ public class StreamingJobUtils {
}
if (!noPrimaryKeyTables.isEmpty()) {
- throw new JobException("The following tables do not have primary
key defined: "
- + String.join(", ", noPrimaryKeyTables));
+ throw new JobException("Source tables require primary keys: "
+ + String.join(", ", noPrimaryKeyTables)
+ + ". Add a primary key or exclude these tables.");
}
return createtblCmds;
}
@@ -514,7 +516,8 @@ public class StreamingJobUtils {
columns.forEach(col -> {
col.setType(getCdcTransportType(col.getType()));
Preconditions.checkArgument(!col.getType().isUnsupported(),
- "Unsupported column type, table:[%s], column:[%s]", table,
col.getName());
+ "Unsupported column type for source column '%s.%s.%s'",
+ database, table, col.getName());
if (col.getType().isVarchar()) {
// The length of varchar needs to be multiplied by 3.
int len = col.getType().getLength() * 3;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java
index d75be51c457..0a529d55424 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java
@@ -338,7 +338,7 @@ public class AlterJobCommand extends AlterCommand
implements ForwardWithSync, Ne
private Pair<List<String>, UnboundTVFRelation> getTargetTableAndTvf(String
sql) throws AnalysisException {
LogicalPlan logicalPlan = new NereidsParser().parseSingle(sql);
if (!(logicalPlan instanceof InsertIntoTableCommand)) {
- throw new AnalysisException("Only support insert command");
+ throw new AnalysisException("Streaming jobs only support INSERT
statements");
}
LogicalPlan logicalQuery = ((InsertIntoTableCommand)
logicalPlan).getLogicalQuery();
List<String> targetTable =
InsertUtils.getTargetTableQualified(logicalQuery, ConnectContext.get());
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateJobInfo.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateJobInfo.java
index 544e649059b..bc0540e5064 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateJobInfo.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateJobInfo.java
@@ -343,8 +343,7 @@ public class CreateJobInfo {
throw new AnalysisException(e.getMessage());
}
} else {
- throw new AnalysisException("Only " +
logicalPlan.getClass().getName()
- + " is supported to use with streaming job together");
+ throw new AnalysisException("Streaming jobs only support INSERT
statements");
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java
index 15f7de54a21..2c22dd7828b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java
@@ -154,11 +154,13 @@ public class CdcStreamTableValuedFunction extends
ExternalFileTableValuedFunctio
}
String offset = properties.get(DataSourceConfigKeys.OFFSET);
if (!DataSourceConfigValidator.isValidOffset(offset,
sourceType.name())) {
- throw new AnalysisException("Invalid value for key 'offset': " +
offset);
+ throw new
AnalysisException(DataSourceConfigValidator.invalidValueMessage(
+ DataSourceConfigKeys.OFFSET, offset));
}
String sslMode = properties.get(DataSourceConfigKeys.SSL_MODE);
if (sslMode != null &&
!DataSourceConfigValidator.isValidSslMode(sslMode)) {
- throw new AnalysisException("Invalid value for key 'ssl_mode': " +
sslMode);
+ throw new
AnalysisException(DataSourceConfigValidator.invalidValueMessage(
+ DataSourceConfigKeys.SSL_MODE, sslMode));
}
try {
DataSourceConfigValidator.validateSslVerifyCaPair(properties);
@@ -184,7 +186,7 @@ public class CdcStreamTableValuedFunction extends
ExternalFileTableValuedFunctio
return;
}
if (!DataSourceConfigValidator.isPositiveInt(value)) {
- throw new AnalysisException("Invalid value for key '" + key + "':
" + value);
+ throw new
AnalysisException(DataSourceConfigValidator.invalidValueMessage(key, value));
}
}
@@ -195,7 +197,7 @@ public class CdcStreamTableValuedFunction extends
ExternalFileTableValuedFunctio
return;
}
if (!DataSourceConfigValidator.isValidPgIdentifier(value)) {
- throw new AnalysisException("Invalid value for key '" + key + "':
" + value);
+ throw new
AnalysisException(DataSourceConfigValidator.invalidValueMessage(key, value));
}
}
@@ -206,7 +208,8 @@ public class CdcStreamTableValuedFunction extends
ExternalFileTableValuedFunctio
return;
}
if (!DataSourceConfigValidator.isValidBoolean(value)) {
- throw new AnalysisException("Invalid value for key '" + key + "':
" + value);
+ throw new AnalysisException("Invalid value for key '" + key + "':
" + value
+ + ". Expected true or false.");
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobCheckDataQualityTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobCheckDataQualityTest.java
index 06c5e1a5c75..8ca602f7527 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobCheckDataQualityTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobCheckDataQualityTest.java
@@ -17,10 +17,12 @@
package org.apache.doris.job.extensions.insert.streaming;
+import org.apache.doris.common.InternalErrorCode;
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.request.CommitOffsetRequest;
+import org.apache.doris.job.common.FailureReason;
import org.apache.doris.job.common.JobStatus;
import org.apache.doris.job.exception.JobException;
@@ -93,6 +95,10 @@ public class StreamingInsertJobCheckDataQualityTest {
thrown = e;
}
Assert.assertNotNull("expected pause when combined ratio exceeds
threshold", thrown);
+
Assert.assertTrue(thrown.getMessage().contains("load.max_filter_ratio"));
+ Assert.assertTrue(thrown.getMessage().contains("RESUME JOB"));
+ Assert.assertEquals(InternalErrorCode.TOO_MANY_FAILURE_ROWS_ERR,
+ new FailureReason(thrown.getMessage()).getCode());
Assert.assertEquals(JobStatus.PAUSED, job.getJobStatus());
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java
index 97a98031dde..ca5098e1b24 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java
@@ -104,7 +104,8 @@ public class StreamingJobPropertiesTest {
Assert.assertEquals(StreamingJobProperties.DEFAULT_MAX_INTERVAL_SECOND,
p.getMaxIntervalSecond());
// but validate() should throw
- Assert.assertThrows(AnalysisException.class, p::validate);
+ AnalysisException exception =
Assert.assertThrows(AnalysisException.class, p::validate);
+ Assert.assertTrue(exception.getMessage().contains("max_interval must
be at least 1 second"));
}
@Test
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
index 1da12b3c97a..e0d6827300b 100644
---
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
@@ -56,8 +56,19 @@ public class JdbcSourceOffsetProviderErrorHandlingTest {
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));
+ Assert.assertEquals(realError, e.getMessage());
+ }
+ }
+
+ @Test
+ public void testParseFailureEnvelopeFallsBackToMessageWithoutTextData() {
+ JdbcSourceOffsetProvider provider = new JdbcSourceOffsetProvider();
+ String response = "{\"code\":1,\"msg\":\"cdc_client is
unavailable\",\"data\":null}";
+ try {
+ provider.parseCdcResponseData(response, new
TypeReference<List<SnapshotSplit>>() {});
+ Assert.fail("a failed envelope must throw");
+ } catch (JobException e) {
+ Assert.assertEquals("cdc_client is unavailable", e.getMessage());
}
}
@@ -83,8 +94,7 @@ public class JdbcSourceOffsetProviderErrorHandlingTest {
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"));
+ Assert.assertEquals(response, e.getMessage());
}
}
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/sink/DorisBatchStreamLoad.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/sink/DorisBatchStreamLoad.java
index dbc68c680a9..370e3c30ee9 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/sink/DorisBatchStreamLoad.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/sink/DorisBatchStreamLoad.java
@@ -572,6 +572,17 @@ public class DorisBatchStreamLoad implements Serializable {
taskId);
return;
}
+ JsonNode data = root.get("data");
+ JsonNode msg = root.get("msg");
+ if (data != null
+ && data.isTextual()
+ && StringUtils.isNotBlank(data.asText())) {
+ reason = data.asText();
+ } else if (msg != null
+ && msg.isTextual()
+ && StringUtils.isNotBlank(msg.asText())) {
+ reason = msg.asText();
+ }
}
LOG.error(
"commit offset failed with {}, reason {}, to
retry",
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/sink/DorisBatchStreamLoadTest.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/sink/DorisBatchStreamLoadTest.java
new file mode 100644
index 00000000000..7c271f046ad
--- /dev/null
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/sink/DorisBatchStreamLoadTest.java
@@ -0,0 +1,84 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.cdcclient.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.apache.doris.cdcclient.exception.StreamLoadException;
+
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.sun.net.httpserver.HttpServer;
+
+class DorisBatchStreamLoadTest {
+
+ @ParameterizedTest
+ @CsvSource(
+ delimiter = '|',
+ value = {
+ "{\"code\":1,\"msg\":\"Error\",\"data\":\"Offset commit
rejected\"}|Offset commit rejected",
+ "{\"code\":1,\"msg\":\"Offset commit
rejected\",\"data\":null}|Offset commit rejected",
+ "{\"code\":1,\"msg\":\"\",\"data\":null}|HTTP/1.1 200 OK"
+ })
+ void commitOffsetPreservesFailureReasonAfterRetries(String response,
String expectedReason)
+ throws Exception {
+ HttpServer server = HttpServer.create(new
InetSocketAddress("127.0.0.1", 0), 0);
+ AtomicInteger requests = new AtomicInteger();
+ server.createContext(
+ "/api/streaming/commit_offset",
+ exchange -> {
+ try {
+ exchange.getRequestBody().readAllBytes();
+ requests.incrementAndGet();
+ byte[] body =
response.getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, body.length);
+ exchange.getResponseBody().write(body);
+ } finally {
+ exchange.close();
+ }
+ });
+ server.start();
+ try {
+ DorisBatchStreamLoad streamLoad = new DorisBatchStreamLoad("1",
"test_db");
+ try {
+ streamLoad.setFrontendAddress("127.0.0.1:" +
server.getAddress().getPort());
+ streamLoad.setToken("test-token");
+ StreamLoadException error = assertThrows(
+ StreamLoadException.class,
+ () -> streamLoad.commitOffset(
+ "2", Collections.emptyList(), 0, new
LoadStatistic(), null));
+ assertEquals(
+ "StreamLoadException: commit offset failed with: " +
expectedReason,
+ ExceptionUtils.getRootCauseMessage(error));
+ assertEquals(4, requests.get());
+ } finally {
+ streamLoad.close();
+ }
+ } finally {
+ server.stop(0);
+ }
+ }
+}
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_create_alter.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_create_alter.groovy
index 2e6443db633..30e3af99afe 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_create_alter.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_create_alter.groovy
@@ -181,7 +181,7 @@ suite("test_streaming_mysql_job_create_alter",
"p0,external,mysql,external_docke
"table.create.properties.replication_num" = "1"
)
"""
- exception "No tables found in database"
+ exception "No source tables found in database"
}
// no match table
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_dup.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_dup.groovy
index 2ddacebde16..ac9c8631280 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_dup.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_dup.groovy
@@ -61,7 +61,7 @@ suite("test_streaming_mysql_job_dup",
"p0,external,mysql,external_docker,externa
"table.create.properties.replication_num" = "1"
)
"""
- exception "The following tables do not have primary key defined:
${table1}"
+ exception "Source tables require primary keys: ${table1}"
}
def jobInfo = sql """
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_dup.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_dup.groovy
index 1bec6cd3a25..cbaab4f153a 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_dup.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_dup.groovy
@@ -64,7 +64,7 @@ suite("test_streaming_postgres_job_dup",
"p0,external,pg,external_docker,externa
"table.create.properties.replication_num" = "1"
)
"""
- exception "The following tables do not have primary key defined:
${table1}"
+ exception "Source tables require primary keys: ${table1}"
}
def jobInfo = sql """
diff --git
a/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_crud.groovy
b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_crud.groovy
index f648767fdaf..8de7f1b2748 100644
---
a/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_crud.groovy
+++
b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_crud.groovy
@@ -112,7 +112,7 @@ suite("test_streaming_insert_job_crud") {
"s3.secret_key" = "${getS3SK()}"
);
"""
- }, "s3.max_batch_files should >=1")
+ }, "s3.max_batch_files must be at least 1")
jobCount = sql """ select count(1) from jobs("type"="insert") where Name =
'${jobNameError}' and ExecuteType='STREAMING' """
assert jobCount.get(0).get(0) == 0
@@ -136,7 +136,7 @@ suite("test_streaming_insert_job_crud") {
"s3.secret_key" = "${getS3SK()}"
);
"""
- }, "s3.max_batch_bytes should between 100MB and 10GB")
+ }, "s3.max_batch_bytes must be between 100 MB and 10 GB")
jobCount = sql """ select count(1) from jobs("type"="insert") where Name =
'${jobNameError}' and ExecuteType='STREAMING' """
assert jobCount.get(0).get(0) == 0
@@ -160,7 +160,7 @@ suite("test_streaming_insert_job_crud") {
"s3.secret_key" = "${getS3SK()}"
);
"""
- }, "max_interval should > 1")
+ }, "max_interval must be at least 1 second")
jobCount = sql """ select count(1) from jobs("type"="insert") where Name =
'${jobNameError}' and ExecuteType='STREAMING' """
assert jobCount.get(0).get(0) == 0
@@ -409,7 +409,7 @@ suite("test_streaming_insert_job_crud") {
"s3.secret_key" = "${getS3SK()}"
);
"""
- exception "s3.max_batch_files should >=1"
+ exception "s3.max_batch_files must be at least 1"
}
// alter session var
diff --git
a/regression-test/suites/job_p0/streaming_job/test_streaming_job_max_retry.groovy
b/regression-test/suites/job_p0/streaming_job/test_streaming_job_max_retry.groovy
index 30fc6eba1c1..b137198e174 100644
---
a/regression-test/suites/job_p0/streaming_job/test_streaming_job_max_retry.groovy
+++
b/regression-test/suites/job_p0/streaming_job/test_streaming_job_max_retry.groovy
@@ -101,6 +101,8 @@ suite("test_streaming_job_max_retry", "nonConcurrent") {
"ErrorMsg should contain CANNOT_RESUME_ERR code, got: " +
errorMsgJson
assert errorMsgJson.contains("Auto resume failed after"),
"ErrorMsg should contain the burn-out message, got: " +
errorMsgJson
+ assert errorMsgJson.contains("RESUME JOB"),
+ "ErrorMsg should explain the manual recovery action, got: " +
errorMsgJson
} finally {
GetDebugPoint().disableDebugPointForAllFEs('StreamingJob.scheduleTask.exception')
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
