[CELEBORN-2391] Eliminate intermediate byte[] allocation in Encoders.Strings via Netty ByteBufUtil

[yew1eb]

What changes were proposed in this pull request?

Replace s.getBytes(StandardCharsets.UTF_8) in Encoders.Strings#encodedLength and #encode with Netty's ByteBufUtil.utf8Bytes / ByteBufUtil.writeUtf8, eliminating the intermediate byte[] allocated for every encoded string.

In #encode, write a zero length slot first, then ByteBufUtil.writeUtf8, then backfill the actual byte length via setInt. This avoids traversing the string twice — once for utf8Bytes(s) to size the length field, again inside writeUtf8 — as suggested by @xhumanoid in the review. #encodedLength still uses utf8Bytes(s), since message framing needs the exact total length up front to size the header buffer.

Also includes:

• EncodersSuiteJ: round-trip tests (ASCII / Chinese / emoji) plus byte-for-byte wire-compatibility tests against the previous getBytes(UTF_8) encoding, covering UTF-8 edge cases (NUL/control chars, 2/3-byte boundaries, BMP max, and lossy unpaired surrogates)
• EncodersJmhBenchmark: the JMH benchmark used for the measurements below, following the existing SlotsAllocatorJmhBenchmark pattern (test-scope jmh-core / jmh-generator-annprocess in common/pom.xml and project/CelebornBuild.scala)
• Exact-fill assertion adjustments in MessageEncoder, SslMessageEncoder and Message#toByteBuffer (see below)

Note on the assertion adjustments. ByteBufUtil.writeUtf8 reserves utf8MaxBytes (up to 3 bytes per char) of capacity before writing, which may grow the target buffer beyond the exact encoded length. The three encoding paths allocate the header buffer with exact capacity and asserted buf.writableBytes() == 0 after encoding; that assert fired under -ea in tests even though exactly encodedLength() bytes were written, because capacity had grown. The asserts now check buf.writerIndex() == expectedLength, which expresses the real invariant — bytes written by encode() equal encodedLength() — without depending on buffer capacity. The wire bytes are identical in both implementations.

Why are the changes needed?

Encoders.Strings is on the hot path of every transport message (PushData, PushMergedData, PushDataHandShake, RegionStart, RegionFinish, ...). Each getBytes(UTF_8) call allocates a throwaway byte[], which adds up to significant GC pressure under load.

Measured with EncodersJmhBenchmark (-prof gc, pure-ASCII strings — the only realistic case for shuffleKey/partitionUniqueId; encode paths reuse a pre-sized destination buffer to isolate the encoding cost). Allocation, before (original getBytes) vs after (this PR):

Benchmark (stringBytes x arrayLen)	Metric	Before	After
encodeString (16)	gc.alloc.rate.norm	200 B/op	≈0 B/op
encodeString (256)	gc.alloc.rate.norm	1160 B/op	≈0 B/op
encodeStringArray (128 x 16)	gc.alloc.rate.norm	25600 B/op	≈0 B/op
encodeStringArray (128 x 256)	gc.alloc.rate.norm	148480 B/op	≈0.01 B/op

Throughput, before (original getBytes) vs after (this PR, including the encode backfill), -f 0 -wi 5 -i 10:

Benchmark (stringBytes x arrayLen)	Before	After	Delta
encodeString (16)	37.0 ns/op	25.6 ns/op	-31%
encodeString (256)	465.7 ns/op	243.8 ns/op	-48%
encodeStringArray (8 x 16)	317.4 ns/op	216.2 ns/op	-32%
encodeStringArray (128 x 256)	59961 ns/op	31142 ns/op	-48%

The per-call byte[] allocation is eliminated in all cases, and throughput improves for both short and long strings. The backfill in #encode removes the extra utf8Bytes(s) traversal from the encode path, which is what previously made the long-string (256) case regress relative to the original getBytes + writeBytes path; with the backfill it now improves across the board.

Note: for ASCII-dominant keys, writeUtf8 reserves 3x capacity and may grow the pooled header buffer once per encode. This reallocation is recycled within Netty's arena and produces no GC garbage, unlike the eliminated per-call byte[].

Does this PR resolve a correctness bug?

No.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

• New unit tests EncodersSuiteJ (round-trip + wire compatibility, incl. UTF-8 edge cases) — re-run after the backfill change, all cases pass
• Existing network suites: RpcIntegrationSuiteJ, SSLRpcIntegrationSuiteJ, AutoSSLRpcIntegrationSuite, SslMessageEncoderSuiteJ, RegistrationSuiteJ
• EncodersJmhBenchmark reproduced with:

build/mvn -pl common -am test-compile
build/mvn -pl common exec:java \
  -Dexec.mainClass=org.apache.celeborn.common.network.protocol.EncodersJmhBenchmark \
  -Dexec.classpathScope=test \
  -Dexec.args="-f 0 -wi 1 -i 1"

[Kalvin2077]

writeUtf8 reserves utf8MaxBytes(s.length())—up to 3 bytes per UTF-16 code unit. Therefore, a buffer initially sized from encodedLength()may still grow.

[yew1eb]

Confirmed -- writeUtf8 reserves utf8MaxBytes (3 bytes/char), so an exactly-sized buffer grows (36 -> 128 for a 30-char ASCII string in a quick check). Wire bytes are unaffected; the updated asserts check writerIndex(0, which is the invariant encode() actually guarantees. Pre-sizing header buffers could avoid the grow-and-copy, but I'd keep this PR focused and revisit if benchmarks show it matters.

[Kalvin2077]

Thanks for confirming.Keeping the pre-sizing optimization out of this PR sounds reasonable; we can revisit it separately if profiling shows the extra allocation and copy are significant.

[xhumanoid]

in agoda we use this version

      int lenIdx = buf.writerIndex();              - remember current index
      buf.writeInt(0);                             - write 0
      int written = ByteBufUtil.writeUtf8(buf, s); - write all data
      buf.setInt(lenIdx, written);                 - update specific offset with actual size

in this case you don't need to traverse string byte[] twice to calculate how much data you will write

original code

      buf.writeInt(ByteBufUtil.utf8Bytes(s));     - read string array and calculate required amount of bytes for each character (utf-8 have variable size for each)
      ByteBufUtil.writeUtf8(buf, s);              - write data

this double size calculation can be reason of increased time for encodeString (256)

[yew1eb]

Thanks for the suggestions!
Done.

[Kalvin2077]

Try \uD800中.

[yew1eb]

unpaired-surrogate strings can't occur on Celeborn's wire: the fields encoded here are shuffleKey, partitionUniqueId, and errorString. So the Netty truncation is unreachable in practice -- and even if it did occur, utf8Bytes/writeUtf8 agree with each other, so framing stays intact.

[Kalvin2077]

I understand your point. If these inputs are truly out of scope, some of these may not be necessary.

[Kalvin2077]

@yew1eb I can’t resolve comment threads right now due to a network issue. If you’ve verified the change, feel free to resolve it on your side.

[xhumanoid]

also can be update to remove unnecessary allocation

      int length = buf.readInt();
      // in java 8 it implemented as
      //         this.value = StringCoding.decode(charset, bytes, offset, length);
      // so we still have new decoded allocation inside StringCoding
      // can cache decode in ThreadLocal, but it additional complexity
      // in fresh java it reuses decoder instance and skips decoder allocation
      // just wait new version, current implementation is good enough for now
      String s = buf.toString(buf.readerIndex(), length, StandardCharsets.UTF_8);
      buf.skipBytes(length);
      return s;

[yew1eb]

tracked as a separate issue CELEBORN-2393 to keep this PR focused on the encode path.