You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "WweiL (via GitHub)" <gi...@apache.org> on 2025/02/04 00:16:10 UTC
[PR] [SPARK-xxxxx][SS][CONNECT] Set timeout for communicating init configs in streaming foreachBatch [spark]
WweiL opened a new pull request, #49784:
URL: https://github.com/apache/spark/pull/49784
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
We observed that when the python process is stuck during initialization (sending init configs, e.g. serialized python function), the jvm has no way to know. As a result, there are cases of the jvm stuck at `dataIn.readInt()` ƒrom a stuck python process. Then the job won't be showing as failed, although no queries are successfully started.
Here we add a timeout for this socket during initialization so we fail fast.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
Spark Connect improvement
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
I can't find a way to write a integration test, so I test it by manually setting the timeout from jvm side to 5 seconds, and let the python foreach_batch_worker to sleep for 30 sec in this PR: https://github.com/apache/spark/pull/49783
And I was able to see the error:
```
>>> q = df.writeStream.foreachBatch(feb2).trigger(processingTime="30 seconds").start()
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/Users/wei.liu/oss-spark/python/pyspark/sql/connect/streaming/readwriter.py", line 656, in start
return self._start_internal(
^^^^^^^^^^^^^^^^^^^^^
File "/Users/wei.liu/oss-spark/python/pyspark/sql/connect/streaming/readwriter.py", line 625, in _start_internal
(_, properties, _) = self._session.client.execute_command(cmd)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/wei.liu/oss-spark/python/pyspark/sql/connect/client/core.py", line 1121, in execute_command
data, _, metrics, observed_metrics, properties = self._execute_and_fetch(
^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/wei.liu/oss-spark/python/pyspark/sql/connect/client/core.py", line 1533, in _execute_and_fetch
for response in self._execute_and_fetch_as_iterator(
File "/Users/wei.liu/oss-spark/python/pyspark/sql/connect/client/core.py", line 1510, in _execute_and_fetch_as_iterator
self._handle_error(error)
File "/Users/wei.liu/oss-spark/python/pyspark/sql/connect/client/core.py", line 1784, in _handle_error
self._handle_rpc_error(error)
File "/Users/wei.liu/oss-spark/python/pyspark/sql/connect/client/core.py", line 1860, in _handle_rpc_error
raise convert_exception(
pyspark.errors.exceptions.connect.SparkConnectGrpcException: (org.apache.spark.api.python.StreamingPythonRunner$StreamingPythonRunnerInitializationTimeoutException) [STREAMING_PYTHON_RUNNER_INITIALIZATION_TIMEOUT_FAILURE] Streaming Runner initialization failed. Socket connection timeout. Cause: Read timed out SQLSTATE: XXKST
JVM stacktrace:
org.apache.spark.api.python.StreamingPythonRunner$StreamingPythonRunnerInitializationTimeoutException
at org.apache.spark.api.python.StreamingPythonRunner.init(StreamingPythonRunner.scala:104)
at org.apache.spark.sql.connect.planner.StreamingForeachBatchHelper$.pythonForeachBatchWrapper(StreamingForeachBatchHelper.scala:140)
at org.apache.spark.sql.connect.planner.SparkConnectPlanner.handleWriteStreamOperationStart(SparkConnectPlanner.scala:3058)
at org.apache.spark.sql.connect.planner.SparkConnectPlanner.process(SparkConnectPlanner.scala:2483)
at org.apache.spark.sql.connect.execution.ExecuteThreadRunner.handleCommand(ExecuteThreadRunner.scala:322)
at org.apache.spark.sql.connect.execution.ExecuteThreadRunner.$anonfun$executeInternal$1(ExecuteThreadRunner.scala:224)
at org.apache.spark.sql.connect.execution.ExecuteThreadRunner.$anonfun$executeInternal$1$adapted(ExecuteThreadRunner.scala:196)
at org.apache.spark.sql.connect.service.SessionHolder.$anonfun$withSession$2(SessionHolder.scala:341)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:794)
at org.apache.spark.sql.connect.service.SessionHolder.$anonfun$withSession$1(SessionHolder.scala:341)
at org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)
at org.apache.spark.sql.artifact.ArtifactManager.$anonfun$withResources$1(ArtifactManager.scala:112)
at org.apache.spark.sql.artifact.ArtifactManager.withClassLoaderIfNeeded(ArtifactManager.scala:106)
at org.apache.spark.sql.artifact.ArtifactManager.withResources(ArtifactManager.scala:111)
at org.apache.spark.sql.connect.service.SessionHolder.withSession(SessionHolder.scala:340)
at org.apache.spark.sql.connect.execution.ExecuteThreadRunner.executeInternal(ExecuteThreadRunner.scala:196)
at org.apache.spark.sql.connect.execution.ExecuteThreadRunner.org$apache$spark$sql$connect$execution$ExecuteThreadRunner$$execute(ExecuteThreadRunner.scala:125)
at org.apache.spark.sql.connect.execution.ExecuteThreadRunner$ExecutionThread.run(ExecuteThreadRunner.scala:347)
```
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
No
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51074][SS][CONNECT] Set timeout for communicating init configs in streaming foreachBatch [spark]
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on code in PR #49784:
URL: https://github.com/apache/spark/pull/49784#discussion_r1940327384
##########
core/src/main/scala/org/apache/spark/api/python/StreamingPythonRunner.scala:
##########
@@ -76,47 +78,58 @@ private[spark] class StreamingPythonRunner(
pythonWorker = Some(worker)
pythonWorkerFactory = Some(workerFactory)
- logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Python worker created")
-
- val stream = new BufferedOutputStream(
- pythonWorker.get.channel.socket().getOutputStream, bufferSize)
+ val socket = pythonWorker.get.channel.socket()
+ val stream = new BufferedOutputStream(socket.getOutputStream, bufferSize)
+ val dataIn = new DataInputStream(new BufferedInputStream(socket.getInputStream, bufferSize))
val dataOut = new DataOutputStream(stream)
- logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Sending necessary information to the " +
- log"Python worker")
+ // Set timeout to 5 minute during initialization config transmission
+ socket.setSoTimeout(5 * 60 * 1000)
- PythonWorkerUtils.writePythonVersion(pythonVer, dataOut)
+ val resFromPython = try {
+ PythonWorkerUtils.writePythonVersion(pythonVer, dataOut)
- // Send sessionId
- if (!sessionId.isEmpty) {
- PythonRDD.writeUTF(sessionId, dataOut)
- }
+ // Send sessionId
+ if (!sessionId.isEmpty) {
+ PythonRDD.writeUTF(sessionId, dataOut)
+ }
- // Send the user function to python process
- PythonWorkerUtils.writePythonFunction(func, dataOut)
- dataOut.flush()
+ // Send the user function to python process
+ PythonWorkerUtils.writePythonFunction(func, dataOut)
+ dataOut.flush()
- logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Reading initialization response from " +
- log"Python runner.")
+ logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Reading initialization response from " +
+ log"Python runner.")
+ dataIn.readInt()
+ } catch {
+ case e: java.net.SocketTimeoutException =>
+ throw new StreamingPythonRunnerInitializationTimeoutException(e.getMessage)
+ case e: Exception =>
+ throw new StreamingPythonRunnerInitializationCommunicationException(e.getMessage)
+ }
- val dataIn = new DataInputStream(
- new BufferedInputStream(pythonWorker.get.channel.socket().getInputStream, bufferSize))
+ // Set timeout back to infinite because code execution time can be arbitrary long
+ socket.setSoTimeout(0)
Review Comment:
Thanks for the suggestion! I mitigated with first reading the timeout and then set it back
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51074][SS][CONNECT] Set timeout for communicating init configs in streaming foreachBatch [spark]
Posted by "siying (via GitHub)" <gi...@apache.org>.
siying commented on code in PR #49784:
URL: https://github.com/apache/spark/pull/49784#discussion_r1941993950
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -4850,12 +4850,24 @@
},
"sqlState" : "42KDE"
},
+ "STREAMING_PYTHON_RUNNER_INITIALIZATION_COMMUNICATION_FAILURE" : {
+ "message" : [
+ "Streaming Runner initialization failed during initial config communication. Cause: <msg>"
+ ],
+ "sqlState" : "XXKST"
+ },
"STREAMING_PYTHON_RUNNER_INITIALIZATION_FAILURE" : {
"message" : [
"Streaming Runner initialization failed, returned <resFromPython>. Cause: <msg>"
],
"sqlState" : "XXKST"
},
+ "STREAMING_PYTHON_RUNNER_INITIALIZATION_TIMEOUT_FAILURE" : {
+ "message" : [
+ "Streaming Runner initialization failed. Socket connection timeout. Cause: <msg>"
+ ],
+ "sqlState" : "XXKST"
Review Comment:
I'm not sure about which error code to use. Perhaps ask someone who is familiar with this to review.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-xxxxx][SS][CONNECT] Set timeout for communicating init configs in streaming foreachBatch [spark]
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on code in PR #49784:
URL: https://github.com/apache/spark/pull/49784#discussion_r1940302582
##########
core/src/main/scala/org/apache/spark/api/python/StreamingPythonRunner.scala:
##########
@@ -76,47 +78,58 @@ private[spark] class StreamingPythonRunner(
pythonWorker = Some(worker)
pythonWorkerFactory = Some(workerFactory)
- logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Python worker created")
-
- val stream = new BufferedOutputStream(
- pythonWorker.get.channel.socket().getOutputStream, bufferSize)
+ val socket = pythonWorker.get.channel.socket()
+ val stream = new BufferedOutputStream(socket.getOutputStream, bufferSize)
+ val dataIn = new DataInputStream(new BufferedInputStream(socket.getInputStream, bufferSize))
val dataOut = new DataOutputStream(stream)
- logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Sending necessary information to the " +
- log"Python worker")
+ // Set timeout to 5 minute during initialization config transmission
+ socket.setSoTimeout(5 * 60 * 1000)
- PythonWorkerUtils.writePythonVersion(pythonVer, dataOut)
+ val resFromPython = try {
+ PythonWorkerUtils.writePythonVersion(pythonVer, dataOut)
- // Send sessionId
- if (!sessionId.isEmpty) {
- PythonRDD.writeUTF(sessionId, dataOut)
- }
+ // Send sessionId
+ if (!sessionId.isEmpty) {
+ PythonRDD.writeUTF(sessionId, dataOut)
+ }
- // Send the user function to python process
- PythonWorkerUtils.writePythonFunction(func, dataOut)
- dataOut.flush()
+ // Send the user function to python process
+ PythonWorkerUtils.writePythonFunction(func, dataOut)
+ dataOut.flush()
- logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Reading initialization response from " +
- log"Python runner.")
+ logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Reading initialization response from " +
+ log"Python runner.")
+ dataIn.readInt()
+ } catch {
+ case e: java.net.SocketTimeoutException =>
+ throw new StreamingPythonRunnerInitializationTimeoutException(e.getMessage)
+ case e: Exception =>
+ throw new StreamingPythonRunnerInitializationCommunicationException(e.getMessage)
+ }
- val dataIn = new DataInputStream(
- new BufferedInputStream(pythonWorker.get.channel.socket().getInputStream, bufferSize))
+ // Set timeout back to infinite because code execution time can be arbitrary long
+ socket.setSoTimeout(0)
- val resFromPython = dataIn.readInt()
if (resFromPython != 0) {
val errMessage = PythonWorkerUtils.readUTF(dataIn)
- throw streamingPythonRunnerInitializationFailure(resFromPython, errMessage)
+ throw new StreamingPythonRunnerInitializationException(resFromPython, errMessage)
}
logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Runner initialization succeeded " +
log"(returned ${MDC(PYTHON_WORKER_RESPONSE, resFromPython)}).")
(dataOut, dataIn)
}
- def streamingPythonRunnerInitializationFailure(resFromPython: Int, errMessage: String):
- StreamingPythonRunnerInitializationException = {
- new StreamingPythonRunnerInitializationException(resFromPython, errMessage)
- }
Review Comment:
Refactored this a bit, didn't see a reason why this this has to be a function
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51074][SS][CONNECT] Set timeout for communicating init configs in streaming foreachBatch [spark]
Posted by "siying (via GitHub)" <gi...@apache.org>.
siying commented on code in PR #49784:
URL: https://github.com/apache/spark/pull/49784#discussion_r1941716733
##########
core/src/main/scala/org/apache/spark/api/python/StreamingPythonRunner.scala:
##########
@@ -76,47 +78,60 @@ private[spark] class StreamingPythonRunner(
pythonWorker = Some(worker)
pythonWorkerFactory = Some(workerFactory)
- logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Python worker created")
-
- val stream = new BufferedOutputStream(
- pythonWorker.get.channel.socket().getOutputStream, bufferSize)
+ val socket = pythonWorker.get.channel.socket()
+ val stream = new BufferedOutputStream(socket.getOutputStream, bufferSize)
+ val dataIn = new DataInputStream(new BufferedInputStream(socket.getInputStream, bufferSize))
val dataOut = new DataOutputStream(stream)
- logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Sending necessary information to the " +
- log"Python worker")
-
- PythonWorkerUtils.writePythonVersion(pythonVer, dataOut)
+ val originalTimeout = socket.getSoTimeout()
+ // Set timeout to 5 minute during initialization config transmission
+ socket.setSoTimeout(5 * 60 * 1000)
- // Send sessionId
- if (!sessionId.isEmpty) {
- PythonRDD.writeUTF(sessionId, dataOut)
- }
+ val resFromPython = try {
+ PythonWorkerUtils.writePythonVersion(pythonVer, dataOut)
- // Send the user function to python process
- PythonWorkerUtils.writePythonFunction(func, dataOut)
- dataOut.flush()
+ // Send sessionId
+ if (!sessionId.isEmpty) {
+ PythonRDD.writeUTF(sessionId, dataOut)
+ }
- logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Reading initialization response from " +
- log"Python runner.")
+ // Send the user function to python process
+ PythonWorkerUtils.writePythonFunction(func, dataOut)
+ dataOut.flush()
- val dataIn = new DataInputStream(
- new BufferedInputStream(pythonWorker.get.channel.socket().getInputStream, bufferSize))
+ logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Reading initialization response from " +
+ log"Python runner.")
+ dataIn.readInt()
+ } catch {
+ case e: java.net.SocketTimeoutException =>
+ throw new StreamingPythonRunnerInitializationTimeoutException(e.getMessage)
+ case e: Exception =>
+ throw new StreamingPythonRunnerInitializationCommunicationException(e.getMessage)
+ } finally {
+ // Set timeout back to the original timeout
+ // Should be infinity by default
+ socket.setSoTimeout(originalTimeout)
Review Comment:
I'm not familiar with Java's socket interface but is it safe to set it after some timeout against the socket without exception? If you've checked and it never throws exceptions I'm OK with this.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51074][SS][CONNECT] Set timeout for communicating init configs in streaming foreachBatch [spark]
Posted by "siying (via GitHub)" <gi...@apache.org>.
siying commented on code in PR #49784:
URL: https://github.com/apache/spark/pull/49784#discussion_r1940313360
##########
core/src/main/scala/org/apache/spark/api/python/StreamingPythonRunner.scala:
##########
@@ -76,47 +78,58 @@ private[spark] class StreamingPythonRunner(
pythonWorker = Some(worker)
pythonWorkerFactory = Some(workerFactory)
- logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Python worker created")
-
- val stream = new BufferedOutputStream(
- pythonWorker.get.channel.socket().getOutputStream, bufferSize)
+ val socket = pythonWorker.get.channel.socket()
+ val stream = new BufferedOutputStream(socket.getOutputStream, bufferSize)
+ val dataIn = new DataInputStream(new BufferedInputStream(socket.getInputStream, bufferSize))
val dataOut = new DataOutputStream(stream)
- logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Sending necessary information to the " +
- log"Python worker")
+ // Set timeout to 5 minute during initialization config transmission
+ socket.setSoTimeout(5 * 60 * 1000)
- PythonWorkerUtils.writePythonVersion(pythonVer, dataOut)
+ val resFromPython = try {
+ PythonWorkerUtils.writePythonVersion(pythonVer, dataOut)
- // Send sessionId
- if (!sessionId.isEmpty) {
- PythonRDD.writeUTF(sessionId, dataOut)
- }
+ // Send sessionId
+ if (!sessionId.isEmpty) {
+ PythonRDD.writeUTF(sessionId, dataOut)
+ }
- // Send the user function to python process
- PythonWorkerUtils.writePythonFunction(func, dataOut)
- dataOut.flush()
+ // Send the user function to python process
+ PythonWorkerUtils.writePythonFunction(func, dataOut)
+ dataOut.flush()
- logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Reading initialization response from " +
- log"Python runner.")
+ logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Reading initialization response from " +
+ log"Python runner.")
+ dataIn.readInt()
+ } catch {
+ case e: java.net.SocketTimeoutException =>
+ throw new StreamingPythonRunnerInitializationTimeoutException(e.getMessage)
+ case e: Exception =>
+ throw new StreamingPythonRunnerInitializationCommunicationException(e.getMessage)
+ }
- val dataIn = new DataInputStream(
- new BufferedInputStream(pythonWorker.get.channel.socket().getInputStream, bufferSize))
+ // Set timeout back to infinite because code execution time can be arbitrary long
+ socket.setSoTimeout(0)
Review Comment:
The change is OK for this specific issue. However, I'm worried that in the future, when someone added a timeout for `socket`, we are resetting it to 0, which confuses the person who sets the timeout and make it prone to error. An alternative is not to set the socket timeout, but do operations above in another thread, with the main thread doing timed wait and interrupting the other thread when the timeout hits.
If you consider that approach and have a good reason for the current choice, I'm fine with it.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51074][SS][CONNECT] Set timeout for communicating init configs in streaming foreachBatch [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #49784:
URL: https://github.com/apache/spark/pull/49784#issuecomment-2638295221
Merged to master and branch-4.0.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51074][SS][CONNECT] Set timeout for communicating init configs in streaming foreachBatch [spark]
Posted by "siying (via GitHub)" <gi...@apache.org>.
siying commented on code in PR #49784:
URL: https://github.com/apache/spark/pull/49784#discussion_r1940309868
##########
core/src/main/scala/org/apache/spark/api/python/StreamingPythonRunner.scala:
##########
@@ -76,47 +78,58 @@ private[spark] class StreamingPythonRunner(
pythonWorker = Some(worker)
pythonWorkerFactory = Some(workerFactory)
- logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Python worker created")
-
- val stream = new BufferedOutputStream(
- pythonWorker.get.channel.socket().getOutputStream, bufferSize)
+ val socket = pythonWorker.get.channel.socket()
+ val stream = new BufferedOutputStream(socket.getOutputStream, bufferSize)
+ val dataIn = new DataInputStream(new BufferedInputStream(socket.getInputStream, bufferSize))
val dataOut = new DataOutputStream(stream)
- logInfo(log"[session: ${MDC(SESSION_ID, sessionId)}] Sending necessary information to the " +
- log"Python worker")
+ // Set timeout to 5 minute during initialization config transmission
+ socket.setSoTimeout(5 * 60 * 1000)
Review Comment:
Thank you for adding it. It is important to add a timeout here.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51074][SS][CONNECT] Set timeout for communicating init configs in streaming foreachBatch [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #49784: [SPARK-51074][SS][CONNECT] Set timeout for communicating init configs in streaming foreachBatch
URL: https://github.com/apache/spark/pull/49784
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org