[improvement](streamingjob) change streaming job source log lag in bytes

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

@@ -0,0 +1,43 @@
+// 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.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+
+import java.util.Collections;
+import java.util.Map;
+
+@Data
+@NoArgsConstructor
+@EqualsAndHashCode(callSuper = true)
+public class FetchEndOffsetRequest extends JobBaseConfig {
+ private Map<String, String> referenceOffset;
+
+ public FetchEndOffsetRequest(
+ String jobId,
+ String dataSource,
+ Map<String, String> config,
+ String frontendAddress,
+ Map<String, String> referenceOffset) {
+ super(jobId, dataSource, config, frontendAddress);
+ // Its presence tells the CDC client that this FE accepts the lag-aware response.
+ this.referenceOffset = referenceOffset == null ? Collections.emptyMap() : referenceOffset;
+ }
+}

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

@@ -0,0 +1,32 @@
+// 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.response;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.Map;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class FetchEndOffsetResult {
+ private Map<String, String> endOffset;
+ private long lagBytes = -1;
+}

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

@@ -105,7 +105,8 @@ public class InsertJob extends AbstractJob<InsertTask, Map<Object, Object>> impl
  .add(new Column("LoadStatistic", ScalarType.createStringType()))
  .add(new Column("ErrorMsg", ScalarType.createStringType()))
  .add(new Column("JobRuntimeMsg", ScalarType.createStringType()))
- .add(new Column("Lag", ScalarType.createStringType()))
+ .add(new Column("LagBytes", ScalarType.createStringType()))
+ .add(new Column("LastSourceEventTimestamp", ScalarType.createStringType()))
  .add(new Column("LastTaskSuccessTime", ScalarType.createStringType()))
  .build();
 
