Classic Transport: Add Parallel Serialization/Deserialization via Netty Pipeline Executor

[He-Pin]

Classic Transport: Add Parallel Serialization/Deserialization via Netty Pipeline Executor

Summary

Classic Transport serializes and deserializes messages synchronously on Actor dispatcher threads. Each association has one EndpointWriter (outbound) and one EndpointReader (inbound), all sharing a fork-join dispatcher pool (default: up to 16 threads, pekko.remote.default-remote-dispatcher). Within a single association, serialization/deserialization is strictly serial — messages are processed one at a time.

Artery solves this with configurable parallel lanes (inbound-lanes / outbound-lanes) using Akka Streams, but Classic Transport has no equivalent. This issue proposes adding parallel serialization/deserialization to Classic Transport by leveraging Netty's EventExecutorGroup for codec handlers in the pipeline — a minimal-invasive approach that avoids restructuring the Actor-based message pipeline.

Why invest in deprecated Classic Transport?

Classic Transport is deprecated (@deprecated("Classic remoting is deprecated, use Artery", "Akka 2.6.0")). However, major projects like Apache Flink still rely on it (see FLINK-33505), and migration to Artery is a large undertaking. This optimization provides incremental performance improvements for users who cannot switch to Artery immediately, and introduces concepts (codec executor, per-recipient partitioning) that ease the eventual migration.

Motivation

Current Architecture (Classic Transport)

Classic Transport is built on Netty 4 (io.netty.*) with a per-association Actor model.

Wire Format: Two-Level Protobuf Encoding

The wire format has two nested Protobuf layers:

1. Outer envelope (PekkoProtocolMessage): Wraps payload, heartbeat, associate, disassociate. Handled by decodePdu() / constructPayload() in PekkoPduProtobufCodec.
2. Inner message (AckAndEnvelopeContainer): Wraps the serialized message, recipient ActorRef, sender ActorRef, and ack. Handled by decodeMessage() / constructMessage() in PekkoPduProtobufCodec.

Both levels are parsed/encoded on Actor dispatcher threads.

Inbound Path (Deserialization)

Netty IO EventLoop thread
  → FrameDecoder (LengthFieldBasedFrameDecoder)              ← byte-level framing
  → TcpServerHandler.onMessage() [TcpSupport.scala:58]       ← copies bytes, notifies listener
    → listener.notify(InboundPayload(ByteString))
      │
      ├─ ProtocolStateActor (always in path, PekkoProtocolTransport.scala:559-584)
      │    → decodePdu(p)                                    ← outer Protobuf: PekkoProtocolMessage
      │    → match: Payload(payload) →
      │      → listener.notify(InboundPayload(payload))       ← forward inner payload
      │        │
      │        └─ EndpointReader.receive() [Endpoint.scala:1163]   ← Actor dispatcher thread
      │             → tryDecodeMessageAndAck() [Endpoint.scala:1276]
      │               → codec.decodeMessage()                ← inner Protobuf: AckAndEnvelopeContainer
      │                 → resolveActorRefWithLocalAddress()   ← ActorRef resolution
      │             → msgDispatch.dispatch() [Endpoint.scala:1178]
      │               → MessageSerializer.deserialize() [Endpoint.scala:83]  ← heaviest operation
      │               → targetActor ! payload
      │
      └─ (during handshake: ProtocolStateActor handles Associate/Disassociate directly)

Key insight: ProtocolStateActor is always in the message path, even post-handshake (in the Open state). It decodes the outer PekkoProtocolMessage envelope on its dispatcher thread, then forwards the inner Payload to EndpointReader, which decodes the inner AckAndEnvelopeContainer and deserializes the actual message — also on a dispatcher thread.

Outbound Path (Serialization)

EndpointWriter Actor thread [Endpoint.scala:651]              ← Actor dispatcher thread
  → serializeMessage() [Endpoint.scala:1065]
    → MessageSerializer.serialize()                          ← heaviest operation
  → codec.constructMessage() [Endpoint.scala:922]            ← inner Protobuf: AckAndEnvelopeContainer
  → handle.write(pdu)
    → PekkoProtocolHandle.write() [PekkoProtocolTransport.scala:240]
      → codec.constructPayload()                             ← outer Protobuf: PekkoProtocolMessage
      → wrappedHandle.write()
        → TcpAssociationHandle.write() [TcpSupport.scala:120]
          → channel.writeAndFlush(Unpooled.wrappedBuffer(...))
            → Netty IO EventLoop thread
              → FrameEncoder (LengthFieldPrepender)          ← add frame header
              → write to network

