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

# 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](https://issues.apache.org/jira/browse/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

```hocon
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](https://issues.apache.org/jira/browse/FLINK-33505): Switch away from Netty 3-based Pekko Classic Remoting — Flink still uses Classic Transport and would benefit from this optimization.
- [FLINK-10718](https://issues.apache.org/jira/browse/FLINK-10718): Use IO executor in RpcService for message serialization — Flink's RPC layer serialization bottleneck, complementary to this transport-level optimization.
- [FLINK-34105](https://issues.apache.org/jira/browse/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`)

Start by reading PekkoProtocolTransport.scala, Endpoint.scala, TcpSupport.scala, and PekkoPduProtobufCodec to trace the current inbound and outbound paths. Review reference.conf for the proposed codec-threads setting and determine how both protobuf layers, protocol state handling, and the AssociationHandle.write contract fit the pipeline change. Done means the executor is configurable, ordering is preserved, and existing behavior remains available when disabled.

Written by the indexing model from the issue text.