diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index d27b0c02b1d5c..c695b4cc7309a 100644 --- a/common/utils/src/main/resources/error/error-conditions.json +++ b/common/utils/src/main/resources/error/error-conditions.json @@ -11411,16 +11411,6 @@ "Can't run submitMapStage on RDD with 0 partitions" ] }, - "_LEGACY_ERROR_TEMP_3024" : { - "message" : [ - "attempted to access non-existent accumulator " - ] - }, - "_LEGACY_ERROR_TEMP_3025" : { - "message" : [ - "TaskSetManagers should only send Resubmitted task statuses for tasks in ShuffleMapStages." - ] - }, "_LEGACY_ERROR_TEMP_3026" : { "message" : [ "duration() called on unfinished task" @@ -11436,31 +11426,11 @@ "Exiting due to error from cluster scheduler: " ] }, - "_LEGACY_ERROR_TEMP_3030" : { - "message" : [ - "Task has not locked block for writing" - ] - }, - "_LEGACY_ERROR_TEMP_3031" : { - "message" : [ - "Block does not exist" - ] - }, - "_LEGACY_ERROR_TEMP_3032" : { - "message" : [ - "Error occurred while waiting for replication to finish" - ] - }, "_LEGACY_ERROR_TEMP_3033" : { "message" : [ "Unable to register with external shuffle server due to : " ] }, - "_LEGACY_ERROR_TEMP_3034" : { - "message" : [ - "Error occurred while waiting for async. reregistration" - ] - }, "_LEGACY_ERROR_TEMP_3035" : { "message" : [ "Unexpected shuffle block with unsupported shuffle resolver " @@ -11476,31 +11446,6 @@ "Block was not found even though it's read-locked" ] }, - "_LEGACY_ERROR_TEMP_3038" : { - "message" : [ - "get() failed for block even though we held a lock" - ] - }, - "_LEGACY_ERROR_TEMP_3039" : { - "message" : [ - "BlockManager returned null for BlockStatus query: " - ] - }, - "_LEGACY_ERROR_TEMP_3040" : { - "message" : [ - "BlockManagerMasterEndpoint returned false, expected true." - ] - }, - "_LEGACY_ERROR_TEMP_3041" : { - "message" : [ - "" - ] - }, - "_LEGACY_ERROR_TEMP_3042" : { - "message" : [ - "Failed to get block , which is not a shuffle block" - ] - }, "_LEGACY_ERROR_TEMP_3052" : { "message" : [ "Unexpected resolved action: " diff --git a/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala b/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala index 4813f3f94107e..873254e491f25 100644 --- a/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala +++ b/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala @@ -248,15 +248,12 @@ private[spark] object SparkCoreErrors { } def accessNonExistentAccumulatorError(id: Long): Throwable = { - new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3024", messageParameters = Map("id" -> s"$id"), cause = null - ) + SparkException.internalError(s"Attempted to access non-existent accumulator $id.") } def sendResubmittedTaskStatusForShuffleMapStagesOnlyError(): Throwable = { - new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3025", messageParameters = Map.empty, cause = null - ) + SparkException.internalError( + "TaskSetManagers should only send Resubmitted task statuses for tasks in ShuffleMapStages.") } def nonEmptyEventQueueAfterTimeoutError(timeoutMillis: Long): Throwable = { @@ -292,22 +289,13 @@ private[spark] object SparkCoreErrors { } def taskHasNotLockedBlockError(currentTaskAttemptId: Long, blockId: BlockId): Throwable = { - new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3030", - messageParameters = Map( - "currentTaskAttemptId" -> s"$currentTaskAttemptId", - "blockId" -> s"$blockId" - ), - cause = null - ) + SparkException.internalError( + s"Task $currentTaskAttemptId has not locked block $blockId for writing.", + category = "STORAGE") } def blockDoesNotExistError(blockId: BlockId): Throwable = { - new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3031", - messageParameters = Map("blockId" -> s"$blockId"), - cause = null - ) + SparkException.internalError(s"Block $blockId does not exist.", category = "STORAGE") } def cannotSaveBlockOnDecommissionedExecutorError(blockId: BlockId): Throwable = { @@ -315,9 +303,7 @@ private[spark] object SparkCoreErrors { } def waitingForReplicationToFinishError(e: Throwable): Throwable = { - new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3032", messageParameters = Map.empty, cause = e - ) + SparkException.internalError("Error occurred while waiting for replication to finish.", e) } def unableToRegisterWithExternalShuffleServerError(e: Throwable): Throwable = { @@ -329,9 +315,7 @@ private[spark] object SparkCoreErrors { } def waitingForAsyncReregistrationError(e: Throwable): Throwable = { - new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3034", messageParameters = Map.empty, cause = e - ) + SparkException.internalError("Error occurred while waiting for async. reregistration.", e) } def unexpectedShuffleBlockWithUnsupportedResolverError( @@ -371,13 +355,8 @@ private[spark] object SparkCoreErrors { } def failToGetBlockWithLockError(blockId: BlockId): Throwable = { - new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3038", - messageParameters = Map( - "blockId" -> s"$blockId" - ), - cause = null - ) + SparkException.internalError( + s"get() failed for block $blockId even though we held a lock.", category = "STORAGE") } def blockNotFoundError(blockId: BlockId): Throwable = { @@ -389,17 +368,14 @@ private[spark] object SparkCoreErrors { } def blockStatusQueryReturnedNullError(blockId: BlockId): Throwable = { - new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3039", - messageParameters = Map("blockId" -> s"$blockId"), - cause = null - ) + SparkException.internalError( + s"BlockManager returned null for BlockStatus query: $blockId.", category = "STORAGE") } - def unexpectedBlockManagerMasterEndpointResultError(): Throwable = { - new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3040", messageParameters = Map.empty, cause = null - ) + def unexpectedBlockManagerMasterEndpointResultError(message: Any): Throwable = { + SparkException.internalError( + s"BlockManagerMasterEndpoint returned false for message $message, expected true.", + category = "STORAGE") } def failToCreateDirectoryError(path: String, maxAttempts: Int): Throwable = { @@ -407,10 +383,6 @@ private[spark] object SparkCoreErrors { s"Failed to create directory ${path} with permission 770 after $maxAttempts attempts!") } - def unsupportedOperationError(): Throwable = { - new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_3041") - } - def noSuchElementError(): Throwable = { new NoSuchElementException() } @@ -428,10 +400,10 @@ private[spark] object SparkCoreErrors { def failToGetNonShuffleBlockError(blockId: BlockId, e: Throwable): Throwable = { new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3042", - messageParameters = Map("blockId" -> s"$blockId"), - cause = e - ) + errorClass = "INTERNAL_ERROR_STORAGE", + messageParameters = Map( + "message" -> s"Failed to get block $blockId, which is not a shuffle block."), + cause = e) } def graphiteSinkInvalidProtocolError(invalidProtocol: String): Throwable = { diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala index 84a70d27047ca..47b6e2a50525f 100644 --- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala +++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala @@ -342,7 +342,7 @@ class BlockManagerMaster( /** Send a one-way message to the master endpoint, to which we expect it to reply with true. */ private def tell(message: Any): Unit = { if (!driverEndpoint.askSync[Boolean](message)) { - throw SparkCoreErrors.unexpectedBlockManagerMasterEndpointResultError() + throw SparkCoreErrors.unexpectedBlockManagerMasterEndpointResultError(message) } } diff --git a/core/src/main/scala/org/apache/spark/storage/DiskBlockObjectWriter.scala b/core/src/main/scala/org/apache/spark/storage/DiskBlockObjectWriter.scala index 9964e64724f6b..3ede609743879 100644 --- a/core/src/main/scala/org/apache/spark/storage/DiskBlockObjectWriter.scala +++ b/core/src/main/scala/org/apache/spark/storage/DiskBlockObjectWriter.scala @@ -22,8 +22,7 @@ import java.nio.channels.{ClosedByInterruptException, FileChannel} import java.nio.file.Files import java.util.zip.Checksum -import org.apache.spark.SparkException -import org.apache.spark.errors.SparkCoreErrors +import org.apache.spark.{SparkException, SparkUnsupportedOperationException} import org.apache.spark.internal.Logging import org.apache.spark.internal.LogKeys._ import org.apache.spark.io.MutableCheckedOutputStream @@ -338,7 +337,11 @@ private[spark] class DiskBlockObjectWriter( recordWritten() } - override def write(b: Int): Unit = throw SparkCoreErrors.unsupportedOperationError() + override def write(b: Int): Unit = throw new SparkUnsupportedOperationException( + errorClass = "UNSUPPORTED_CALL.WITHOUT_SUGGESTION", + messageParameters = Map( + "className" -> classOf[DiskBlockObjectWriter].getName, + "methodName" -> "write")) override def write(kvBytes: Array[Byte], offs: Int, len: Int): Unit = { if (!streamOpen) { diff --git a/core/src/test/scala/org/apache/spark/storage/BlockInfoManagerSuite.scala b/core/src/test/scala/org/apache/spark/storage/BlockInfoManagerSuite.scala index 4b34c13706ad8..016c49c2d1a4b 100644 --- a/core/src/test/scala/org/apache/spark/storage/BlockInfoManagerSuite.scala +++ b/core/src/test/scala/org/apache/spark/storage/BlockInfoManagerSuite.scala @@ -313,9 +313,12 @@ class BlockInfoManagerSuite extends SparkFunSuite { test("removing a non-existent block throws SparkException") { withTaskId(0) { - intercept[SparkException] { - blockInfoManager.removeBlock("non-existent-block") - } + checkError( + exception = intercept[SparkException] { + blockInfoManager.removeBlock("non-existent-block") + }, + condition = "INTERNAL_ERROR_STORAGE", + parameters = Map("message" -> "Block test_non-existent-block does not exist.")) } } diff --git a/core/src/test/scala/org/apache/spark/storage/DiskBlockObjectWriterSuite.scala b/core/src/test/scala/org/apache/spark/storage/DiskBlockObjectWriterSuite.scala index 4352436c872fe..f103f3c305cf3 100644 --- a/core/src/test/scala/org/apache/spark/storage/DiskBlockObjectWriterSuite.scala +++ b/core/src/test/scala/org/apache/spark/storage/DiskBlockObjectWriterSuite.scala @@ -21,7 +21,7 @@ import java.nio.ByteBuffer import scala.reflect.ClassTag -import org.apache.spark.{SparkConf, SparkException, SparkFunSuite} +import org.apache.spark.{SparkConf, SparkException, SparkFunSuite, SparkUnsupportedOperationException} import org.apache.spark.executor.ShuffleWriteMetrics import org.apache.spark.serializer.{DeserializationStream, JavaSerializer, SerializationStream, Serializer, SerializerInstance, SerializerManager} import org.apache.spark.util.Utils @@ -214,6 +214,18 @@ class DiskBlockObjectWriterSuite extends SparkFunSuite { assert(bs.isClosed) assert(objOut.isClosed) } + + test("write(b: Int) is not supported") { + val (writer, _, _) = createWriter() + checkError( + exception = intercept[SparkUnsupportedOperationException] { + writer.write(1) + }, + condition = "UNSUPPORTED_CALL.WITHOUT_SUGGESTION", + parameters = Map( + "className" -> "org.apache.spark.storage.DiskBlockObjectWriter", + "methodName" -> "write")) + } } trait CloseDetecting {