RPC

Spark uses Netty for all in-cluster RPC: driver-executor coordination, BlockManager master/slave, the standalone Master/Worker protocol, and the external shuffle service. Spark Connect adds a separate gRPC layer on top, but the engine's internal RPC is its own abstraction.

RpcEnv

RpcEnv (core/.../rpc/RpcEnv.scala) is the abstraction every RPC user holds. It owns:

• RpcEndpoint - registered receivers (each with a name like "BlockManagerMaster").
• RpcEndpointRef - remote handles to endpoints; can send (one-way) or ask (request-reply).
• RpcAddress - host + port.

The default and only implementation is NettyRpcEnv (core/.../rpc/netty/NettyRpcEnv.scala). It wraps common/network-common's TransportClient and TransportServer.

Wiring

graph LR
    SC[SparkContext] --> RE[RpcEnv]
    RE --> NRE[NettyRpcEnv]
    NRE --> TS[TransportServer]
    NRE --> TC[TransportClient]
    TS --> NM[Netty NIO]
    TC --> NM

    BMM[BlockManagerMasterEndpoint] -. registered .-> RE
    HR[HeartbeatReceiver] -. registered .-> RE
    OCC[OutputCommitCoordinator] -. registered .-> RE
    CGSB[CoarseGrainedSchedulerBackend] -. registered .-> RE

Each RpcEndpoint has a single dispatcher thread. Messages received over the wire are deserialized, looked up by endpoint name, and dispatched to the endpoint's receive/receiveAndReply method.

Endpoints in core

Endpoint	What it does
CoarseGrainedSchedulerBackend.DriverEndpoint	Driver-side scheduler endpoint. Talks to executors.
CoarseGrainedExecutorBackend	Executor-side; registers, runs tasks, sends status.
BlockManagerMasterEndpoint	Driver-side block-manager registry.
BlockManagerStorageEndpoint	Per-executor block-manager endpoint.
HeartbeatReceiver (core/.../HeartbeatReceiver.scala)	Aggregates executor heartbeats; declares dead executors.
OutputCommitCoordinator	Decides which task attempt commits when output writers race.
Master / Worker (standalone)	Standalone deploy daemons.
MapOutputTrackerMasterEndpoint	Serves shuffle output locations to executors.

Wire format

Messages are Java-serialized by default; Kryo is sometimes used for performance-critical paths. Headers and framing come from common/network-common.

The transport layer supports:

• TLS (configured via spark.network.crypto.enabled and spark.ssl.*).
• SASL or the new AuthEngine for in-cluster authentication.
• Block transfer over the same connection (mux'd by stream id).

Block transfer service

BlockTransferService (common/network-common) is conceptually a sibling of RpcEnv. It streams large block bodies efficiently rather than buffering them in memory. Both the internal block manager and the external shuffle service use it.

Configuration

• spark.driver.host / spark.driver.port - driver bind address.
• spark.driver.bindAddress - bind to a different interface than host.
• spark.network.timeout - default RPC ask timeout.
• spark.rpc.numRetries, spark.rpc.retry.wait.
• spark.network.io.preferDirectBufs - controls Netty pool behavior.

Threading model

• One IO selector thread group per NettyRpcEnv.
• One dispatcher thread per registered endpoint (i.e., serial in-order processing).
• A shared callback executor for ask futures.

This means an endpoint that blocks its dispatcher will queue all subsequent messages. Long-running work is moved to a dedicated ThreadUtils.newDaemonCachedThreadPool and results are sent back via RpcCallContext.reply.

Spark Connect vs core RPC

Spark Connect speaks gRPC, not Netty/Spark RPC. Its protocol lives in sql/connect/common/.../protobuf/spark/connect/. The Connect server runs inside the driver and consumes the same SparkSession, but the wire is unrelated to RpcEnv. See modules/connect.md.

Integration points

• Every executor-driver interaction uses RpcEnv. The block transfer for shuffle reads uses the same Netty connection but bypasses the dispatcher for efficiency.
• The DAG scheduler does not use RPC directly; it delegates through CoarseGrainedSchedulerBackend.DriverEndpoint.
• Standalone mode's Master/Worker bootstrap is one of the first users of RPC at startup.

Entry points for modification

• Add a new endpoint: implement RpcEndpoint, register with RpcEnv.setupEndpoint. The driver typically registers in SparkContext/SparkEnv.create.
• Add a wire field to an existing message: extend the case class and bump the RpcEndpoint's receive to handle both old and new shapes for rolling upgrades.
• Add a transport-level feature: edit common/network-common/.../TransportConf.java, TransportClient.java, or the SASL/AuthEngine handshake under common/network-common/.../sasl/.