github-actions[bot] commented on code in PR #66409:
URL: https://github.com/apache/doris/pull/66409#discussion_r3709333322
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/controller/ClientController.java:
##########
@@ -103,12 +106,20 @@ public Object writeRecord(@RequestBody WriteRecordRequest
recordReq) {
/** Fetch lastest end meta */
@RequestMapping(path = "/api/fetchEndOffset", method = RequestMethod.POST)
- public Object fetchEndOffset(@RequestBody JobBaseConfig jobConfig) {
+ public Object fetchEndOffset(@RequestBody FetchEndOffsetRequest jobConfig)
{
LOG.info("Fetching end offset for job {}", jobConfig.getJobId());
try {
SourceReader reader = Env.getCurrentEnv().getMetaReader(jobConfig);
Env.getCurrentEnv().keepAlive(jobConfig.getJobId());
- return RestResponse.success(reader.getEndOffset(jobConfig));
+ Map<String, String> endOffset = reader.getEndOffset(jobConfig);
+ long lagBytes;
+ try {
+ lagBytes = reader.getLagBytes(jobConfig, endOffset);
+ } catch (Exception ex) {
+ lagBytes = -1;
+ LOG.warn("Failed to calculate source log lag, jobId={}",
jobConfig.getJobId(), ex);
+ }
+ return RestResponse.success(new FetchEndOffsetResult(endOffset,
lagBytes));
Review Comment:
[P1] Preserve `fetchEndOffset` across rolling upgrades. This unversioned
endpoint used to put the offset map directly in `data`, while the new response
nests it under `endOffset`. A new FE talking to an old bound BE/cdc-client
cannot convert `{file,pos}`/`{lsn}` into `FetchEndOffsetResult`; an old FE
talking to a new jar cannot convert the nested `endOffset` object into
`Map<String,String>`. The BE relays this JSON opaquely, and either decode
failure reaches `fetchMeta()` and pauses an otherwise healthy job. Please
version/negotiate the response (or preserve the old endpoint plus a new lag
endpoint) so both rolling directions work, and add both mixed-version fixtures.
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java:
##########
@@ -312,6 +325,26 @@ public void fetchRemoteMeta(Map<String, String>
properties) throws Exception {
}
}
+ Map<String, String> getLagReferenceOffset() {
+ if (isSnapshotOnlyMode() || sourceType == DataSourceType.POSTGRES) {
+ 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());
+ }
+ }
+ return finishedSplits.stream()
+ .map(SnapshotSplit::getHighWatermark)
+ .filter(MapUtils::isNotEmpty)
+ .findFirst()
Review Comment:
[P2] Use the same snapshot reference that the binlog reader will replay
from. `finishedSplits` is not ordered by source position under parallel
snapshotting, but this `findFirst()` can select `file10:100` even when another
committed split has `file9:900`. `MySqlSourceReader.createBinlogSplit()` later
scans all watermarks and starts from the minimum, so Lag omits the tail of file
9 (and under-reports the backlog); OceanBase inherits the same path. The new
test even locks in this contradictory order. Please derive/reuse the minimum
comparable high watermark until a committed binlog offset exists, and test the
exact cross-file byte delta.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/controller/ClientController.java:
##########
@@ -103,12 +106,20 @@ public Object writeRecord(@RequestBody WriteRecordRequest
recordReq) {
/** Fetch lastest end meta */
@RequestMapping(path = "/api/fetchEndOffset", method = RequestMethod.POST)
- public Object fetchEndOffset(@RequestBody JobBaseConfig jobConfig) {
+ public Object fetchEndOffset(@RequestBody FetchEndOffsetRequest jobConfig)
{
LOG.info("Fetching end offset for job {}", jobConfig.getJobId());
try {
SourceReader reader = Env.getCurrentEnv().getMetaReader(jobConfig);
Env.getCurrentEnv().keepAlive(jobConfig.getJobId());
- return RestResponse.success(reader.getEndOffset(jobConfig));
+ Map<String, String> endOffset = reader.getEndOffset(jobConfig);
+ long lagBytes;
+ try {
+ lagBytes = reader.getLagBytes(jobConfig, endOffset);
Review Comment:
[P1] Keep best-effort lag IO from gating end-offset progress. After
`getEndOffset()` succeeds, this call synchronously opens another source
connection and runs an unbounded query (`SHOW BINARY LOGS` for MySQL/OceanBase
or the slot query for PostgreSQL). The inner catch cannot help until that work
returns; if it exceeds the FE light-RPC deadline, FE times out and
`fetchMeta()` pauses the job. Snapshot-only PostgreSQL pays this cost even
though FE discards the value. Please isolate/cache lag outside the required
metadata RPC, or at least share the source operation and enforce a strict
remaining-deadline/statement timeout so a lag failure cannot pause ingestion.
##########
fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobLagTest.java:
##########
@@ -0,0 +1,55 @@
+// 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.extensions.insert.streaming;
+
+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.offset.jdbc.JdbcSourceOffsetProvider;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class StreamingInsertJobLagTest {
+
+ @Test
+ public void testExplicitOffsetChangeKeepsLastObservedLagUntilNextFetch()
throws Exception {
+ StreamingInsertJob job =
Deencapsulation.newInstance(StreamingInsertJob.class);
+ Map<String, String> properties = new HashMap<>();
+ properties.put(StreamingJobProperties.MAX_INTERVAL_SECOND_PROPERTY,
"10");
+ Deencapsulation.setField(job, "properties", properties);
+ Deencapsulation.setField(job, "jobProperties", new
StreamingJobProperties(properties));
+
+ JobExecutionConfiguration configuration = new
JobExecutionConfiguration();
+ configuration.setTimerDefinition(new TimerDefinition());
+ Deencapsulation.setField(job, "jobConfig", configuration);
+
+ JdbcSourceOffsetProvider provider = new JdbcSourceOffsetProvider();
+ provider.setLagBytes(4096);
+ Deencapsulation.setField(job, "offsetProvider", provider);
+
+ Map<String, String> alterProperties = new HashMap<>();
+ alterProperties.put(StreamingJobProperties.OFFSET_PROPERTY,
"{\"lsn\":\"200\"}");
+ Deencapsulation.invoke(job, "modifyPropertiesInternal",
alterProperties);
+
+ Assert.assertEquals(4096, provider.getLagBytes());
Review Comment:
[P2] This expected value preserves an observation for the old checkpoint
after the job's reference offset has changed. `ALTER JOB` is allowed only while
PAUSED, and manual PAUSE does not call `fetchMeta()`, so `4096` can remain
visible indefinitely even though it no longer describes the configured
`lsn=200`; under the new contract there is no valid observation for that
checkpoint yet. Please reset `lagBytes` to `-1` when an explicit offset/source
identity is installed, then replace it after the first successful post-resume
fetch, and make this test expect `-1`.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReader.java:
##########
@@ -491,6 +494,43 @@ public Map<String, String> getEndOffset(JobBaseConfig
jobConfig) {
}
}
+ @Override
+ public long getLagBytes(FetchEndOffsetRequest request, Map<String, String>
endOffset) {
+ PostgresSourceConfig sourceConfig = getSourceConfig(request);
+ PostgresDialect dialect = new PostgresDialect(sourceConfig);
+ try (JdbcConnection jdbcConnection =
dialect.openJdbcConnection(sourceConfig)) {
+ return PostgresWalLagCalculator.calculate(
+ dialect.getSlotName(),
+ slotName -> {
+ try (PreparedStatement statement =
+ jdbcConnection
+ .connection()
+ .prepareStatement(
+ "SELECT
pg_wal_lsn_diff(pg_current_wal_lsn(),"
Review Comment:
[P1] Measure PostgreSQL lag from the latest FE checkpoint, not the previous
task's slot acknowledgement. `cleanupReaderResources()` commits only the stream
split's starting offset `O0` before the advanced state is flushed and committed
to FE as `O1`. Once FE sees `end == O1`, `hasMoreDataToConsume()` suppresses
another task, so `confirmed_flush_lsn` can remain at `O0` indefinitely while
the source is idle and this query keeps reporting `head - O0` instead of zero.
That recreates the false backlog this change is meant to remove. Pass/validate
the FE-committed LSN (or synchronize the slot after FE commit) and add an exact
caught-up/paused test.
##########
fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java:
##########
@@ -1447,11 +1447,11 @@ 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, MetricUnit.BYTES,
Review Comment:
[P2] Do not change physical units behind the existing Prometheus series.
`PrometheusMetricVisitor` does not export `MetricUnit`, so existing queries for
`doris_fe_streaming_job_per_job_lag` will continue matching but silently
interpret byte counts as seconds (and alerts can change by orders of
magnitude). Publish this value under a unit-qualified series such as
`streaming_job_per_job_lag_bytes`; remove or deprecate the old seconds series
explicitly instead of reusing its identity, and update the regression to assert
the new name.
--
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]