The Bottleneck

With N remote connections, Classic Transport has N EndpointWriters and N EndpointReaders, all sharing a fork-join dispatcher pool (default: parallelism-factor = 0.5, parallelism-max = 16, per reference.conf:280-289). The bottleneck manifests as:

1. Within a single association, serialization/deserialization is strictly serial — messages are processed one at a time per Actor.
2. CPU-intensive serialization blocks the Actor thread — large Protobuf messages or Java serialization stall the message queue, increasing latency for all messages on that association.
3. Dispatcher pool contention — under high load, the shared dispatcher pool becomes saturated, causing backpressure across all associations.

Artery Comparison

Artery solves this with a fundamentally different architecture based on Akka Streams:

Dimension	Classic Transport	Artery
Inbound deserialization	Single EndpointReader Actor per association	FixedSizePartitionHub partitions messages across N inbound lanes (default 4), each with its own Deserializer GraphStage running in parallel
Outbound serialization	Single EndpointWriter Actor per association	N outbound lanes (default 1) per association, each with its own Encoder GraphStage, merged via MergeHub
Message ordering	Naturally ordered (single thread)	Preserved via consistent hashing on recipient.path.uid
Serialization API	Serializer.toBinary() -> Array[Byte]	ByteBufferSerializer.toBinary(ByteBuffer) (zero-copy)
Memory management	Per-message byte array allocation	EnvelopeBufferPool with buffer recycling
Compression	None	ActorRef and manifest compression tables
Parallelism granularity	Per-association (one thread)	Per-message (partitioned by recipient)
Deprecation status	Deprecated since Akka 2.6.0	Recommended transport
Configuration	No parallelism config	inbound-lanes = 4, outbound-lanes = 1 (reference.conf:926, 935)

Key Artery code references:

• Inbound lane partitioning: ArteryTransport.scala:471-484
• Lane setup (TCP): ArteryTcpTransport.scala:405-467
• Deserializer per lane: Codecs.scala:645-717
• Outbound lane selection: Association.scala:476-505

Proposed Solution: Netty Pipeline Executor for Codec Handlers

Approach

Netty 4 supports assigning an independent EventExecutorGroup to specific pipeline handlers via pipeline.addLast(EventExecutorGroup, name, handler). Handlers bound to a custom executor run on that executor's thread pool instead of the IO EventLoop, enabling:

1. Cross-connection parallelism: Multiple connections' encode/decode operations run concurrently on the executor pool. With N connections and K codec threads, up to K connections can encode/decode concurrently.
2. Per-connection ordering preserved: Netty guarantees events for the same channel are delivered in order to the assigned executor thread.

Important limitation: Netty binds each channel to a specific thread in the EventExecutorGroup for its lifetime. This means:

• Parallelism is across connections, not within a single connection.
• A "hot" connection always uses the same executor thread, potentially creating load imbalance.
• This is fundamentally different from Artery's lane-based approach, which partitions by recipient and provides within-connection parallelism.

Proposed Pipeline Change

Before:
  SslHandler -> FlushConsolidationHandler -> FrameDecoder -> FrameEncoder -> ServerHandler

After:
  SslHandler -> FlushConsolidationHandler -> FrameDecoder
    -> [codecExecutor] PekkoMessageDecoder
    -> [codecExecutor] PekkoMessageEncoder
    -> FrameEncoder -> ServerHandler

• PekkoMessageDecoder: On the codec executor thread, decodes the outer PekkoProtocolMessage envelope (what decodePdu() does today in ProtocolStateActor) and, for Payload messages, also decodes the inner AckAndEnvelopeContainer (what decodeMessage() does today in EndpointReader). Outputs a structured object that downstream handlers can consume directly.
• PekkoMessageEncoder: On the codec executor thread, encodes the outer PekkoProtocolMessage envelope (what constructPayload() does today in PekkoProtocolHandle) before frame encoding.

Configuration

pekko.remote.classic.netty.tcp {
  # Number of threads for codec (serialization/deserialization) operations.
  # 0 = disabled (codec runs on Actor dispatcher threads, current behavior).
  # Recommended: min(CPU cores / 2, number of remote connections).
  codec-threads = 0
}

