Serverless Scala JAR: foreachBatch fails with RST_STREAM PROTOCOL_ERROR

adhi_databricks
Contributor

Tuesday

Hi everyone,

We're migrating a Scala Structured Streaming application (Kinesis → Delta) from classic compute to Databricks Serverless Jobs for cost optimization. During the migration, we're consistently seeing what appears to be a Spark Connect / gRPC transport issue whenever foreachBatch is used.

Environment

  • Compute: Databricks Serverless Jobs (Spark JAR task)
  • Language: Scala (big JAR entry point; not a notebook and not using Databricks Connect)
  • Trigger: Trigger.AvailableNow() (required on Serverless)
  • Source: Databricks native Kinesis connector with Unity Catalog service credential
  • Checkpoint: S3 checkpoint location
  • Failed foreachBatch executions show empty start offsets (batch never commits)
  • Spark config: spark.sql.shuffle.partitions=auto

Minimal Reproduction

Works:

df.writeStream
.format("delta")
.option("checkpointLocation", checkpointPath)
.trigger(Trigger.AvailableNow())
.start(targetPath)

Fails (even with an empty foreachBatch):

df.writeStream
.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
// no-op
}
.option("checkpointLocation", checkpointPath)
.trigger(Trigger.AvailableNow())
.start()

Observed Behavior

  • ✅ Kinesis → AvailableNow → Delta sink → Success
  • ❌ Kinesis → AvailableNow → foreachBatch (empty) → Fails
  • ❌ Kinesis → AvailableNow → foreachBatch (full parse/merge/denorm) → Fails

Failure Symptoms

  • UDF_ERROR.INTERNAL
  • Underlying gRPC error: RST_STREAM with PROTOCOL_ERROR
  • Query History and Spark UI provide limited diagnostics for this JAR execution path
  • When execution plans are available, shuffle looks healthy (spark.sql.shuffle.partitions=auto)

Since the Delta sink succeeds while even an empty foreachBatch fails, this doesn't appear to be related to:

  • Kinesis authentication
  • Unity Catalog service credentials
  • Checkpoint configuration
  • Merge/business logic inside foreachBatch

What We've Already Ruled Out

  • Verified the Unity Catalog service credential (Kinesis ingestion succeeds using the Delta sink)
  • Removed persist()/cache() after previously hitting NOT_SUPPORTED_WITH_SERVERLESS
  • Removed logging-only .count() calls; only use limit(1).count() where needed
  • Avoid calling SparkSession.builder.getOrCreate() inside foreachBatch; use the existing Spark session
  • Confirmed AvailableNow + Kinesis initialPosition requirements (latest is invalid; testing with earliest and at_timestamp)

Questions

  1. Is Scala foreachBatch from a Serverless Jobs JAR officially supported today, or are there known limitations on the Spark Connect execution path used by Serverless?

  2. Is UDF_ERROR.INTERNAL wrapping RST_STREAM / PROTOCOL_ERROR a known Spark Connect transport issue with foreachBatch, rather than an actual UDF bug?

  3. Before opening a support case, are there any recommended diagnostics we should collect (specific log sections, Spark Connect debug flags, or minimal repro expectations)?

Happy to share a sanitized stack trace in a follow-up if that would be helpful.

Thanks!

Labels:

GabFernandes
Contributor

Tuesday

Hi @adhi_databricks !

This is a very specific and subtle architectural edge case when migrating traditional Scala JAR workloads to Databricks Serverless Compute.

The core issue comes down to how Databricks Serverless Jobs execute code via Spark Connect.

Unlike Classic Compute—where your Scala JAR runs directly inside the JVM on the driver node—Serverless Compute executes user code through the Spark Connect architecture (gRPC transport layer).

Here is the breakdown answering your specific questions:

1. Is Scala foreachBatch from a Serverless Jobs JAR officially supported?
While Scala JAR tasks are supported on Serverless, using arbitrary Scala closures within foreachBatch on Spark Connect / Serverless has known limitations.

In Classic Compute, foreachBatch executes a local JVM closure directly on the driver. In Serverless (Spark Connect), foreachBatch requires serializing and streaming function calls and DataFrame operations back and forth over a gRPC channel between the client context and the remote Spark Connect server.

When complex Scala closure state (or compiler-generated synthetic classes) is passed over this gRPC transport layer, it frequently triggers serialization or protocol mismatches, leading to stream termination.

