From 2694e5e7d8098eef471e22773614ce633d6e673f Mon Sep 17 00:00:00 2001 From: Ankit Khullar Date: Fri, 29 May 2026 19:04:35 +0530 Subject: [PATCH 1/2] Implement getErrorClass explicitly in connector SparkThrowable exceptions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PulsarIllegalStateException and PulsarIllegalArgumentException implement SparkThrowable but override only getCondition, relying on the deprecated SparkThrowable.getErrorClass default (which delegates to getCondition). Some Spark distributions — notably Databricks Runtime 18 (Spark 4.1) — leave getErrorClass abstract rather than providing that default. When one of these exceptions escapes a task, Spark's TaskResultGetter virtual-calls getErrorClass(), finds no implementation, and raises: java.lang.AbstractMethodError: 'java.lang.String org.apache.spark.SparkThrowable.getErrorClass()' at org.apache.spark.sql.pulsar.PulsarIllegalStateException.getErrorClass This kills the result-getter thread, so the original failure (e.g. an incompatible-schema produce) is never reported and the Spark job hangs indefinitely instead of failing. Implement getErrorClass explicitly (delegating to the error class, same as getCondition) so the body lives on the class itself. This is harmless on OSS Spark, where it simply overrides the deprecated default, and restores correct error propagation on distributions that drop it. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../spark/sql/pulsar/PulsarExceptions.scala | 11 +++++ .../sql/pulsar/PulsarExceptionsSuite.scala | 42 +++++++++++++++++++ 2 files changed, 53 insertions(+) create mode 100644 src/test/scala/org/apache/spark/sql/pulsar/PulsarExceptionsSuite.scala diff --git a/src/main/scala/org/apache/spark/sql/pulsar/PulsarExceptions.scala b/src/main/scala/org/apache/spark/sql/pulsar/PulsarExceptions.scala index dee4445..3313d85 100644 --- a/src/main/scala/org/apache/spark/sql/pulsar/PulsarExceptions.scala +++ b/src/main/scala/org/apache/spark/sql/pulsar/PulsarExceptions.scala @@ -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 } @@ -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 } diff --git a/src/test/scala/org/apache/spark/sql/pulsar/PulsarExceptionsSuite.scala b/src/test/scala/org/apache/spark/sql/pulsar/PulsarExceptionsSuite.scala new file mode 100644 index 0000000..682dc7d --- /dev/null +++ b/src/test/scala/org/apache/spark/sql/pulsar/PulsarExceptionsSuite.scala @@ -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.pulsarSinkInvalidSchema + assert(e.getCondition === "PULSAR_SINK_INVALID_SCHEMA") + val method = classOf[PulsarIllegalArgumentException].getDeclaredMethod("getErrorClass") + assert(method.invoke(e) === "PULSAR_SINK_INVALID_SCHEMA") + } +} From 320acfaae2f972874dbe008031a958d44b1b5ed3 Mon Sep 17 00:00:00 2001 From: Ankit Khullar Date: Fri, 29 May 2026 19:20:01 +0530 Subject: [PATCH 2/2] Fix test to use a defined error class PulsarExceptions.pulsarSinkInvalidSchema references the error class "PULSAR_SINK_INVALID_SCHEMA", which is not present in error/pulsar-error-classes.json (only "PULSAR_SINK_INVALID_SCHEMA_TYPE" is), so constructing it raises INTERNAL_ERROR. Use pulsarProviderInvalidSaveMode, whose error class is defined, to exercise PulsarIllegalArgumentException. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../org/apache/spark/sql/pulsar/PulsarExceptionsSuite.scala | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/test/scala/org/apache/spark/sql/pulsar/PulsarExceptionsSuite.scala b/src/test/scala/org/apache/spark/sql/pulsar/PulsarExceptionsSuite.scala index 682dc7d..edbf73f 100644 --- a/src/test/scala/org/apache/spark/sql/pulsar/PulsarExceptionsSuite.scala +++ b/src/test/scala/org/apache/spark/sql/pulsar/PulsarExceptionsSuite.scala @@ -34,9 +34,9 @@ class PulsarExceptionsSuite extends SparkFunSuite { } test("PulsarIllegalArgumentException declares its own getErrorClass") { - val e = PulsarExceptions.pulsarSinkInvalidSchema - assert(e.getCondition === "PULSAR_SINK_INVALID_SCHEMA") + val e = PulsarExceptions.pulsarProviderInvalidSaveMode("Append") + assert(e.getCondition === "PULSAR_PROVIDER_INVALID_SAVE_MODE") val method = classOf[PulsarIllegalArgumentException].getDeclaredMethod("getErrorClass") - assert(method.invoke(e) === "PULSAR_SINK_INVALID_SCHEMA") + assert(method.invoke(e) === "PULSAR_PROVIDER_INVALID_SAVE_MODE") } }