ziting-openai commented on code in PR #5627:
URL: https://github.com/apache/datafusion-comet/pull/5627#discussion_r3908927899
##########
spark/src/main/java/org/apache/comet/shuffle/CelebornShufflePartitionPusher.java:
##########
@@ -763,55 +814,54 @@ private void reconcileAcceptedPushes() {
ArrayList<PushReservation> completed = new ArrayList<>();
IOException detectedFailure = null;
synchronized (lifecycleLock) {
+ Iterator<PushReservation> pending = pendingPushes.iterator();
+ while (pending.hasNext()) {
+ PushReservation reservation = pending.next();
+ if (!reservation.submitted) {
+ continue;
+ }
+ registerPushStateFallback(reservation);
+ if (!reservation.pushStateFallback &&
reservation.transportPush.isComplete()) {
+ pending.remove();
+ completed.add(reservation);
+ }
+ }
for (ObservedPushState observed : observedPushStates.values()) {
Object failure = observed.exception.get();
boolean cleanedUp =
failure instanceof IOException
&& "Cleaned Up".equals(((IOException) failure).getMessage());
if (failure instanceof IOException && !cleanedUp) {
IOException observedFailure = (IOException) failure;
- // Only the raw transport callback's pinned message proves request
termination.
- // PushState can also publish interruption/backpressure exceptions
while an older
- // request is still live; those failures must never free that
request's admission.
- if (!observed.terminalFailureCredited &&
isTerminalRawPushFailure(observedFailure)) {
- observed.terminalFailureCredited = true;
- }
if (asynchronousFailure == null) {
asynchronousFailure = observedFailure;
detectedFailure = asynchronousFailure;
}
}
- if (transportCallbacks != null) {
- // Stock counters can remain positive after cancellation. The
transport path instead
- // tracks each request through the raw call, transport callbacks,
and queued retries.
+ if (hasPendingTransportOwnership(observed)) {
+ // A stock counter cannot distinguish a fallback request from a
transport-owned request.
+ // Wait for precise requests on this push state before crediting
fallback completions.
continue;
}
- long retainedRequests =
- Math.max(
- 0L, observed.inFlightRequests.sum() -
(observed.terminalFailureCredited ? 1L : 0L));
- long completions = Math.max(0L, observed.submittedPushes -
retainedRequests);
- while (observed.releasedPushes < completions) {
- PushReservation reservation = smallestSubmittedReservation(observed);
- if (reservation == null) {
+ boolean terminalFailure =
+ failure instanceof IOException
+ && !cleanedUp
+ && isTerminalRawPushFailure((IOException) failure);
+ long counterCredits = terminalFailure ? 1L : 0L;
+ long retainedRequests = Math.max(0L, observed.inFlightRequests.sum() -
counterCredits);
+ long completions = Math.max(0L, observed.fallbackSubmittedPushes -
retainedRequests);
Review Comment:
Re-reviewed `9a75f9de3e37136968694a74a5d6ad2623780385`. The isolated probe
now passes the single-request case and cancellation sealed before a later
downgrade. This P2 still occurs when **two requests are already precisely
tracked before a global bootstrap failure downgrades both**.
`awaitPushStateFallbackSlot()` cannot serialize those already submitted
requests. The first failed stock callback sets the terminal exception, the
second returns on that exception, and the counter remains 2. With two fallback
submissions and only one counter credit, only one reservation is released.
I reproduced this against the unchanged new-head classes with the Celeborn
behavior fixture: after both callbacks completed, only 64 of 128 test bytes
were available, cleanup for that map remained deferred, and the next valid
112-byte reservation blocked in the actual admission wait. Please cover the
two-outstanding-request downgrade sequence and preserve a terminal ownership
signal for each request, or safely seal/cancel eligible precise requests before
the downgrade. This was an isolated JDK 21 probe with transport fixtures and a
logging shim, not a full Celeborn/Netty run.
_[ :robot: posted by Codex on behalf of ziting-openai using the
spark-pr-review-memo skill :robot: ]_
--
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]
