[improve](streaming-job) make from-to streaming task timeout progress-aware

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -1193,6 +1193,12 @@ public class Config extends ConfigBase {
  @ConfField(mutable = true, masterOnly = true)
  public static int streaming_task_timeout_multiplier = 10;
 
+ /**
+ * streaming task min timeout second.
+ */
+ @ConfField(mutable = true, masterOnly = true)
+ public static int streaming_task_min_timeout_sec = 300;
+
  @ConfField(mutable = true, masterOnly = true)
  public static int streaming_cdc_light_rpc_timeout_sec = 90;
 

fe/fe-common/src/main/java/org/apache/doris/job/cdc/StreamingTaskStatus.java

@@ -0,0 +1,50 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.cdc;
+
+// cdc_client -> FE running task status, pulled by the FE only when the local timeout
+// budget is already exceeded. scannedRows is the read-end heartbeat used to renew the
+// deadline (scannedRows < 0 means no progress info: not scanning, or scan finished);
+// failReason carries the recorded write error so a kill can report the real cause.
+public class StreamingTaskStatus {
+ private long scannedRows = -1;
+ private String failReason = "";
+
+ public StreamingTaskStatus() {}
+
+ public StreamingTaskStatus(long scannedRows, String failReason) {
+ this.scannedRows = scannedRows;
+ this.failReason = failReason;
+ }
+
+ public long getScannedRows() {
+ return scannedRows;
+ }
+
+ public void setScannedRows(long scannedRows) {
+ this.scannedRows = scannedRows;
+ }
+
+ public String getFailReason() {
+ return failReason;
+ }
+
+ public void setFailReason(String failReason) {
+ this.failReason = failReason;
+ }
+}

fe/fe-common/src/main/java/org/apache/doris/job/cdc/request/TaskFailureRequest.java

@@ -0,0 +1,46 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.cdc.request;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+// cdc_client -> FE push: report a hard write failure of a running task immediately,
+// so it is failed within seconds instead of waiting out the timeout budget.
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+@Builder
+public class TaskFailureRequest {
+ public long jobId;
+ public long taskId;
+ public String reason;
+
+ @Override
+ public String toString() {
+ return "TaskFailureRequest{"
+ + "jobId=" + jobId
+ + ", taskId=" + taskId
+ + ", reason='" + reason + "'"
+ + "}";
+ }
+}

fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/StreamingJobAction.java

@@ -22,6 +22,7 @@
 import org.apache.doris.httpv2.exception.UnauthorizedException;
 import org.apache.doris.job.base.AbstractJob;
 import org.apache.doris.job.cdc.request.CommitOffsetRequest;
+import org.apache.doris.job.cdc.request.TaskFailureRequest;
 import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob;
 
 import com.google.common.base.Strings;
@@ -39,17 +40,40 @@ public class StreamingJobAction extends RestBaseController {
 
  @RequestMapping(path = "/api/streaming/commit_offset", method = RequestMethod.PUT)
  public Object commitOffset(@RequestBody CommitOffsetRequest offsetRequest, HttpServletRequest request) {
+ checkAuth(request);
+ return updateOffset(offsetRequest);
+ }
+
+ @RequestMapping(path = "/api/streaming/report_task_failure", method = RequestMethod.PUT)
+ public Object reportTaskFailure(@RequestBody TaskFailureRequest failureRequest, HttpServletRequest request) {
+ checkAuth(request);
+ return failTask(failureRequest);
+ }
+
+ private void checkAuth(HttpServletRequest request) {
  String authToken = request.getHeader("token");
- // if auth token is not null, check it first
- if (!Strings.isNullOrEmpty(authToken)) {
- if (!checkClusterToken(authToken)) {
- throw new UnauthorizedException("Invalid token: " + authToken);
- }
- return updateOffset(offsetRequest);
- } else {
- // only use for token
+ if (Strings.isNullOrEmpty(authToken)) {
  throw new UnauthorizedException("Miss token");
  }
+ if (!checkClusterToken(authToken)) {
+ throw new UnauthorizedException("Invalid token: " + authToken);
+ }
+ }
+
+ private Object failTask(TaskFailureRequest failureRequest) {
+ AbstractJob job = Env.getCurrentEnv().getJobManager().getJob(failureRequest.getJobId());
+ if (!(job instanceof StreamingInsertJob)) {
+ return ResponseEntityBuilder
+ .okWithCommonError("Job " + failureRequest.getJobId() + " is not a streaming job");
+ }
+ try {
+ LOG.info("Reporting task failure with {}", failureRequest.toString());
+ ((StreamingInsertJob) job).reportTaskFailure(failureRequest);
+ return ResponseEntityBuilder.ok("Task failure reported successfully");
+ } catch (Exception e) {
+ LOG.warn("Failed to report task failure for job {}: {}", failureRequest.getJobId(), e.getMessage());
+ return ResponseEntityBuilder.okWithCommonError(e.getMessage());
+ }
  }
 
  private Object updateOffset(CommitOffsetRequest offsetRequest) {

fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java

@@ -39,7 +39,9 @@
 import org.apache.doris.job.base.JobExecutionConfiguration;
 import org.apache.doris.job.base.TimerDefinition;
 import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.StreamingTaskStatus;
 import org.apache.doris.job.cdc.request.CommitOffsetRequest;
+import org.apache.doris.job.cdc.request.TaskFailureRequest;
 import org.apache.doris.job.common.DataSourceType;
 import org.apache.doris.job.common.FailureReason;
 import org.apache.doris.job.common.IntervalUnit;
@@ -139,7 +141,7 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M
  private String tvfType;
  private Map<String, String> originTvfProps;
  @Getter
- AbstractStreamingTask runningStreamTask;
+volatile AbstractStreamingTask runningStreamTask;
  SourceOffsetProvider offsetProvider;
  @Getter
  @Setter
@@ -1404,20 +1406,51 @@ public void gsonPostProcess() throws IOException {
  }
  }
 
+ /**
+ * Push from cdc_client: fail the running task immediately on a hard write failure.
+ * Reports whose taskId no longer matches the current running task are dropped.
+ */
+ public void reportTaskFailure(TaskFailureRequest request) throws JobException {
+ AbstractStreamingTask task = this.runningStreamTask;
+ if (!(task instanceof StreamingMultiTblTask)) {
+ return;
+ }
+ StreamingMultiTblTask runningMultiTask = (StreamingMultiTblTask) task;
+ if (runningMultiTask.getTaskId() != request.getTaskId()) {
+ return;
+ }
+ writeLock();
+ try {
+ if (this.runningStreamTask == runningMultiTask
+ && runningMultiTask.getTaskId() == request.getTaskId()
+ && TaskStatus.RUNNING.equals(runningMultiTask.getStatus())) {
+ runningMultiTask.onFail(request.getReason());
+ }
+ } finally {
+ writeUnlock();
+ }
+ }
+
  /**
  * The current streamingTask times out; create a new streamingTask.
  * Only applies to StreamingMultiTask.
  */
  public void processTimeoutTasks() throws JobException {
- if (!(runningStreamTask instanceof StreamingMultiTblTask)) {
+ AbstractStreamingTask task = this.runningStreamTask;
+ if (!(task instanceof StreamingMultiTblTask)) {
+ return;
+ }
+ StreamingMultiTblTask runningMultiTask = (StreamingMultiTblTask) task;
+ if (!runningMultiTask.isLocalTimeout()) {
  return;
  }
+ StreamingTaskStatus status = runningMultiTask.fetchTaskStatus();
  writeLock();
  try {
-StreamingMultiTblTask runningMultiTask = (StreamingMultiTblTask) this.runningStreamTask;
-if (TaskStatus.RUNNING.equals(runningMultiTask.getStatus())
- && runningMultiTask.isTimeout()) {
- String timeoutReason = runningMultiTask.getTimeoutReason();
+if (this.runningStreamTask == runningMultiTask
+ && TaskStatus.RUNNING.equals(runningMultiTask.getStatus())
+ && runningMultiTask.isTimeout(status)) {
+ String timeoutReason = status == null ? "" : status.getFailReason();
  if (StringUtils.isEmpty(timeoutReason)) {
  timeoutReason = "task failed cause timeout";
  }

fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java

@@ -25,6 +25,7 @@
 import org.apache.doris.httpv2.rest.RestApiStatusCode;
 import org.apache.doris.job.base.Job;
 import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.StreamingTaskStatus;
 import org.apache.doris.job.cdc.request.CommitOffsetRequest;
 import org.apache.doris.job.cdc.request.JobBaseConfig;
 import org.apache.doris.job.cdc.request.WriteRecordRequest;
@@ -83,7 +84,9 @@ public class StreamingMultiTblTask extends AbstractStreamingTask {
  private long loadBytes = 0L;
  private long filteredRows = 0L;
  private long loadedRows = 0L;
- private long runningBackendId;
+ private volatile long runningBackendId;
+ long lastScannedRows = -1;
+ long lastProgressMs = 0;
 
  public StreamingMultiTblTask(Long jobId,
  long taskId,
@@ -111,8 +114,9 @@ public void before() throws Exception {
  log.info("streaming multi task has been canceled, task id is {}", getTaskId());
  return;
  }
- this.status = TaskStatus.RUNNING;
  this.startTimeMs = System.currentTimeMillis();
+ this.lastProgressMs = this.startTimeMs;
+ this.status = TaskStatus.RUNNING;
  this.runningOffset = offsetProvider.getNextOffset(null, sourceProperties);
  log.info("streaming multi task {} get running offset: {}", taskId, runningOffset.toString());
  }
@@ -361,72 +365,69 @@ private String getFrontendAddress() {
  return Env.getCurrentEnv().getMasterHost() + ":" + Env.getCurrentEnv().getMasterHttpPort();
  }
 
- public boolean isTimeout() {
+ // Local pre-check, no RPC: gates whether to pull real progress this tick.
+ boolean isLocalTimeout() {
+ if (startTimeMs == null) {
+ return false;
+ }
+ return System.currentTimeMillis() - lastProgressMs > getTaskTimeoutMs();
+ }
+
+ boolean isTimeout(StreamingTaskStatus status) {
  if (startTimeMs == null) {
  // It's still pending, waiting for scheduling.
  return false;
  }
+ long now = System.currentTimeMillis();
+ if (status != null && status.getScannedRows() > lastScannedRows) {
+ lastScannedRows = status.getScannedRows();
+ lastProgressMs = now;
+ }
  long timeoutMs = getTaskTimeoutMs();
- long elapsed = System.currentTimeMillis() - startTimeMs;
+ long elapsed = now - lastProgressMs;
  if (elapsed > timeoutMs) {
- log.info("Task {} timeout detected: elapsed={}ms, timeoutMs={}ms", taskId, elapsed, timeoutMs);
+ log.info("Task {} timeout detected: no progress for {}ms, timeoutMs={}ms",
+ taskId, elapsed, timeoutMs);
  return true;
  }
  return false;
  }
 
  // Read multiplier live so config changes affect already-running tasks.
  private long getTaskTimeoutMs() {
- return Config.streaming_task_timeout_multiplier * jobProperties.getMaxIntervalSecond() * 1000L;
+ return Math.max(
+ Config.streaming_task_timeout_multiplier * jobProperties.getMaxIntervalSecond() * 1000L,
+ Config.streaming_task_min_timeout_sec * 1000L);
  }
 
- /**
- * When a task encounters a write error, it will time out.
- * The job needs to obtain the reason for the timeout,
- * such as a data quality error, and needs to expose it to the user.
- */
- public String getTimeoutReason() {
+ StreamingTaskStatus fetchTaskStatus() {
  if (runningBackendId <= 0) {
- log.info("No running backend for task {}", runningBackendId);
- return "";
+ return null;
  }
  Backend backend = Env.getCurrentSystemInfo().getBackend(runningBackendId);
+ if (backend == null) {
+ return null;
+ }
  try {
  InternalService.PRequestCdcClientRequest request = InternalService.PRequestCdcClientRequest.newBuilder()
- .setApi("/api/getFailReason/" + getTaskId())
+ .setApi("/api/getTaskStatus/" + getTaskId())
  .build();
  TNetworkAddress address = new TNetworkAddress(backend.getHost(), backend.getBrpcPort());
- InternalService.PRequestCdcClientResult result = null;
  Future<PRequestCdcClientResult> future = BackendServiceProxy.getInstance()
  .requestCdcClient(address, request, Config.streaming_cdc_light_rpc_timeout_sec);
- result = future.get(Config.streaming_cdc_light_rpc_timeout_sec, TimeUnit.SECONDS);
- TStatusCode code = TStatusCode.findByValue(result.getStatus().getStatusCode());
- if (code != TStatusCode.OK) {
- log.warn("Failed to get task timeout reason, {}", result.getStatus().getErrorMsgs(0));
- return "";
+ PRequestCdcClientResult result = future.get(Config.streaming_cdc_light_rpc_timeout_sec, TimeUnit.SECONDS);
+ if (TStatusCode.findByValue(result.getStatus().getStatusCode()) != TStatusCode.OK) {
+ return null;
  }
- String response = result.getResponse();
- try {
- ResponseBody<String> responseObj = objectMapper.readValue(
- response,
- new TypeReference<ResponseBody<String>>() {
- }
- );
- if (responseObj.getCode() == RestApiStatusCode.OK.code) {
- return responseObj.getData();
- }
- } catch (JsonProcessingException e) {
- log.warn("Failed to get task timeout reason, response: {}", response);
- }
- } catch (TimeoutException te) {
- log.warn("cdc_client RPC timeout api=/api/getFailReason jobId={} taskId={} backend={}:{} "
- + "timeout_sec={}",
- getJobId(), getTaskId(), backend.getHost(), backend.getBrpcPort(),
- Config.streaming_cdc_light_rpc_timeout_sec);
- } catch (ExecutionException | InterruptedException ex) {
- log.warn("Send get task fail reason request failed: ", ex);
+ ResponseBody<StreamingTaskStatus> body = objectMapper.readValue(
+ result.getResponse(),
+ new TypeReference<ResponseBody<StreamingTaskStatus>>() {
+ });
+ return body.getCode() == RestApiStatusCode.OK.code ? body.getData() : null;
+ } catch (Exception e) {
+ log.warn("fetch task status failed, job {} task {}", getJobId(), getTaskId(), e);
+ return null;
  }
- return "";
  }
 
  @Override