perf: build each write request once and replay it as is

[merlimat]

For #412 (the "Each send rebuilds the whole WriteRequest" item).

Problem

WriteBatch hands ManagedWriteStream a Supplier<WriteRequest> instead of a built request. The stream calls it on the first send, and again for every in-flight write each time it reconnects. When a shard's leader moves, every reconnect attempt rebuilds every in-flight batch's full request (one message per operation, plus a buffer wrapper per value) while the client is recovering. Each send also allocates the bound method reference.

Example

// WriteBatch.send()
writeStream.send(this::toProto)

// ManagedWriteStream.initWithRecovery(), on every reconnect
for (InflightWrite inflightWrite : inflightWrites) {
    subStreamObserver.send(inflightWrite.requestSupplier.get()); // toProto() again
}

With 8 batches in flight and 3 reconnect attempts before the new leader answers, toProto() runs 32 times for 8 requests.

The supplier came from #310. LightProto 0.6.6 wrote bytes fields with _b.writeBytes(value), which advanced the value buffer's reader index, so a replayed put sent an empty value. LightProto 0.7.1 (streamnative/lightproto#6) writes them with value.getBytes(value.readerIndex(), ...), which leaves the buffer as it was, and the client is on 0.9.0.

Change

send() takes the built request, the in-flight entry keeps it, and replay sends that same object:

// A replay sends the same request again: serializing it doesn't consume its buffers
record InflightWrite(
        WriteRequest request, CompletableFuture<WriteResponse> future, long timestampNanos) {}

public CompletableFuture<WriteResponse> send(WriteRequest request) { ... }
...
subStreamObserver.send(inflightWrite.request);

WriteBatch calls writeStream.send(toProto()). The generated gRPC marshaller serializes inside stream(), and send() and initWithRecovery() both run under the stream lock, so a request is never serialized twice at the same time.

The request now stays reachable until its response instead of becoming garbage right after the send. It wraps the operations' key strings and value arrays without copying them, so the extra memory is only the message objects.

Testing

• ManagedWriteStreamTest now passes requests directly. replaysInflightWritesAfterRetryableError sends each request object to a stale leader and then replays the same object to the new leader, and checks the values that arrive. With LightProto pinned to 0.7.0 it fails (["???????", "???????"] instead of ["value-1", "value-2"]); on 0.9.0 it passes.
• ./gradlew :client:test: all 402 tests pass locally, including the Docker integration tests (ClientReconnectIT, OxiaClientIT, ...).