github-actions[bot] commented on code in PR #66238:
URL: https://github.com/apache/doris/pull/66238#discussion_r3711374141
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -1638,12 +1646,19 @@ private void checkDataQuality(CommitOffsetRequest
offsetRequest) throws JobExcep
}
}
- private void persistOffsetProviderIfNeed() {
- // only for jdbc
- this.offsetProviderPersist = offsetProvider.getPersistInfo();
- if (this.offsetProviderPersist != null) {
- logUpdateOperation();
+ private void persistOffsetProviderIfNeed(
+ JdbcSourceOffsetProvider jdbcOffsetProvider, long currentTimeMs) {
+ this.offsetProviderPersist = jdbcOffsetProvider.getPersistInfo();
+ if (this.offsetProviderPersist == null) {
+ return;
}
+
+ if (!jdbcOffsetProvider.shouldPersistOffset(lastOffsetPersistTimeMs,
currentTimeMs)) {
Review Comment:
This return acknowledges the snapshot chunk even though its source/data
progress exists only in the live FE. A concrete failure is: chunk B
stream-loads row K and commits inside the interval; K is then deleted at the
source; the master fails before another forced journal. Replay rebuilds B from
the older provider JSON and scans it again, but the new snapshot emits neither
K nor a delete marker, so the already-loaded UNIQUE KEY row remains. The
replayed high watermark is now after the delete, which also lets a later binlog
handoff skip it. Please keep each non-transactional snapshot commit durable
before returning success, or add an atomic/reconciliation protocol that makes
an absent row replay-safe; a time-only journal throttle is not safe here.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -1059,6 +1065,8 @@ public void replayOnUpdated(StreamingInsertJob replayJob)
{
setFailedTaskCount(replayJob.getFailedTaskCount());
setCanceledTaskCount(replayJob.getCanceledTaskCount());
setLastTaskSuccessTime(replayJob.getLastTaskSuccessTime());
+ setStartTimeMs(replayJob.getStartTimeMs());
+ setFailureReason(replayJob.getFailureReason());
Review Comment:
The journal producer does not yet contain the value copied here.
`JobManager.alterJobStatus(String, ...)` first updates the status and calls
`logUpdateOperation()`, then calls `onManualStatusAltered()` to set
`MANUAL_PAUSE_ERR` or clear it on RESUME. Consequently the RESUME journal can
contain `PENDING` plus the stale manual-pause reason; after failover this line
restores that pair. A subsequent task failure preserves the stale manual reason
and is excluded from auto-resume; fetch/advance failures are also suppressed as
if the job were manually paused. Please finalize the status, manual reason, and
retry state together under the job lock before one update journal, or otherwise
make the combined transition atomic, and cover PAUSE/RESUME through real
serialization and replay.
--
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]
