github-actions[bot] commented on code in PR #68007:
URL: https://github.com/apache/doris/pull/68007#discussion_r4014174169
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java:
##########
@@ -168,14 +179,22 @@ public void fetchRemoteMeta(Map<String, String>
properties) throws Exception {
throw new java.io.IOException("debug point: simulated S3 auth
error");
}
GlobListing globListing =
fileSystem.globListWithLimit(Location.of(filePath), startFile, 1, 1);
- if (!globListing.getFiles().isEmpty() &&
StringUtils.isNotEmpty(globListing.getMaxFile())) {
+ boolean hasFiles = !globListing.getFiles().isEmpty();
+ if (onceMode && startFile != null && !hasFiles) {
Review Comment:
[P2] Keep an initially empty ONCE source out of the failure budget
With no committed offset, this branch deliberately leaves reachedEnd false,
while hasMoreDataToConsume returns true for currentOffset == null. The PENDING
path does not probe metadata, so a normal empty glob is dispatched;
getNextOffset throws, the task retries four times, and repeated auto-resumes
can end in CANNOT_RESUME_ERR. ONCE should either finish an empty snapshot or
wait through the no-data delay path, but it should not treat zero matches as a
task failure. Please cover the full PENDING -> scheduler -> task state
transition, not only the raw provider exception.
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java:
##########
@@ -35,8 +36,12 @@ public class SourceOffsetProviderFactory {
map.put("cdc_stream", JdbcTvfSourceOffsetProvider.class);
}
- public static SourceOffsetProvider createSourceOffsetProvider(String
sourceType) {
+ public static SourceOffsetProvider createSourceOffsetProvider(
+ String sourceType, StreamingJobProperties jobProperties) {
try {
+ if ("s3".equalsIgnoreCase(sourceType) &&
jobProperties.isS3OnceMode()) {
Review Comment:
[P1] Fence ONCE jobs from older FE leaders
The mode is persisted only as a key in the generic properties map. An FE at
the base version replays that map without validating unknown keys, then its
one-argument factory always reconstructs the no-arg lexical S3 provider. During
a rolling FE upgrade, an older follower that becomes leader can therefore turn
a PENDING ONCE job into indefinite polling and ingest later files instead of
reaching FINISHED. Please gate creation until every electable FE supports this
mode, or persist a backward-compatible discriminator that makes old leaders
fail closed, with a mixed-version replay/failover test.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -917,7 +917,7 @@ public void onStreamTaskSuccess(AbstractStreamingTask task)
throws JobException
}
Env.getCurrentEnv().getJobManager().getStreamingTaskManager().removeRunningTask(task);
- if (offsetProvider.hasReachedEnd()) {
+ if (offsetProvider.hasReachedEnd(task.getRunningOffset())) {
Review Comment:
[P1] Make ONCE completion terminal across scheduler races
The commit thread holds the job lock through this block, but the recurring
metadata task performs S3 I/O before acquiring it. If that listing fails while
the final batch commits, this block sets FINISHED and removes the transaction
callback; the waiting fetchMeta path then changes the job to PAUSED because
AbstractJob permits FINISHED -> PAUSED. Auto/manual resume can reopen a job
whose callback is no longer registered. Both new scheduler completion paths
have the inverse TOCTOU: their earlier eligibility checks are released before
an unconditional FINISHED write, so they can overwrite a concurrent
PAUSED/STOPPED state. Please make these transitions conditional under the same
job lock, or enforce that terminal states are absorbing, and add latch-based
coverage.
##########
fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java:
##########
@@ -122,6 +122,11 @@ private void scheduleOneTask(AbstractStreamingTask task) {
}
// reject task if no more data to consume
if (!job.hasMoreDataToConsume()) {
+ if (job.hasReachedEnd()) {
Review Comment:
[P2] Persist the cloud-replayed offset before finishing
After a crash following a cloud commit, replayOnCloudMode restores the
authoritative offset only into offsetProvider; unlike local replay, it does not
refresh offsetProviderPersist. This new exhaustion branch can then journal
FINISHED with the stale/null persisted string. Because Gson excludes the
provider object and a restored FINISHED job does not replay Meta Service again,
a later checkpoint restart permanently shows the old final offset. Please
synchronize offsetProviderPersist whenever cloud replay applies an attachment
before allowing either natural-completion path to persist FINISHED, and add a
cloud recovery plus image-round-trip test.
##########
fe/fe-core/src/main/java/org/apache/doris/job/scheduler/StreamingTaskScheduler.java:
##########
@@ -122,6 +122,11 @@ private void scheduleOneTask(AbstractStreamingTask task) {
}
// reject task if no more data to consume
if (!job.hasMoreDataToConsume()) {
+ if (job.hasReachedEnd()) {
+ job.updateJobStatus(JobStatus.FINISHED);
+ job.logUpdateOperation();
+ return;
Review Comment:
[P2] Retire the placeholder task when recovery finishes
A recovered ONCE job records a PENDING placeholder task; its first scheduler
pass delays that task and sets the "will retry" runtime message. After the
metadata probe marks the source exhausted, the delayed pass reaches this return
without executing, canceling, finalizing, or removing the task and without
clearing the message. The job is therefore shown as FINISHED while its task
remains PENDING with no finish time and the job still says it will retry.
Please finalize/cancel the placeholder and clear its reference/message as part
of the terminal transition, with a recovered-empty-tail state test.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java:
##########
@@ -119,6 +124,15 @@ public void validate() throws AnalysisException {
&& v <= (long) (1024 * 1024 * 1024) * 10,
StreamingJobProperties.S3_MAX_BATCH_BYTES_PROPERTY + " should
between 100MB and 10GB");
+ String ingestionMode = getS3IngestionMode();
Review Comment:
[P2] Reject the S3-only mode on non-S3 sources
This validation runs before the source type is resolved, so cdc_stream and
FROM-source JDBC jobs accept s3.ingestion_mode=ONCE even though their provider
paths ignore it and continue indefinitely. The same irrelevant property can
still reject those providers' supported offset setting here. Please validate
the explicit property after source/provider resolution in CREATE and ALTER,
rejecting it outside S3, and add negative coverage for both non-S3 paths.
--
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]
