Serverless Scala JAR: Scala UDFs that read `Row` input fail

[ZZX]

### 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`? 

[Louis_Frolio]

Hi @ZZX ,

First, thank you for the quality of this write-up. The pass/fail matrix, the captured cause chain, and the signature comparison across Apache Spark 4.0.0 and the published Databricks Connect clients make this one of the cleaner bug reports I've seen on this board. You did the hard part already.

Your diagnosis holds together. The $default$4 in RowEncoder$.encoderForDataType$default$4() is a Scala compiler-generated accessor for a default argument, so a NoSuchMethodError there means the server-side sandbox UDF runner was compiled against a Spark build whose encoderForDataType carries a fourth parameter, while the classes loaded at execution time only have three. Every failing frame sits in the Databricks safe-execution code, not in your closure or your HTTP client, so this reads as a platform compatibility defect rather than an application problem. It also explains your matrix neatly: the Row deserializer is built lazily on the first next(), so empty closures, count(), and typed Dataset inputs (which use your case class encoder instead of the runtime RowEncoder) never touch the broken path. The versions you list also match the public compatibility guidance for environment versions 4 and 5, which further points away from user error.

On your two questions, here's my honest take:

1. I can't find anything in the public serverless release notes or docs that acknowledges this as a known issue, so I can't confirm which release introduced it or whether a fix is already deployed. Your evidence pointing at a DBR 18-based serverless rollout is plausible, but only Databricks engineering can confirm that from the inside. This deserves a formal support ticket at https://help.databricks.com. Include the minimal repro, the exact environment version, the resolved databricks-connect_2.13 version, your JAR dependency tree (confirming Spark and Connect classes stay provided and aren't bundled), and the full cause chain you captured. Ask them directly whether the server-side RowEncoder signature and the client-visible API are out of sync.

2. Agreed that the error surfacing is the second bug here. The real NoSuchMethodError being swallowed behind UDF_ERROR.INTERNAL / RST_STREAM PROTOCOL_ERROR is the same masking pattern seen in the earlier thread on this board, Serverless Scala JAR: foreachBatch fails with RST_STREAM PROTOCOL_ERROR, where you added your findings and where @AbhilashNagilla (Databricks) and @GabFernandes shared useful diagnostics guidance. Your trick of wrapping the closure in Try and writing the cause chain to a Unity Catalog volume is a genuinely useful workaround for the observability gap, and worth calling out in the ticket as its own issue.

In the meantime, you've already found the practical workaround: stay on the typed path and avoid materializing Row at all. Select only the columns you need and convert to a case class before foreachPartition:

case class Outbound(id: Long, payload: String)

val outbound = batch
  .select("id", "payload")
  .as[Outbound](Encoders.product[Outbound])

outbound.foreachPartition { rows =>
  // create or reuse the HTTP client per partition, then send rows
}

This sidesteps the runtime RowEncoder entirely, and your own table shows the full typed write to the external service commits cleanly. Keeping the projection narrow also keeps the Arrow transfer lean. If your schema is genuinely dynamic and can't be represented by a typed projection, the safer temporary option is running this workload on classic dedicated compute, where foreachBatch executes in the driver JVM. Note that standard access mode also runs on Spark Connect, so dedicated is the mode that truly bypasses this path.

A few references for anyone landing here later:

• Create and run JARs on serverless compute
• Databricks Connect usage requirements
• Serverless environment versions
• Serverless environment version 4 release notes
• Serverless compute limitations
• Databricks Support

The takeaway: your client build is correct, the skew appears to live server-side, and typed encoders are the safe path until a fix ships. Please do open that support case, and if you hear back on which release introduced the regression, posting the answer here would help the next fella who hits this.

Regards, Louis