ziting-openai commented on code in PR #5627:
URL: https://github.com/apache/datafusion-comet/pull/5627#discussion_r3908247906
##########
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:
Confirmed this P2 with an isolated probe against unchanged head
`36f418e61672243ab90651f462ea7b34f7e44009`. The client fixture follows the
released Celeborn 0.6.0/0.7.0 cleanup and failed-callback behavior. After abort
and terminal callback completion, the payload was gone but the counter stayed
at 1, all 64 test bytes remained reserved, and the next map reached the real
`ExecutorShufflePushAdmission.acquire` wait. One hundred reconciliation calls
left the same state. Success and abort-then-success controls released
admission. The parent revision rejected the instrumentation mismatch before
publishing a raw request.
The existing `failed transport after cleanup releases only its completed
shared admission` regression should cover this with bootstrap tracking
disabled. A safe terminal-completion signal is still needed in that mode.
Crediting `Cleaned Up` alone would release bytes before transport retirement.
This probe used a fixture and a no-op logging shim on JDK 21, not a full
Celeborn/Netty integration 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]
