srowen commented on a change in pull request #34980:
URL: https://github.com/apache/spark/pull/34980#discussion_r774597380
##########
File path: core/src/main/scala/org/apache/spark/storage/BlockManager.scala
##########
@@ -926,30 +936,62 @@ private[spark] class BlockManager(
})
Some(new BlockResult(ci, DataReadMethod.Memory, info.size))
} else if (level.useDisk && diskStore.contains(blockId)) {
- val diskData = diskStore.getBytes(blockId)
- val iterToReturn: Iterator[Any] = {
- if (level.deserialized) {
- val diskValues = serializerManager.dataDeserializeStream(
- blockId,
- diskData.toInputStream())(info.classTag)
- maybeCacheDiskValuesInMemory(info, blockId, level, diskValues)
- } else {
- val stream = maybeCacheDiskBytesInMemory(info, blockId, level,
diskData)
- .map { _.toInputStream(dispose = false) }
- .getOrElse { diskData.toInputStream() }
- serializerManager.dataDeserializeStream(blockId,
stream)(info.classTag)
- }
- }
- val ci = CompletionIterator[Any, Iterator[Any]](iterToReturn, {
- releaseLockAndDispose(blockId, diskData, taskContext)
- })
- Some(new BlockResult(ci, DataReadMethod.Disk, info.size))
+ getLocalValuesFromDisk(blockId, info, taskContext)
} else {
handleLocalReadFailure(blockId)
}
}
}
+ private def getLocalValuesFromDisk(blockId: BlockId, info: BlockInfo,
Review comment:
I missed why this needed to be refactored
##########
File path: core/src/main/scala/org/apache/spark/storage/BlockManager.scala
##########
@@ -342,6 +343,15 @@ private[spark] class BlockManager(
iter.close()
false
}
+ } catch {
+ case ex: KryoException
+ // We need to have clear error message to catch environmental
problems easily.
+ // Further details: https://issues.apache.org/jira/browse/SPARK-37710
+ if ex.getMessage.toLowerCase(Locale.ROOT)
Review comment:
Can we match on the cause of the KryoException to detect IOException?
would that catch it more cleanly, or are only some IOExceptions of interest?
##########
File path: core/src/main/scala/org/apache/spark/storage/BlockManager.scala
##########
@@ -926,30 +936,62 @@ private[spark] class BlockManager(
})
Some(new BlockResult(ci, DataReadMethod.Memory, info.size))
} else if (level.useDisk && diskStore.contains(blockId)) {
- val diskData = diskStore.getBytes(blockId)
- val iterToReturn: Iterator[Any] = {
- if (level.deserialized) {
- val diskValues = serializerManager.dataDeserializeStream(
- blockId,
- diskData.toInputStream())(info.classTag)
- maybeCacheDiskValuesInMemory(info, blockId, level, diskValues)
- } else {
- val stream = maybeCacheDiskBytesInMemory(info, blockId, level,
diskData)
- .map { _.toInputStream(dispose = false) }
- .getOrElse { diskData.toInputStream() }
- serializerManager.dataDeserializeStream(blockId,
stream)(info.classTag)
- }
- }
- val ci = CompletionIterator[Any, Iterator[Any]](iterToReturn, {
- releaseLockAndDispose(blockId, diskData, taskContext)
- })
- Some(new BlockResult(ci, DataReadMethod.Disk, info.size))
+ getLocalValuesFromDisk(blockId, info, taskContext)
} else {
handleLocalReadFailure(blockId)
}
}
}
+ private def getLocalValuesFromDisk(blockId: BlockId, info: BlockInfo,
+ taskContext: Option[TaskContext]): Option[BlockResult] = {
+ try {
+ val level = info.level
+ val diskData = diskStore.getBytes(blockId)
+ val iterToReturn: Iterator[Any] = {
+ if (level.deserialized) {
+ val diskValues = serializerManager.dataDeserializeStream(
+ blockId,
+ diskData.toInputStream())(info.classTag)
+ maybeCacheDiskValuesInMemory(info, blockId, level, diskValues)
+ } else {
+ val stream = maybeCacheDiskBytesInMemory(info, blockId, level,
diskData)
+ .map { _.toInputStream(dispose = false) }
+ .getOrElse { diskData.toInputStream() }
+ serializerManager.dataDeserializeStream(blockId,
stream)(info.classTag)
+ }
+ }
+ val ci = CompletionIterator[Any, Iterator[Any]](iterToReturn, {
+ releaseLockAndDispose(blockId, diskData, taskContext)
+ })
+ Some(new BlockResult(ci, DataReadMethod.Disk, info.size))
+ } catch {
+ case ex: KryoException
+ // We need to have clear error message to catch environmental problems
easily.
+ // Further details: https://jira2.workday.com/browse/PRISM-102331
+ if ex.getMessage.toLowerCase(Locale.ROOT)
+ .contains("java.io.ioexception: input/output error") => {
+ processKryoException(ex, blockId)
+ throw ex
+ }
+ }
+ }
+
+ private def processKryoException(ex: KryoException, blockId: BlockId): Unit
= {
+ val errorMessage =
+ new StringBuffer(s"${ex.getMessage} usually occurs due to environmental
problems " +
Review comment:
Does this really provide more info?
--
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]
