github-actions[bot] commented on code in PR #66409:
URL: https://github.com/apache/doris/pull/66409#discussion_r3820352757
##########
fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java:
##########
@@ -1463,16 +1469,44 @@ public Long getValue() {
DORIS_METRIC_REGISTER.addMetrics(failedTaskCount);
GaugeMetric<Long> lag = new GaugeMetric<Long>(
- STREAMING_JOB_PER_JOB_LAG, MetricUnit.SECONDS,
- "per job lag in seconds of streaming job, -1 means
N/A") {
+ STREAMING_JOB_PER_JOB_LAG_BYTES, MetricUnit.BYTES,
+ "latest successfully observed source log lag in bytes,
-1 means no valid observation") {
@Override
public Long getValue() {
- return sJob.getLagSeconds();
+ return sJob.getLagBytes();
}
};
lag.addLabel(new MetricLabel("job_id", jobId))
.addLabel(new MetricLabel("job_name", jobName));
DORIS_METRIC_REGISTER.addMetrics(lag);
+
+ long lastSourceEventTimestampSeconds =
sJob.getLastSourceEventTimestampSeconds();
+ GaugeMetric<Long> lastSourceEventTimestamp = new
GaugeMetric<Long>(
+
STREAMING_JOB_PER_JOB_LAST_SOURCE_EVENT_TIMESTAMP_SECONDS, MetricUnit.SECONDS,
+ "timestamp of the latest source binlog or WAL event
recorded in the job's committed offset"
+ + " as Unix seconds, 0 means unavailable") {
+ @Override
+ public Long getValue() {
+ return lastSourceEventTimestampSeconds;
+ }
+ };
+ lastSourceEventTimestamp.addLabel(new MetricLabel("job_id",
jobId))
+ .addLabel(new MetricLabel("job_name", jobName));
+ DORIS_METRIC_REGISTER.addMetrics(lastSourceEventTimestamp);
+
+ long lastTaskSuccessTimeSeconds =
sJob.getLastTaskSuccessTimeSeconds();
+ GaugeMetric<Long> lastTaskSuccessTime = new GaugeMetric<Long>(
+ STREAMING_JOB_PER_JOB_LAST_TASK_SUCCESS_TIME_SECONDS,
MetricUnit.SECONDS,
Review Comment:
[P2] Persist this timestamp with the successful task transition before
exporting it as the latest completion time. In the transaction-backed path, the
durable commit attachment/replay advances the offset and success count but not
`lastTaskSuccessTime`; live `onSuccess()` sets it afterward without a job
update. In the multi-table path, `commitOffset()` writes any job update before
`successCallback()` sets it. After failover this new gauge can therefore return
`0` or task N-1 while the replayed offset/count include task N. Please journal
or reconstruct the timestamp in the same durable transition and add
immediate-failover coverage for both paths.
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java:
##########
@@ -960,45 +997,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 Long.parseLong(timestampSeconds);
+ }
+ String timestampMicros = offsetMap.get("ts_usec");
+ if (timestampMicros != null) {
+ return Long.parseLong(timestampMicros) / 1_000_000;
Review Comment:
[P2] Normalize PostgreSQL's no-event timestamp before exporting it. An
`offset=latest`/initial reader can commit
[`PostgresOffset.INITIAL_OFFSET`](https://github.com/apache/flink-cdc/blob/release-3.6.0/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/offset/PostgresOffset.java#L39-L59),
whose `ts_usec` is `Conversions.toEpochMicros(Instant.MIN)`
(`Long.MIN_VALUE`); this branch returns `-9223372036854`, and the new
Prometheus gauge publishes it even though its contract says `0` means
unavailable (SHOW only hides it via a separate `> 0` check). Please normalize
the connector's unavailable sentinel to `0` and add an idle baseline PostgreSQL
fixture using the real sentinel.
##########
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()))
Review Comment:
[P2] Keep this TVF schema compatible across mixed FE versions. The query FE
builds projected names from its local `InsertJob.SCHEMA`, the BE forwards those
names to the master, and the master resolves them with its own
`COLUMN_TO_INDEX`. During a rolling upgrade, a new query FE sends
`LagBytes`/`LastSourceEventTimestamp` to an old master, while an old query FE
sends the removed `Lag` to a new master; either unknown name yields a null
index and fails the metadata fetch. Please gate the schema by master/cluster
compatibility or provide versioned aliases/defaults, and test both rolling
directions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
