### Setup
- Serverless jar task, environment version 4 (Databricks Connect 17.3.2, Scala 2.13.16, JDK 17). Also reproduced on version 5 (Databricks Connect 18.0.0).
- JAR built with databricks-connect_2.13 as provided
- Structured Streaming from a Unity Catalog Delta table, Trigger.AvailableNow(), foreachBatch → Dataset.foreachPartition with a Scala closure that posts rows to an external HTTP service.
### Symptom
Every batch with data fails. The only visible error:
```
[FOREACH_BATCH_USER_FUNCTION_ERROR] ... [UDF_ERROR.INTERNAL] Execution of function :foreach_batch failed with an
internal error: INTERNAL: RST_STREAM closed stream. HTTP/2 error code: PROTOCOL_ERROR
at com.databricks.sql.execution.safespark.ForeachBatchLakeguardSink.close(SafeSparkUDFRunner.scala:945)
```
### What narrows it down
| foreachBatch body | Result |
| batch.count() | ok |
| batch.foreachPartition((_: Iterator[Row]) => ()) | ok, batch commits |
| batch.limit(4).foreachPartition(_.foreach(_ => ())) | fails |
| batch.select("<one INT column>").foreachPartition(_.foreach(_ => ())) | fails |
| batch.as(Encoders.product[MyCaseClass]).foreachPartition(_.foreach(_ => ())) | ok |
| typed case class + full write to the external service | ok |
The only difference between pass and fail is whether the closure pulls a Row from the iterator.
### The real exception
Captured by wrapping the nested call in `Try` inside the `foreachBatch` function and writing the cause chain to a UC volume (the platform never shows it):
```
org.apache.spark.SparkRuntimeException: [UDF_USER_CODE_ERROR.GENERIC] Execution of function failed.
UDF error: UDF invocation failed. Error type: class java.lang.NoSuchMethodError.
Error message: 'boolean org.apache.spark.sql.catalyst.encoders.RowEncoder$.encoderForDataType$default$4()'
at com.databricks.spark.safespark.udf.utils.UDFUtils$.rowEncoder$1(UDFUtils.scala:107)
at com.databricks.spark.safespark.udf.utils.UDFUtils$.getDeserializers(UDFUtils.scala:118)
at com.databricks.spark.safespark.udf.utils.UDFUtils$.$anonfun$invokeIteratorIterator$1(UDFUtils.scala:320)
at com.databricks.spark.safespark.udf.utils.ArrowBasedExecution.processBatches(Payload.scala:124)
```
Every frame between the throw site and my closure is Databricks code. The Row deserializer is built lazily on the first next(), which is why empty closures and typed inputs pass.
`RowEncoder$.encoderForDataType` has 2 parameters in Apache Spark 4.0.0 and 3 parameters (only `$default$3`) in Databricks Connect 17.3.2, 18.0.0 and 18.3.4. The sandbox runtime calls `$default$4`, i.e. it was compiled against an internal server-side Spark with a 4-parameter signature that no published client has. Looks like a version skew introduced with the DBR 18-based serverless release.
### Minimal repro
```scala
spark.readStream.format("delta").table(table)
.writeStream.option("checkpointLocation", freshVolumePath)
.trigger(Trigger.AvailableNow())
.foreachBatch { (batch: DataFrame, _: Long) =>
batch.foreachPartition((_: Iterator[Row]) => ()) // passes
batch.limit(4).foreachPartition((rows: Iterator[Row]) => rows.foreach(_ => ())) // NoSuchMethodError
}
.start().awaitTermination()
```
### Questions
1. Is this a known regression, and which serverless release introduced it?
2. Could the real exception be surfaced instead of `UDF_ERROR.INTERNAL / RST_STREAM`?