Note: The config path is netty.tcp to match the existing NettyTransportSettings hierarchy (reference.conf:513).

Key Design Considerations

1. ProtocolStateActor is always in the message path: Even post-handshake (in the Open state), ProtocolStateActor receives every InboundPayload, calls decodePdu() to parse the outer PekkoProtocolMessage, and forwards Payload to EndpointReader (PekkoProtocolTransport.scala:559-584). The proposed PekkoMessageDecoder must handle both layers of Protobuf encoding, and downstream consumers (ProtocolStateActor and EndpointReader) must be adapted to accept pre-decoded objects.

2. Two-level Protobuf decoding: The decoder must handle:

   • Outer: PekkoProtocolMessage (payload, heartbeat, associate, disassociate)
   • Inner: AckAndEnvelopeContainer (message, recipient, sender, ack) — only for Payload types
     Control messages (heartbeat, associate, disassociate) should pass through with minimal overhead.

3. Write path restructuring: EndpointWriter.writeSend() currently calls serializeMessage() + codec.constructMessage() + handle.write() synchronously. PekkoProtocolHandle.write() calls codec.constructPayload(). To offload the outer envelope encoding to the Netty executor, the AssociationHandle.write() contract may need to change — this has binary compatibility implications (MiMa checks).

4. True message serialization (MessageSerializer.serialize/deserialize): This is the heaviest operation but is tightly coupled to the Actor system's Serialization extension and Serialization.currentTransportInformation thread-local context. Moving it to the Netty pipeline requires propagating the serialization context, which is more invasive.

5. Netty per-channel thread affinity: Each channel is bound to one thread in the EventExecutorGroup for its lifetime. Load balancing depends on the number of channels, not message volume. Users with few connections but high message volume per connection will see minimal benefit from this approach alone.

6. Interaction with use-dispatcher-for-io: The existing use-dispatcher-for-io config (reference.conf:588) allows using a custom dispatcher for Netty IO. The codec-threads config should be independent — it creates a separate executor pool for codec operations, regardless of the IO dispatcher setting.

7. SSL interaction: If SSL is enabled, SslHandler is added first in the pipeline. The PekkoMessageDecoder receives decrypted bytes, which is correct. The EventExecutorGroup assignment does not interfere with SSL handshake processing (SSL handshake is handled by SslHandler on the EventLoop).

8. Backpressure: If the codec executor queue grows (e.g., burst of large messages), consider ChannelOption.WRITE_BUFFER_WATER_MARK or pausing reads when the executor queue is full.

Implementation Scope

Phase 1 (This Issue): Protobuf Codec Offloading

Offload the two-level Protobuf encoding/decoding (PekkoProtocolMessage + AckAndEnvelopeContainer) to the Netty codec executor. This removes Protobuf parsing from Actor dispatcher threads and enables cross-connection parallelism.

• Add PekkoMessageDecoder / PekkoMessageEncoder handlers to NettyTransport.newPipeline() with a configurable EventExecutorGroup
• Adapt ProtocolStateActor to accept pre-decoded PekkoPdu objects instead of raw ByteString
• Adapt EndpointReader to accept pre-decoded Message objects instead of raw ByteString
• Add configuration: pekko.remote.classic.netty.tcp.codec-threads
• Add metrics: codec time per message (via RemoteMetrics)
• Evaluate binary compatibility impact (MiMa)

Phase 2 (Future): Message Serialization Offloading

Offload MessageSerializer.serialize/deserialize to a dedicated thread pool. This is the heaviest operation and provides the largest throughput gain.

