Handle stale fetch streams after worker restart

[mzhou-oai]

Problem Statement

A worker restart can leave an in-flight reducer holding a stale streamId even though the worker comes back on the same stable hostname and still has the shuffle data on disk. In that case the worker correctly reports:

• Stream <id> is not registered with worker. This can happen if the worker was restart recently.

That is a restart-specific condition, not necessarily a hard worker failure. Live stream registrations are process-local and are not reconstructed by recoverPath, so the old streamId cannot be used after the worker process comes back.

Before this change, CelebornInputStream treated that response like a generic fetch failure. It excluded the worker, consumed normal retry budget and backoff, and entered the usual peer-failover or retry path instead of reopening the stream on the same worker. As a result, a Celeborn worker restart can strand active Spark tasks in Celeborn fetch retry loops even when the worker is already back and the shuffle data is still available.

Proposal

This change makes stale-stream handling explicit in the client retry path:

• mark the stale-stream case with a stable coded failure marker on the existing ChunkFetchFailure error string
• parse that marker into ChunkFetchFailureException and treat it as the structured signal for same-worker reopen
• keep a fallback to the legacy human-readable message so new clients still recover from older workers that do not send the code yet
• do not classify that specific failure as a critical fetch cause
• do not exclude the restarted worker before retrying
• recreate the reader on the same PartitionLocation with pbStreamHandler = null so the client issues a fresh OPEN_STREAM
• reuse checkpoint metadata so already returned chunks are skipped on the reopened stream

Validation

• mvn -pl client -am -DskipTests compile
• mvn -pl common -am -Dtest=ExceptionUtilsSuiteJ,TransportResponseHandlerSuiteJ test
• Verified the fix e2e on a running celeborn cluster in k8s, ran tpc-ds 3TB benchmark.

[eolivelli]

Can you please add an end to end test with a worker that is restarted ?

[eolivelli]

This check is too fragile, we need a better way.
Can we add some error code ?

[mzhou-oai]

ok, just did an attempt, ChunkFetchFailureException doesn't carry an error code ,and it's hard to wire in RPC. see the commit history for the attempts.

[mzhou-oai]

@eolivelli Please let me know which revision is preferred

[mzhou-oai]

Can you please add an end to end test with a worker that is restarted ?

Yes, it's verified e2e. let me update the PR body about that.