Ngone51 commented on a change in pull request #32401:
URL: https://github.com/apache/spark/pull/32401#discussion_r668409304
##########
File path:
core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockResolver.scala
##########
@@ -360,13 +389,41 @@ private[spark] class IndexShuffleBlockResolver(
if (dataTmp != null && dataTmp.exists() &&
!dataTmp.renameTo(dataFile)) {
throw new IOException("fail to rename file " + dataTmp + " to " +
dataFile)
}
+
+ // write the checksum file
+ checksumTmpOpt.zip(checksumFileOpt).foreach { case (checksumTmp,
checksumFile) =>
+ val out = new DataOutputStream(
+ new BufferedOutputStream(
+ new FileOutputStream(checksumTmp)
+ )
+ )
+ Utils.tryWithSafeFinally {
+ checksums.foreach(out.writeLong)
+ } {
+ out.close()
+ }
+
+ if (checksumFile.exists()) {
+ checksumFile.delete()
+ }
+ if (!checksumTmp.renameTo(checksumFile)) {
+ // It's not worthwhile to fail here after index file and data
file are already
+ // successfully stored due to checksum is only used for the
corner error case.
+ logWarning("fail to rename file " + checksumTmp + " to " +
checksumFile)
+ }
+ }
}
}
} finally {
logDebug(s"Shuffle index for mapId $mapId: ${lengths.mkString("[", ",",
"]")}")
if (indexTmp.exists() && !indexTmp.delete()) {
logError(s"Failed to delete temporary index file at
${indexTmp.getAbsolutePath}")
}
+ checksumTmpOpt.foreach { checksumTmp =>
+ if (checksumTmp.exists() && !checksumTmp.delete()) {
Review comment:
Good point! We won't propagate the error. I'll handle it.
--
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: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
