Skip to content

[SPARK-58461][CORE] Convert unreachable _LEGACY_ERROR_TEMP_3021-3042 invariants in SparkCoreErrors to internal errors #57663

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
LuciferYang wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[SPARK-58461][CORE] Convert unreachable _LEGACY_ERROR_TEMP_3021-3042 invariants in SparkCoreErrors to internal errors #57663

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 0 additions & 55 deletions common/utils/src/main/resources/error/error-conditions.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11411,16 +11411,6 @@
"Can't run submitMapStage on RDD with 0 partitions"
]
},
"_LEGACY_ERROR_TEMP_3024" : {
"message" : [
"attempted to access non-existent accumulator <id>"
]
},
"_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"
Expand All @@ -11436,31 +11426,11 @@
"Exiting due to error from cluster scheduler: <message>"
]
},
"_LEGACY_ERROR_TEMP_3030" : {
"message" : [
"Task <currentTaskAttemptId> has not locked block <blockId> for writing"
]
},
"_LEGACY_ERROR_TEMP_3031" : {
"message" : [
"Block <blockId> 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 : <message>"
]
},
"_LEGACY_ERROR_TEMP_3034" : {
"message" : [
"Error occurred while waiting for async. reregistration"
]
},
"_LEGACY_ERROR_TEMP_3035" : {
"message" : [
"Unexpected shuffle block <blockId> with unsupported shuffle resolver <shuffleBlockResolver>"
Expand All @@ -11476,31 +11446,6 @@
"Block <blockId> was not found even though it's read-locked"
]
},
"_LEGACY_ERROR_TEMP_3038" : {
"message" : [
"get() failed for block <blockId> even though we held a lock"
]
},
"_LEGACY_ERROR_TEMP_3039" : {
"message" : [
"BlockManager returned null for BlockStatus query: <blockId>"
]
},
"_LEGACY_ERROR_TEMP_3040" : {
"message" : [
"BlockManagerMasterEndpoint returned false, expected true."
]
},
"_LEGACY_ERROR_TEMP_3041" : {
"message" : [
""
]
},
"_LEGACY_ERROR_TEMP_3042" : {
"message" : [
"Failed to get block <blockId>, which is not a shuffle block"
]
},
"_LEGACY_ERROR_TEMP_3052" : {
"message" : [
"Unexpected resolved action: <other>"
Expand Down
70 changes: 21 additions & 49 deletions core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 = {
Expand Down Expand Up @@ -292,32 +289,21 @@ 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 = {
new BlockSavedOnDecommissionedBlockManagerException(blockId)
}

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 = {
Expand All @@ -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(
Expand Down Expand Up @@ -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 = {
Expand All @@ -389,28 +368,21 @@ 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 = {
new IOException(
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()
}
Expand All @@ -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 = {
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
}

Expand Down
9 changes: 6 additions & 3 deletions core/src/main/scala/org/apache/spark/storage/DiskBlockObjectWriter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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) {
Expand Down
9 changes: 6 additions & 3 deletions core/src/test/scala/org/apache/spark/storage/BlockInfoManagerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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."))
}
}

Expand Down
14 changes: 13 additions & 1 deletion core/src/test/scala/org/apache/spark/storage/DiskBlockObjectWriterSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 {
Expand Down