github-actions[bot] commented on code in PR #66409:
URL: https://github.com/apache/doris/pull/66409#discussion_r3712969814
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java:
##########
@@ -1174,13 +1176,64 @@ public synchronized void finishSplitRecords() {
}
@Override
- public Map<String, String> getEndOffset(JobBaseConfig jobConfig) {
- MySqlSourceConfig sourceConfig = getSourceConfig(jobConfig);
+ public FetchEndOffsetResult fetchEndOffset(FetchEndOffsetRequest request) {
+ MySqlSourceConfig sourceConfig = getSourceConfig(request);
try (MySqlConnection jdbc =
DebeziumUtils.createMySqlConnection(sourceConfig)) {
- BinlogOffset binlogOffset =
DebeziumUtils.currentBinlogOffset(jdbc);
- return binlogOffset.getOffset();
- } catch (SQLException ex) {
- throw new RuntimeException(ex);
+ Map<String, String> endOffset =
DebeziumUtils.currentBinlogOffset(jdbc).getOffset();
+ long lagBytes;
+ try {
+ lagBytes = calculateLagBytes(request, endOffset, jdbc);
+ } catch (Exception exception) {
+ lagBytes = -1;
+ LOG.warn(
+ "Failed to calculate source log lag for job {}",
+ request.getJobId(),
+ exception);
+ }
+ return new FetchEndOffsetResult(endOffset, lagBytes);
+ } catch (SQLException exception) {
+ throw new RuntimeException(exception);
+ }
+ }
+
+ private long calculateLagBytes(
+ FetchEndOffsetRequest request,
+ Map<String, String> endOffset,
+ MySqlConnection jdbc)
+ throws SQLException {
+ if (MapUtils.isEmpty(request.getReferenceOffset())) {
+ return -1;
+ }
+ try (Statement statement = jdbc.connection().createStatement();
+ ResultSet resultSet = statement.executeQuery("SHOW BINARY
LOGS")) {
+ List<MySqlBinlogLagCalculator.BinlogFile> binlogFiles = new
ArrayList<>();
+ while (resultSet.next()) {
+ binlogFiles.add(
+ new MySqlBinlogLagCalculator.BinlogFile(
+ resultSet.getString(1), resultSet.getLong(2)));
Review Comment:
[P2] Normalize encrypted binlog sizes before combining them with event
positions. MySQL documents that `SHOW BINARY LOGS` includes a 512-byte header
in `File_size` for encrypted rows, but replication `pos` values exclude that
header ([SHOW BINARY
LOGS](https://dev.mysql.com/doc/refman/8.0/en/show-binary-logs.html)); toggling
`binlog_encryption` rotates immediately and leaves older files in their prior
state ([binary-log
encryption](https://dev.mysql.com/doc/refman/8.0/en/replication-binlog-encryption.html)),
so this list can legitimately mix encrypted and unencrypted rows. Because this
code drops the `Encrypted` column, one formula cannot measure both transition
directions in a consistent coordinate system. Please consume the per-row
encryption flag and normalize the chosen physical/logical coordinates, while
retaining a two-column fallback for MySQL 5.7/OceanBase, and add both
mixed-state rotation fixtures.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -1167,8 +1158,7 @@ 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()));
Review Comment:
[P2] Give the byte value a unit-qualified SHOW identity. `InsertJob.SCHEMA`
still exposes this position as the existing string column `Lag`, so clients
that already parse that numeric field as seconds will keep succeeding while
silently interpreting bytes as time. This is separate from the
Prometheus-series unit issue because SQL consumers never see `MetricUnit`
either. Please preserve/deprecate the old contract explicitly and add a
`LagBytes` field (or otherwise version/rename the SHOW schema), with a
SHOW-level compatibility test.
##########
regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_lag.groovy:
##########
@@ -81,7 +81,21 @@ suite("test_streaming_mysql_job_lag",
}
def lagValue = jobInfo[0][1] as String
log.info("lag value: " + lagValue)
- return lagValue != null && lagValue != "" &&
lagValue.isNumber()
+ return lagValue != null && lagValue != ""
Review Comment:
[P2] Make this regression fail under the old seconds-based implementation.
Both this check and the new PAUSED check accept any nonnegative integer, so
`(now - lastEventTime) / 1000` would still pass; the other changed
smoke/regression assertions use the same oracle. Please assert a controlled
byte delta reaches SHOW (and the per-job metric), or at minimum capture the
paused/idle value, wait beyond the old clock granularity with no source writes,
and prove it remains unchanged.
--
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]
