cloud-fan commented on code in PR #56559:
URL: https://github.com/apache/spark/pull/56559#discussion_r3470052283
##########
core/src/main/scala/org/apache/spark/storage/PushBasedFetchHelper.scala:
##########
@@ -17,23 +17,20 @@
package org.apache.spark.storage
-import java.util.concurrent.TimeUnit
-
-import scala.collection
-import scala.collection.mutable
-import scala.collection.mutable.ArrayBuffer
-import scala.util.{Failure, Success}
-
-import org.roaringbitmap.RoaringBitmap
-
import org.apache.spark.MapOutputTracker
import org.apache.spark.MapOutputTracker.SHUFFLE_PUSH_MAP_ID
-import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys._
+import org.apache.spark.internal.Logging
import org.apache.spark.network.shuffle.{BlockStoreClient, MergedBlockMeta,
MergedBlocksMetaListener}
import org.apache.spark.shuffle.ShuffleReadMetricsReporter
import org.apache.spark.storage.BlockManagerId.SHUFFLE_MERGER_IDENTIFIER
import org.apache.spark.storage.ShuffleBlockFetcherIterator._
+import org.roaringbitmap.RoaringBitmap
+
+import java.util.concurrent.TimeUnit
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+import scala.util.{Failure, Success}
Review Comment:
These imports are now ordered spark → 3rdParty → java → scala, but
scalastyle's `ImportOrderChecker` enforces `java, scala, 3rdParty, spark` at
error level (`scalastyle-config.xml:578`), so the lint CI gate will fail. Looks
like an IDE "optimize imports" with non-Spark settings; restore the grouping:
```suggestion
import java.util.concurrent.TimeUnit
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import scala.util.{Failure, Success}
import org.roaringbitmap.RoaringBitmap
import org.apache.spark.MapOutputTracker
import org.apache.spark.MapOutputTracker.SHUFFLE_PUSH_MAP_ID
import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.Logging
import org.apache.spark.network.shuffle.{BlockStoreClient, MergedBlockMeta,
MergedBlocksMetaListener}
import org.apache.spark.shuffle.ShuffleReadMetricsReporter
import org.apache.spark.storage.BlockManagerId.SHUFFLE_MERGER_IDENTIFIER
import org.apache.spark.storage.ShuffleBlockFetcherIterator._
```
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -105,6 +105,30 @@ private class ShuffleStatus(
*/
private[spark] val checksumMismatchIndices: Set[Int] = Set()
+ /**
+ * Set of stale pushed partition indexes for this shuffle. Each entry is a
partitionId (which
+ * equals mapIndex, not MapStatus.mapId). When task retry or speculation
causes multiple
+ * attempts for the same map output to push, the merger may include data
from a stale attempt.
+ * We record the stale partition indexes here so the reduce side can check
chunkBitmaps and
+ * fallback if stale data is present in a merged block.
+ */
+ private[this] val staleMapIndexes = new java.util.HashSet[Int]()
+
+ /**
+ * Mark a partition as having stale (redundant) push attempts. Called from
TaskSetManager when it
+ * detects that multiple task attempts for the same map output pushed data
to the merger.
+ * @param mapIndex the partition index (== mapIndex) of the stale
(redundant) attempt;
+ * this is NOT MapStatus.mapId
+ */
+ def markStalePushedPartition(mapIndex: Int): Unit = withWriteLock {
+ staleMapIndexes.add(mapIndex)
+ }
+
+ /**
+ * Get all stale pushed partition indexes for this shuffle. Returns empty
set if none exist.
+ */
+ def getStaleMapIndexes: Set[Int] = staleMapIndexes.asScala
Review Comment:
`.asScala` returns a live, mutable view backed by the non-thread-safe
`staleMapIndexes` HashSet, and this read takes no `withReadLock` — yet
`markStalePushedPartition` mutates the same set under `withWriteLock`. When the
reduce side uses the master tracker, `checkStaleMapIdInMergedBlock` iterates
this view (`isEmpty` / `exists`) while the driver may be marking a new stale
partition, which is a data race and can throw `ConcurrentModificationException`
mid-iteration. Both siblings already handle this correctly — the worker-side
accessor returns a defensive copy (line 1359) and the serialization path takes
`withReadLock` (line 466). Mirror them here:
```suggestion
def getStaleMapIndexes: Set[Int] = withReadLock {
new java.util.HashSet[Int](staleMapIndexes).asScala
}
```
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1644,6 +1717,48 @@ private[spark] object MapOutputTracker extends Logging {
}
}
+ /**
+ * Serialize a set of stale pushed partition indexes into a compact byte
array.
+ * Uses DataOutputStream for a simple, efficient binary format:
+ * [int: count][int: mapId1][int: mapId2]...
Review Comment:
These elements are mapIndexes (partitionIds), not `MapStatus.mapId` — the
exact term this PR renamed everywhere to avoid. Match the diagram to the rest
of the doc:
```suggestion
* [int: count][int: mapIndex1][int: mapIndex2]...
```
--
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]