2. Is UDF_ERROR.INTERNAL wrapping RST_STREAM / PROTOCOL_ERROR a known transport issue?
Yes. RST_STREAM with PROTOCOL_ERROR is a lower-level HTTP/2 and gRPC transport error.

It indicates that the gRPC channel between the Spark Connect driver client and the Serverless backend was abruptly closed due to an unhandled serialization failure or channel reset during the execution of the closure.

The UDF_ERROR.INTERNAL is simply a high-level Databricks wrapper error catching the failed RPC invocation. It is not an error in your business code logic, which is why even a no-op foreachBatch fails with the exact same error.

3. Workarounds & Alternatives
Until full parity for native Scala closures over Spark Connect in Serverless is reached, here are the recommended architectural workarounds:

Option A (Recommended for Serverless): Delta Live Tables (DLT) or Declarative Pipelines
If you are running streaming CDC/Kinesis pipelines to Delta with foreachBatch, moving the pipeline to Delta Live Tables (DLT) / Auto Loader with CDC APIs (apply_changes) completely bypasses foreachBatch and runs natively on Serverless without gRPC closure serialization overhead.

Option B: Run as a Classic Compute Job
If foreachBatch with custom Scala logic (complex merges/side-effects) is strictly required for your architecture, running this specific streaming task on a single-node or small Auto-scaling Classic Single-User/Shared Cluster remains the most stable path today.

Option C: Refactor to Pure DataFrame / SQL Writes
If your foreachBatch logic can be expressed purely via native Delta MERGE or DataFrame writes using .writeStream (without custom UDFs or Scala closures inside the batch function), Spark Connect can execute the logical plan natively.

4. Recommended Diagnostics Before Opening a Support Case
If you proceed with opening a Databricks Support ticket (which is recommended to help the product team track this Spark Connect edge case), gather the following:

log4j.logger.org.apache.spark.sql.connect=DEBUG
log4j.logger.io.grpc=DEBUG
  • Simplified Standalone Repro: Submit your minimal no-op foreachBatch snippet as a standalone single-class JAR job.
  • Driver Log4j Dump: Export the full driver log showing the exact gRPC stack trace right before RST_STREAM.
  • Hope this helps clarify why the Delta sink works while foreachBatch fails on Serverless!

If my answer was helpful, please consider marking it as accepted solution!

0 Kudos

AbhilashNagilla
Databricks Employee
Databricks Employee

Wednesday

On (1): the docs don't gate Scala foreachBatch on serverless. Limitations with Databricks Connect for Scala lists streaming foreachBatch as unavailable only on Databricks Runtime 13.3 LTS and below, and serverless limitations names only Trigger.ProcessingTime and Trigger.Continuous as blocked triggers. Serverless Scala and Java JAR jobs are in Public Preview. I'm reading published docs here, not speaking for support.

On (2): I can't find any published Databricks doc or KB that names RST_STREAM or PROTOCOL_ERROR, so I wouldn't put a label on it from out here. One thing worth raising with support: [UDF_ERROR](https://docs.databricks.com/aws/en/error-messages/udf-error-error-class) has a dedicated ENV_LOST sub-condition for an execution environment that was lost mid-run, and you're getting INTERNAL instead. Whether that distinction is meaningful here is a question for whoever picks up the ticket. Your empty no-op result already rules out the merge logic.

On (3), three things before the case:

  1. Make sure your application logs are actually being emitted. On serverless, the SLF4J logging API defaults to a no-op backend and application log output is silently discarded unless you add the log4j-slf4j2-impl bridge matching the log4j-api version in your environment version (2.20.0 on environment version 5). See Configure logging for serverless compute. One note on the log4j.logger.org.apache.spark.sql.connect=DEBUG suggestion above: Spark logs aren't available on serverless, only client-side application logs, only six Spark properties are settable and none are log4j-related, and changing the log4j level through SparkContext is listed as unavailable. Use the query profile in place of the Spark UI.

  2. Match Scala, JDK, and your Spark API dependency to your environment version. You mentioned you're not building against Databricks Connect; spark-sql-api is a documented alternative, but Databricks recommends databricks-connect marked provided, so that's worth ruling out. Don't bundle log4j-api or log4j-core.

  3. Isolate foreachBatch itself. Swap Kinesis for a small Delta table, Trigger.AvailableNow(), empty foreachBatch, single-class JAR. Delta is a documented serverless streaming source, so a failure there isolates foreachBatch cleanly.

If that still fails, it's a clean repro to attach to a support case. On fallback, note that standard access mode also runs on Spark Connect; dedicated compute is the mode that uses the classic Spark architecture.