HyukjinKwon commented on code in PR #59219:
URL: https://github.com/apache/spark/pull/59219#discussion_r4216170281
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSourceProvider.scala:
##########
@@ -374,13 +374,16 @@ private[kafka010] class KafkaSourceProvider extends
DataSourceRegister
STARTING_OFFSETS_OPTION_KEY, EarliestOffsetRangeLimit) match {
case EarliestOffsetRangeLimit => // good to go
case LatestOffsetRangeLimit =>
- throw new IllegalArgumentException("starting offset can't be latest " +
- "for batch queries on Kafka")
+ throw KafkaExceptions.invalidBatchOffset(
Review Comment:
**Non-blocking (P2):** This now throws a `KafkaIllegalArgumentException`
whose condition, `KAFKA_INVALID_BATCH_OFFSET.GLOBAL` or `.TOPIC_OR_PARTITION`,
is defined only in the connector's `kafka-error-conditions.json`. Classic
sessions get exactly the intended error. Spark Connect Scala clients do not.
- The Connect server forwards the condition and parameters (`ErrorUtils`).
- On the client, `GrpcExceptionConverter.errorsToThrowable` maps the
`java.lang.IllegalArgumentException` hierarchy to `new
SparkIllegalArgumentException(errorClass = ...)`.
- That constructor re-renders the message through
`SparkThrowableHelper.getMessage`, which reads only the core
`error-conditions.json`.
- `ErrorClassesJsonReader.getMessageTemplate` then throws `[INTERNAL_ERROR]
Cannot find main error class 'KAFKA_INVALID_BATCH_OFFSET.GLOBAL'`. Nothing
between that constructor and `convert` catches it.
So on a Scala Connect client,
`spark.read.format("kafka").option("startingOffsets",
"latest")...load().collect()` (or `endingOffsets=earliest`, or a per-topic or
per-partition JSON entry) fails with INTERNAL_ERROR (SQLSTATE XX000). Before
this PR, these inputs carried no condition and reached the client as an
`IllegalArgumentException` (legacy class `_LEGACY_ERROR_TEMP_3105`) with the
actionable message. Now they lose the option name, the exception type, and the
new 22023 SQLSTATE. Existing connector-only Kafka conditions, such as
`KAFKA_DISALLOWED_OPTION`, already hit the same converter gap. This change
extends it to a common misconfiguration, and the new tests run only in classic
sessions.
**Recommended change:** In the Spark Connect Scala client's exception
converter, stop re-rendering a server condition that the client's error
registry does not define. Build the mapped exception from the server-supplied
message, keeping the server's errorClass, messageParameters, query context, and
sqlState. Add a private[spark] message-preserving construction path on the
affected Spark exception classes where one is not already reachable. Cover it
with Connect client converter tests.
**Why this works:** Gate the existing re-rendering constructors on
SparkThrowableHelper.isValidErrorClass. Conditions known to the client keep
today's construction. For an unknown condition, the converter uses the server
message, which already contains the rendered text, so no registry lookup can
throw. The exception stays an IllegalArgumentException (or the matching mapped
type) and still reports KAFKA_INVALID_BATCH_OFFSET.* with SQLSTATE 22023.
**Scope:** Connect client error reconstruction learns to fall back to the
server-rendered message for conditions outside its local registry, with
converter-level tests. The Kafka connector change stays as proposed.
**Compatibility:** Client-side rendering of conditions defined in the core
error registry, and the legacy-class fallback for errors without a condition,
remain unchanged.
**Risks:** A fallback that skips client-side rendering must not change
messages for conditions the client already knows; gate strictly on registry
membership. New private[spark] constructors must not collide with existing
overloads or alter the Connect JVM client's checked binary surface.
**Constraints:** Keep the server-supplied message as the source of truth for
unknown conditions; do not rebuild it from possibly truncated parameters.
**Success:** On a Scala Connect client, Kafka batch reads with
startingOffsets=latest or endingOffsets=earliest (global, topic, or partition)
fail with an IllegalArgumentException whose message names the option, condition
KAFKA_INVALID_BATCH_OFFSET.GLOBAL or .TOPIC_OR_PARTITION, and SQLSTATE 22023,
not INTERNAL_ERROR. Server conditions defined in the client's core registry,
and errors without a condition, reconstruct exactly as before.
**Verification:**
- **Regression:** Converting a server error in the IllegalArgumentException
hierarchy whose condition is absent from the client registry (with message
parameters and sqlState 22023) yields an IllegalArgumentException carrying the
server message, that condition, those parameters, and that SQLSTATE, with no
internal error.
- **Compatibility:** For every mapped exception type, a condition present in
the core registry still renders client-side with the same message, condition,
parameters, and SQLSTATE, and a missing condition still falls back to its
legacy error class.
- **Compile:** The common-utils and Connect client modules build, and the
Connect JVM client's compatibility check still passes with the added
construction path.
--
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]
