This is an automated email from the ASF dual-hosted git repository. yiguolei pushed a commit to branch branch-4.2 in repository https://gitbox.apache.org/repos/asf/doris.git
commit 10dd002771c7423bee663b7cec21a5d049a59ba7 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]
