[SPARK-51055][SS][CONNECT] Streaming foreachBatch should call init logic inside a try

[WweiL]

What changes were proposed in this pull request?

So that any error can be propagated to the jvm. Especially the python version issue.

handle_worker_exception would write out a special int to the jvm, and wil be caught here:

spark/core/src/main/scala/org/apache/spark/api/python/StreamingPythonRunner.scala

Lines 99 to 103 in c920210

	val resFromPython = dataIn.readInt()
	if (resFromPython != 0) {
	val errMessage = PythonWorkerUtils.readUTF(dataIn)
	throw streamingPythonRunnerInitializationFailure(resFromPython, errMessage)
	}

Why are the changes needed?

Spark Connect improvements

Does this PR introduce any user-facing change?

No

How was this patch tested?

Existing unit test

Was this patch authored or co-authored using generative AI tooling?

No

[hvanhovell]

LGTM