@@ -572,6 +573,7 @@ public TRow getTvfInfo() {
  trow.addToColumnValue(new TCell().setStringVal(failMsg == null ? FeConstants.null_string : failMsg.getMsg()));
  trow.addToColumnValue(new TCell().setStringVal(FeConstants.null_string));
  trow.addToColumnValue(new TCell().setStringVal(FeConstants.null_string));
+ trow.addToColumnValue(new TCell().setStringVal(FeConstants.null_string));
  trow.addToColumnValue(new TCell().setStringVal(lastTaskSuccessTime > 0
  ? TimeUtils.longToTimeString(lastTaskSuccessTime) : FeConstants.null_string));
  return trow;

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

@@ -1000,21 +1000,19 @@ public void onReplayCreate() throws JobException {
  }
 
  public String getLag() {
- return offsetProvider != null ? offsetProvider.getLag() : "";
+ return offsetProvider != null ? offsetProvider.getLag() : "-1";
  }
 
- // Numeric lag for metrics. Returns -1 when lag is not applicable (S3, snapshot phase)
- // or unparseable, so dashboards can filter N/A jobs via lag >= 0.
- public long getLagSeconds() {
- String lagStr = getLag();
- if (lagStr == null || lagStr.isEmpty()) {
- return -1L;
- }
- try {
- return Long.parseLong(lagStr);
- } catch (NumberFormatException e) {
- return -1L;
- }
+ public long getLagBytes() {
+ return offsetProvider != null ? offsetProvider.getLagBytes() : -1;
+ }
+
+ public long getLastSourceEventTimestampSeconds() {
+ return offsetProvider != null ? offsetProvider.getLastSourceEventTimestampSeconds() : 0;
+ }
+
+ public long getLastTaskSuccessTimeSeconds() {
+ return lastTaskSuccessTime / 1000L;
  }
 
  /**
@@ -1077,6 +1075,7 @@ private void modifyPropertiesInternal(Map<String, String> inputProperties) throw
  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());
@@ -1174,8 +1173,10 @@ public TRow getTvfInfo() {
  ? "" : GsonUtils.GSON.toJson(failureReason)));
  trow.addToColumnValue(new TCell().setStringVal(jobRuntimeMsg == null
  ? "" : jobRuntimeMsg));
- trow.addToColumnValue(new TCell().setStringVal(
- offsetProvider != null ? offsetProvider.getLag() : ""));
+ trow.addToColumnValue(new TCell().setStringVal(getLag()));
+ long lastSourceEventTimestampSeconds = getLastSourceEventTimestampSeconds();
+ trow.addToColumnValue(new TCell().setStringVal(lastSourceEventTimestampSeconds > 0
+ ? String.valueOf(lastSourceEventTimestampSeconds) : ""));
  trow.addToColumnValue(new TCell().setStringVal(lastTaskSuccessTime > 0
  ? TimeUtils.longToTimeString(lastTaskSuccessTime) : ""));
  return trow;

fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java

@@ -206,15 +206,22 @@ default int pendingSplitCount() {
  return 0;
  }
 
- /**
- * Get the lag of the data source in seconds.
- * For CDC sources, lag = (now - last consumed event timestamp) in seconds.
- *
- * @return lag in seconds as string, empty string if not applicable
- */
+ /** Get the latest successfully observed source-log lag in bytes, or -1 before any observation. */
+ default long getLagBytes() {
+ return -1;
+ }
+
+ /** Get the source event timestamp at the committed offset as Unix seconds, or 0 if unavailable. */
+ default long getLastSourceEventTimestampSeconds() {
+ return 0;
+ }
+
+ /** Discard a lag value that was calculated from an offset explicitly replaced by the user. */
+ default void resetLag() {}
+
+ /** Get source lag as a numeric string for SHOW output. */
  default String getLag() {
- return "";
+ return String.valueOf(getLagBytes());
  }
 
 }
-

fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java

@@ -24,8 +24,10 @@
 import org.apache.doris.httpv2.rest.RestApiStatusCode;
 import org.apache.doris.job.cdc.DataSourceConfigKeys;
 import org.apache.doris.job.cdc.request.CompareOffsetRequest;
+import org.apache.doris.job.cdc.request.FetchEndOffsetRequest;
 import org.apache.doris.job.cdc.request.FetchTableSplitsRequest;
 import org.apache.doris.job.cdc.request.JobBaseConfig;
+import org.apache.doris.job.cdc.response.FetchEndOffsetResult;
 import org.apache.doris.job.cdc.split.AbstractSourceSplit;
 import org.apache.doris.job.cdc.split.BinlogSplit;
 import org.apache.doris.job.cdc.split.SnapshotSplit;
@@ -106,6 +108,8 @@ public class JdbcSourceOffsetProvider implements SourceOffsetProvider {
 
  volatile boolean hasMoreData = true;
 
+ transient volatile long lagBytes = -1;
+
  transient volatile String cloudCluster;
 
  // Route fetchEndOffset/compareOffset to the bound BE (synced from job, not persisted).
@@ -304,8 +308,13 @@ public void setBoundBackendId(long boundBackendId) {
  @Override
  public void fetchRemoteMeta(Map<String, String> properties) throws Exception {
  Backend backend = StreamingJobUtils.selectBackend(cloudCluster, boundBackendId);
- JobBaseConfig requestParams =
- new JobBaseConfig(getJobId().toString(), sourceType.name(), sourceProperties, getFrontendAddress());
+ FetchEndOffsetRequest requestParams =
+ new FetchEndOffsetRequest(
+ getJobId().toString(),
+ sourceType.name(),
+ sourceProperties,
+ getFrontendAddress(),
+ getLagReferenceOffset());
  InternalService.PRequestCdcClientRequest request = InternalService.PRequestCdcClientRequest.newBuilder()
  .setApi("/api/fetchEndOffset")
  .setParams(new Gson().toJson(requestParams)).build();
@@ -322,14 +331,17 @@ public void fetchRemoteMeta(Map<String, String> properties) throws Exception {
  "Failed to get end offset from backend," + result.getStatus().getErrorMsgs(0) + ", response: "
  + result.getResponse());
  }
-Map<String, String> newEndOffset = parseCdcResponseData(
- result.getResponse(), new TypeReference<Map<String, String>>() {});
+FetchEndOffsetResult fetchResult = parseFetchEndOffsetResponse(result.getResponse());
+ Map<String, String> newEndOffset = fetchResult.getEndOffset();
  synchronized (splitsLock) {
  // null→value also counts as a change: upstream may have advanced while fetch was blocked.
  if (endBinlogOffset == null || !endBinlogOffset.equals(newEndOffset)) {
  hasMoreData = true;
  }
  endBinlogOffset = newEndOffset;
+ if (!isSnapshotOnlyMode()) {
+ updateLagBytes(fetchResult.getLagBytes());
+ }
  }
  } catch (TimeoutException te) {
  log.warn("cdc_client RPC timeout api=/api/fetchEndOffset jobId={} backend={}:{} timeout_sec={}",
@@ -342,6 +354,30 @@ public void fetchRemoteMeta(Map<String, String> properties) throws Exception {
  }
  }
 
+ Map<String, String> getLagReferenceOffset() {
+ if (isSnapshotOnlyMode()) {
+ return null;
+ }
+ synchronized (splitsLock) {
+ if (currentOffset != null && !currentOffset.snapshotSplit()) {
+ BinlogSplit binlogSplit = (BinlogSplit) currentOffset.getSplits().get(0);
+ if (MapUtils.isNotEmpty(binlogSplit.getStartingOffset())) {
+ return new HashMap<>(binlogSplit.getStartingOffset());
+ }
+ }
+ if (sourceType == DataSourceType.POSTGRES) {
+ // PostgreSQL can use the replication slot's confirmed flush LSN during snapshot.
+ return null;
+ }
+ return finishedSplits.stream()
+ .map(SnapshotSplit::getHighWatermark)
+ .filter(MapUtils::isNotEmpty)
+ .findFirst()
+ .map(HashMap::new)
+ .orElse(null);
+ }
+ }
+
  @Override
  public boolean hasMoreDataToConsume() {
  if (currentOffset == null) {
@@ -945,6 +981,23 @@ <T> T parseCdcResponseData(String response, TypeReference<T> dataType) throws Jo
  }
  }
 
+ FetchEndOffsetResult parseFetchEndOffsetResponse(String response) throws JobException {
+ JsonNode data = parseCdcResponseData(response, new TypeReference<JsonNode>() {});
+ if (data == null) {
+ throw new JobException(response);
+ }
+ try {
+ if (data.has("endOffset")) {
+ return objectMapper.convertValue(data, FetchEndOffsetResult.class);
+ }
+ Map<String, String> endOffset = objectMapper.convertValue(
+ data, new TypeReference<Map<String, String>>() {});
+ return new FetchEndOffsetResult(endOffset, -1);
+ } catch (IllegalArgumentException exception) {
+ throw new JobException(response);
+ }
+ }
+
  protected boolean checkNeedSplitChunks(Map<String, String> sourceProperties) {
  String startMode = sourceProperties.get(DataSourceConfigKeys.OFFSET);
  if (startMode == null) {
@@ -960,45 +1013,46 @@ protected boolean isSnapshotOnlyMode() {
  }
 
  @Override
- public String getLag() {
- if (currentOffset == null || currentOffset.snapshotSplit()) {
- return "";
- }
- // Source is idle (last task consumed no data), report zero lag
- if (!hasMoreData) {
- return "0";
- }
- BinlogSplit binlogSplit = (BinlogSplit) currentOffset.getSplits().get(0);
- Map<String, String> offsetMap = binlogSplit.getStartingOffset();
- if (MapUtils.isEmpty(offsetMap)) {
- return "";
- }
- long eventTimeMs = extractEventTimeMs(offsetMap);
- if (eventTimeMs <= 0) {
- return "0";
- }
- long lagSec = (System.currentTimeMillis() - eventTimeMs) / 1000;
- return String.valueOf(Math.max(lagSec, 0));
+ public long getLagBytes() {
+ return lagBytes;
  }
 
- /**
- * Extract event timestamp in milliseconds from binlog offset map.
- * MySQL: ts_sec (seconds), PostgreSQL: ts_usec (microseconds).
- */
- protected long extractEventTimeMs(Map<String, String> offsetMap) {
- try {
- String tsSec = offsetMap.get("ts_sec");
- if (tsSec != null) {
- return Long.parseLong(tsSec) * 1000;
+ @Override
+ public long getLastSourceEventTimestampSeconds() {
+ synchronized (splitsLock) {
+ if (currentOffset == null || currentOffset.snapshotSplit()) {
+ return 0;
  }
- String tsUsec = offsetMap.get("ts_usec");
- if (tsUsec != null) {
- return Long.parseLong(tsUsec) / 1000;
+ BinlogSplit binlogSplit = (BinlogSplit) currentOffset.getSplits().get(0);
+ Map<String, String> offsetMap = binlogSplit.getStartingOffset();
+ if (MapUtils.isEmpty(offsetMap)) {
+ return 0;
  }
- } catch (NumberFormatException e) {
- log.warn("Failed to parse event timestamp from offset: {}", offsetMap, e);
+ try {
+ String timestampSeconds = offsetMap.get("ts_sec");
+ if (timestampSeconds != null) {
+ return Math.max(Long.parseLong(timestampSeconds), 0);
+ }
+ String timestampMicros = offsetMap.get("ts_usec");
+ if (timestampMicros != null) {
+ return Math.max(Long.parseLong(timestampMicros) / 1_000_000, 0);
+ }
+ } catch (NumberFormatException e) {
+ log.warn("Failed to parse source event timestamp from offset: {}", offsetMap, e);
+ }
+ return 0;
+ }
+ }
+
+ @Override
+ public void resetLag() {
+ lagBytes = -1;
+ }
+
+ void updateLagBytes(long fetchedLagBytes) {
+ if (fetchedLagBytes >= 0) {
+ lagBytes = fetchedLagBytes;
  }
- return -1;
  }
 
  @Override