• Propagate Serialization.currentTransportInformation to the codec executor
• Ensure message ordering per recipient (similar to Artery's inboundLanePartitioner — hash on recipient.path.uid)
• Adapt DefaultMessageDispatcher to accept pre-deserialized messages from the pipeline

Expected Benefits

1. Cross-connection parallelism: With N connections and K codec threads, up to K connections can encode/decode concurrently.
2. Actor dispatcher relief: Codec work moves off the shared default-remote-dispatcher pool, reducing contention for Actor message processing.
3. Incremental migration path: Users who cannot switch to Artery immediately get a performance improvement.
4. Foundation for Phase 2: The codec executor infrastructure enables the later addition of MessageSerializer offloading.

Limitations:

• Single-connection throughput is not improved (Netty per-channel thread affinity).
• For users with few connections but high per-connection volume, Phase 2 (message serialization offloading with recipient-based partitioning) is needed.

Related Issues

• FLINK-33505: Switch away from Netty 3-based Pekko Classic Remoting — Flink still uses Classic Transport and would benefit from this optimization.
• FLINK-10718: Use IO executor in RpcService for message serialization — Flink's RPC layer serialization bottleneck, complementary to this transport-level optimization.
• FLINK-34105: Akka timeout in TPC-DS benchmarks — Real-world evidence that serialization overhead impacts cluster stability at scale (10TB TPC-DS).

References

• Classic Transport inbound path: Endpoint.scala:1163 (EndpointReader.receive), Endpoint.scala:1276 (tryDecodeMessageAndAck), Endpoint.scala:1178 (msgDispatch.dispatch), Endpoint.scala:83 (lazy val payload = MessageSerializer.deserialize)
• Classic Transport outbound path: Endpoint.scala:651 (EndpointWriter), Endpoint.scala:1065 (serializeMessage), Endpoint.scala:922 (codec.constructMessage)
• ProtocolStateActor (always in path): PekkoProtocolTransport.scala:559-584 (Open state, decodePdu on every message)
• PekkoProtocolHandle write: PekkoProtocolTransport.scala:240 (codec.constructPayload)
• Classic Transport codec (two-level Protobuf): PekkoPduCodec.scala:83-126 (decodePdu/encodePdu/decodeMessage/constructMessage)
• Netty pipeline setup: NettyTransport.scala:390-406 (newPipeline)
• Netty pipeline initializer: NettyTransport.scala:440-453 (where codec executor would be added)
• Dispatcher pool config: reference.conf:280-289 (pekko.remote.default-remote-dispatcher)
• Artery inbound lane partitioning: ArteryTransport.scala:471-484
• Artery lane setup (TCP): ArteryTcpTransport.scala:405-467
• Artery Deserializer per lane: Codecs.scala:645-717
• Artery outbound lane selection: Association.scala:476-505
• Artery lane config: reference.conf:926 (inbound-lanes = 4), reference.conf:935 (outbound-lanes = 1)

[pjfanning]

I don't see why we need to enhance a deprecated transport option and Flink should be encouraged to move onto Artery. Frankly, I don't think they are too interested in this. They didn't even want to use Pekko. They initially planned to write their own replacement for what they used in akka-remote.

[He-Pin]

I believe the new Artery transport primarily introduces parallel encoding/decoding and offers greater stability. Since Flink is currently using Pekko, I think continuing to maintain the "classic" transport—while improving its stability—would benefit both projects. We still need to determine the best way to implement the parallel encoding/decoding feature. So, I’d like to open this issue for now; if someone actually needs it, it won't be too late to develop it then. @ferenc-csaky FYI

[He-Pin]

I think switching to Artery is basically out of the question. Our company has a huge number of online nodes, and practically no one wants to take on a task like that—it’s a thankless job, and if something goes wrong, the consequences fall squarely on you. Someone actually opened an issue a while back suggesting we switch to gRPC, but in the end, nobody did it; gRPC wouldn't be an improvement over what we have now anyway. Besides, even if a new transport protocol comes out, it’s not guaranteed that anyone would dare to use it right away. It requires a massive investment of effort for testing—I still remember how migrating to Netty 4 cost me two months' worth of weekends.

[He-Pin]

Our company has a pretty impressive setup: we use the Node.js client for Pekko Remoting, which allows our cluster to orchestrate services running within Node.js.

[pjfanning]

@He-Pin I can be convinced if your company uses this and will test it. Using Flink to justify any changes is something that I'll have reservations about - Flink have not engaged with us on issues like this and and I'm concerned that neither they nor their users would be well-positioned to properly test changes like this. It is better to have Pekko users test this.

[He-Pin]

I’ve consulted on other systems as well, and generally, there is little appetite for upgrading them to the new transport layer. Our clusters are typically deployed across multiple data centers. Occasionally, we conduct network disconnection drills, which cause the clusters to experience "split-brain" scenarios; we used to handle this by manually restarting them. We eventually resolved the issue by managing cluster state within a database (acting as a centralized configuration store). However, Artery doesn't suffer from these problems, so all our new systems are now built using Artery. For this issue, our system's message density isn't particularly high, whereas Flink deals with very high message density. So, if the Flink community is interested, they might want to get involved; otherwise, if we were to do it ourselves, it would simply be a matter of performance improvement.