Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,13 @@ private[pulsar] class PulsarIllegalStateException(
override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava

override def getCondition: String = errorClass

// Some Spark distributions (notably Databricks Runtime) leave the deprecated
// SparkThrowable.getErrorClass abstract instead of defaulting to getCondition.
// Implement it explicitly so a thrown exception can be surfaced by Spark's
// TaskResultGetter instead of raising AbstractMethodError (which silently hangs
// the job). Harmless on OSS Spark, where it overrides the deprecated default.
override def getErrorClass: String = errorClass
}


Expand All @@ -125,4 +132,8 @@ private[pulsar] class PulsarIllegalArgumentException(
override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava

override def getCondition: String = errorClass

// See PulsarIllegalStateException above: explicit override guards against Spark
// builds (e.g. Databricks Runtime) that leave SparkThrowable.getErrorClass abstract.
override def getErrorClass: String = errorClass
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
/*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.spark.sql.pulsar

import org.apache.spark.SparkFunSuite

class PulsarExceptionsSuite extends SparkFunSuite {

// Regression guard for AbstractMethodError on Spark distributions that leave the
// deprecated SparkThrowable.getErrorClass abstract (e.g. Databricks Runtime). The
// connector exceptions must declare their OWN getErrorClass body; relying on the
// interface default causes Spark's TaskResultGetter to raise AbstractMethodError
// when the exception escapes a task, which silently hangs the job.

test("PulsarIllegalStateException declares its own getErrorClass") {
val e = PulsarExceptions.pulsarSinkIncompatibleSchema(
"persistent://public/default/t", new RuntimeException("boom"))
assert(e.getCondition === "PULSAR_SINK_INCOMPATIBLE_SCHEMA")
// getDeclaredMethod throws NoSuchMethodException if the body is not on the class
// itself (i.e. if it falls back to the SparkThrowable default).
val method = classOf[PulsarIllegalStateException].getDeclaredMethod("getErrorClass")
assert(method.invoke(e) === "PULSAR_SINK_INCOMPATIBLE_SCHEMA")
}

test("PulsarIllegalArgumentException declares its own getErrorClass") {
val e = PulsarExceptions.pulsarProviderInvalidSaveMode("Append")
assert(e.getCondition === "PULSAR_PROVIDER_INVALID_SAVE_MODE")
val method = classOf[PulsarIllegalArgumentException].getDeclaredMethod("getErrorClass")
assert(method.invoke(e) === "PULSAR_PROVIDER_INVALID_SAVE_MODE")
}
}
Loading