You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by pt...@apache.org on 2026/09/17 16:45:30 UTC
(spark) branch branch-4.3 updated: [MINOR] Improve resilence on raw socket stream and steer people towards higher level streaming APIs
This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new ecf1ff8f2b0a [MINOR] Improve resilence on raw socket stream and steer people towards higher level streaming APIs
ecf1ff8f2b0a is described below
commit ecf1ff8f2b0a1c7e5caa20643d96ade455f9fdce
Author: Holden Karau <ho...@snowflake.com>
AuthorDate: Tue Sep 1 06:33:31 2026 +0000
[MINOR] Improve resilence on raw socket stream and steer people towards higher level streaming APIs
Co-authored-by: Cursor <cu...@cursor.com>
Co-Authored-By: Claude Opus 5 (1M context) <no...@anthropic.com>
Co-Authored-By: Holden Karau <ho...@pigscanfly.ca>
### Backport notes for branch-4.3
`.version` reads `4.3.1`, this branch's next release. That is the only
difference from the master commit, and it matches what the other branches did:
`4.4.0` on branch-4.x, `4.2.1` on branch-4.2, `4.0.5` on branch-4.0, `3.5.10`
on branch-3.5.
(cherry picked from commit ca8e43ba97f9d94db19f0aa511c8fbfd37b9e731)
---
.../org/apache/spark/streaming/StreamingConf.scala | 11 ++++++++++
.../apache/spark/streaming/StreamingContext.scala | 5 +++++
.../spark/streaming/dstream/RawInputDStream.scala | 24 ++++++++++++++++++++--
3 files changed, 38 insertions(+), 2 deletions(-)
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/StreamingConf.scala b/streaming/src/main/scala/org/apache/spark/streaming/StreamingConf.scala
index 39f3d495bdbf..9785a7fba6e6 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/StreamingConf.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/StreamingConf.scala
@@ -20,6 +20,7 @@ package org.apache.spark.streaming
import java.util.concurrent.TimeUnit
import org.apache.spark.internal.config.ConfigBuilder
+import org.apache.spark.network.util.ByteUnit
import org.apache.spark.streaming.util.OpenHashMapBasedStateMap.DELTA_CHAIN_LENGTH_THRESHOLD
object StreamingConf {
@@ -41,6 +42,16 @@ object StreamingConf {
.version("2.0.0")
.fallbackConf(RECEIVER_MAX_RATE)
+ private[streaming] val RAW_INPUT_MAX_BLOCK_SIZE =
+ ConfigBuilder("spark.streaming.raw.maxBlockSizeBytes")
+ .doc("Maximum size in bytes of a single block frame accepted by rawSocketStream " +
+ "receivers. The receiver allocates a buffer of exactly the size announced in " +
+ "the frame header before reading the frame body; frames announcing a negative " +
+ "length or one larger than this limit are rejected with an error.")
+ .version("4.3.1")
+ .bytesConf(ByteUnit.BYTE)
+ .createWithDefaultString("1g")
+
private[streaming] val BLOCK_INTERVAL =
ConfigBuilder("spark.streaming.blockInterval")
.version("0.8.0")
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala b/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
index 139b83ba0d07..0e0843507af8 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
@@ -336,6 +336,11 @@ class StreamingContext private[streaming] (
* @param storageLevel Storage level to use for storing the received objects
* (default: StorageLevel.MEMORY_AND_DISK_SER_2)
* @tparam T Type of the objects in the received blocks
+ *
+ * @note rawSocketStream is a low-level API; most applications should prefer
+ * higher-level streaming sources. Frames announcing a negative length or one
+ * larger than `spark.streaming.raw.maxBlockSizeBytes` (default 1g) stop the
+ * receiver with an error.
*/
def rawSocketStream[T: ClassTag](
hostname: String,
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/dstream/RawInputDStream.scala b/streaming/src/main/scala/org/apache/spark/streaming/dstream/RawInputDStream.scala
index 8466e9f70843..1e66d394975c 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/dstream/RawInputDStream.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/dstream/RawInputDStream.scala
@@ -17,7 +17,7 @@
package org.apache.spark.streaming.dstream
-import java.io.EOFException
+import java.io.{EOFException, IOException}
import java.net.InetSocketAddress
import java.nio.ByteBuffer
import java.nio.channels.{ReadableByteChannel, SocketChannel}
@@ -25,9 +25,10 @@ import java.util.concurrent.ArrayBlockingQueue
import scala.reflect.ClassTag
+import org.apache.spark.SparkEnv
import org.apache.spark.internal.{Logging, LogKeys}
import org.apache.spark.storage.StorageLevel
-import org.apache.spark.streaming.StreamingContext
+import org.apache.spark.streaming.{StreamingConf, StreamingContext}
import org.apache.spark.streaming.receiver.Receiver
/**
@@ -56,6 +57,7 @@ class RawNetworkReceiver(host: String, port: Int, storageLevel: StorageLevel)
var blockPushingThread: Thread = null
def onStart(): Unit = {
+ val maxBlockSize = SparkEnv.get.conf.get(StreamingConf.RAW_INPUT_MAX_BLOCK_SIZE)
// Open a socket to the target address and keep reading from it
logInfo(log"Connecting to ${MDC(LogKeys.HOST, host)}:${MDC(LogKeys.PORT, port)}")
val channel = SocketChannel.open()
@@ -84,6 +86,7 @@ class RawNetworkReceiver(host: String, port: Int, storageLevel: StorageLevel)
readFully(channel, lengthBuffer)
lengthBuffer.flip()
val length = lengthBuffer.getInt()
+ RawNetworkReceiver.checkBlockLength(length, maxBlockSize)
val dataBuffer = ByteBuffer.allocate(length)
readFully(channel, dataBuffer)
dataBuffer.flip()
@@ -105,3 +108,20 @@ class RawNetworkReceiver(host: String, port: Int, storageLevel: StorageLevel)
}
}
}
+
+private[streaming] object RawNetworkReceiver {
+
+ /**
+ * Validates a frame length read from the wire before it can drive an allocation.
+ */
+ def checkBlockLength(length: Int, maxBlockSize: Long): Unit = {
+ if (length < 0) {
+ throw new IOException(s"Invalid negative block length $length received on raw socket.")
+ }
+ if (length > maxBlockSize) {
+ throw new IOException(s"Block length $length received on raw socket exceeds " +
+ s"${StreamingConf.RAW_INPUT_MAX_BLOCK_SIZE.key} ($maxBlockSize bytes); refusing " +
+ "to allocate. Raise the limit if the sender legitimately batches larger blocks.")
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org