[fix](streaming) reset offset provider state when ALTER JOB switches offset to initial

[maks3201]

Proposed changes

Related PR: #66559 (independent fix in the same subsystem)

Problem Summary

When an operator runs PAUSE JOB → ALTER JOB ... offset='initial' → RESUME JOB to force a fresh CDC snapshot, JdbcSourceOffsetProvider still holds the old binlog position and split progress in memory. On the next scheduler tick the job dispatches a task from the stale offset instead of starting a fresh snapshot. The operator believes they reset the job; they did not.

Why this is not just clearSnapshotState(): The existing clearSnapshotState() method is invoked during the normal snapshot-to-binlog transition inside updateOffset(). It deliberately preserves currentOffset, binlogOffsetPersist, and endBinlogOffset because the incoming binlog phase depends on them. Our new resetToInitialState() is a strict superset: it calls clearSnapshotState() for the split/progress subset, then also clears the binlog-phase fields (currentOffset, binlogOffsetPersist, endBinlogOffset, tableSchemas, hasMoreData, boundBackendId), because the intent is a complete fresh start — not a phase transition within an active job.

Fix

1. Add SourceOffsetProvider.resetToInitialState() as a default no-op method on the interface (S3 and other providers are unaffected).
2. Implement in JdbcSourceOffsetProvider: clear all cached split and binlog state (11 fields — see exhaustive list in the Javadoc).
3. In StreamingInsertJob.alterJob(): when the ALTER explicitly sets offset to initial or snapshot, call resetToInitialState() and null out offsetProviderPersist so that replayIfNeed() takes the fresh-start branch.

After the reset, the next handlePendingState() tick sees empty state and initiates a full snapshot — identical to a brand-new job.

Field-by-field reset audit

Field	Cleared?	Reason
remainingSplits	✅ via clearSnapshotState	Stale snapshot chunks
finishedSplits	✅ via clearSnapshotState	Old split records
chunkHighWatermarkMap	✅ via clearSnapshotState	Old HW would prevent re-splitting
committedSplitProgress	✅ via clearSnapshotState	Stale cursor
cdcSplitProgress	✅ via clearSnapshotState	Stale cursor; noMoreSplits() would return wrong answer
currentOffset	✅ → null	Prevents getNextOffset from returning stale binlog
binlogOffsetPersist	✅ → null	Prevents replayIfNeed from restoring old offset
endBinlogOffset	✅ → null	Stale end offset from previous phase
tableSchemas	✅ → null	Schema may have changed; will be re-fetched
hasMoreData	✅ → true	Fresh start should assume data exists
boundBackendId	✅ → 0	Old pinning was for the binlog reader; snapshot may use different BE
jobId	preserved	Identity
sourceType	preserved	Configuration
sourceProperties	preserved	Updated by alterJob itself
snapshotParallelism	preserved	Configuration
cloudCluster	preserved	Routing config, set externally
cachedSyncTables	preserved	Re-set by replayIfNeed from job.getSyncTables()

Issue Number

Closes #66562

Checklist

• I have read the Contributing to Apache Doris guide.
• I have completed the CLA agreement (via first-time PR bot prompt).
• I have reviewed the PR naming conventions.
• I have added appropriate tests (unit test for resetToInitialState covering binlog-phase reset, snapshot-phase reset, idempotency, noMoreSplits correctness, and configuration preservation).

---

cc @JNSimba — you maintain the streaming job subsystem; this touches the offset provider interface and the alterJob() flow.

Please also consider applying dev/4.1.x label for backport — this bug affects any 4.1.x user attempting manual CDC recovery via ALTER JOB.

[maks3201]

This bug is present in released 4.1.x (confirmed on 4.1.2). If the fix is accepted, the dev/4.1.x label would be appropriate for backport. I cannot apply labels as a non-committer — requesting a maintainer's help.

[hello-stephen]

Thank you for your contribution to Apache Doris.
Don't know what should be done next? See How to process your PR.

Please clearly describe your PR:

1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?

[JNSimba]

CDC jobs can only modify the position in the PAUSED state, and the offset only accepts the exact position specified in the JSON above; initial, snapshot, earliest, or latest offsets are not accepted.

I remember there being this limitation. Are you sure you can modify it successfully? https://github.com/apache/doris/blob/master/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_special_offset.groovy#L216-L220

[maks3201]

@JNSimba you're right, thank you for catching this. I re-checked the code and confirmed validateAlterOffset() in JdbcSourceOffsetProvider rejects any offset that is not JSON so ALTER JOB ... PROPERTIES('offset'='initial') throws before it ever reaches alterJob(). The reset code in this PR is attached to that unreachable path.

I mixed up two different callers from my local branch. The one I upstreamed here (on ALTER) is dead code on master. The real trigger is internal, it's a fallback that runs when a streaming task fails with a specific error and the job needs to re snapshot automatically, that path is reachable and it's where this reset logic actually belongs.

I'm going to close this PR and move the reset logic into a follow up that includes the real trigger, so the fix is tied to code that actually executes. Thanks again for reviewing, sorry for the noise.

[maks3201]

Closing per discussion above - code path unreachable on master, will resubmit tied to the actual trigger.