BulkStreamWriter does not recover after TCP connection reset

## Problem

`BulkStreamWriter` detects a broken TCP connection and exposes a `FlightRuntimeException`, but it does not recreate the underlying Arrow Flight `DoPut` stream. Once the stream becomes terminal, the writer remains unusable even when the endpoint is reachable again. The terminal state is permanent in the current implementation because the completion future cannot be reset and `startPut()` is never called again.

Applications must currently discard the failed writer and create a new one manually. At the protocol level, reconnecting a transport cannot revive an already-terminal bidirectional RPC; a new `DoPut` RPC is required.

The reproduction also observed an Arrow allocator leak error while closing the failed writer. This may be a cleanup interaction in the test harness and needs separate confirmation.

## Reproduction

Environment:

- Client commit: `37fc57cf9e01d3695d9c8e0732461d919e9578a9`
- GreptimeDB endpoint: `localhost:24001`
- GreptimeDB version: unknown; endpoint was provided through a local `k9s` port-forward
- Java: OpenJDK 21.0.10
- OS: Ubuntu 24.04, Linux 6.17.0-22-generic, x86_64
- Arrow Flight: 14.0.2
- gRPC: 1.78.0
- gRPC transport observed in the stack trace: Netty NIO
- Client settings: route refresh disabled, per-message timeout 5 seconds, max in-flight requests 4, zero-copy enabled

The reproduction used a disposable TCP proxy between the client and `localhost:24001`:

1. Use the regular write API to create a test table.
2. Create a client whose endpoint is the local TCP proxy.
3. Create one `BulkStreamWriter` and write one batch successfully.
4. Configure `SO_LINGER=0` and close both active proxy sockets to inject a TCP RST while keeping the proxy listener available.
5. Wait one second and write another batch with the same writer.
6. Create a replacement `BulkStreamWriter` through the same proxy and write another batch.

The relevant client calls were:

```java
assertEquals(1, write(writer, "before-reset", 1.0));
proxy.resetActiveConnections();
Thread.sleep(1000);

// Throws FlightRuntimeException: UNAVAILABLE: io exception.
write(writer, "after-reset", 2.0);

BulkStreamWriter replacement = client.bulkStreamWriter(schema, config);
assertEquals(1, write(replacement, "replacement-writer", 3.0));
```

`write(...)` creates a new `TableBufferRoot`, adds one row, calls `complete()`, and waits on `writer.writeNext().get(10, TimeUnit.SECONDS)`.

Command used for the integration harness:

```shell
mvn -pl ingester-integration-tests -am \
-Dit.test=BulkWriteReconnectIT \
-DargLine="--add-opens=java.base/java.nio=ALL-UNNAMED" \
verify
```

Relevant output:

```text
Caused by: java.net.SocketException: Connection reset

REPRODUCED: existing writer failed after TCP reset;
acceptedConnectionsBeforeReset=1,
acceptedConnectionsNow=1,
errorType=org.apache.arrow.flight.FlightRuntimeException,
error=org.apache.arrow.flight.FlightRuntimeException: UNAVAILABLE: io exception

REPRODUCED: replacement writer succeeded; acceptedConnections=2

Memory was leaked by query. Memory leaked: (16)
Allocator(BufferAllocator(Location{uri=grpc+tcp://localhost:})) ...
```

The harness passed because it asserted the current behavior: the original writer fails and a manually created replacement succeeds. Both successful writes returned an affected-row count of 1. Persistence was not independently queried because the port-forwarded instance's SQL endpoint required credentials unavailable to this harness.

The reset was injected between batches. This reproduction therefore demonstrates failure to recover the writer for later batches; it does not establish the server-side outcome of a batch interrupted while in flight.

## Current implementation

- `BulkWriteClient.bulkStreamWriteTo()` routes once and creates one `BulkWriteManager`: `ingester-protocol/src/main/java/io/greptime/BulkWriteClient.java`
- `BulkWriteService` calls `manager.startPut(...)` only in its constructor and stores the resulting listener permanently: `ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java`
- A transport error reaches `AsyncPutListener.onError()` and completes tracked in-flight futures exceptionally.
- There is no retry, reroute, manager replacement, channel replacement, or `startPut()` re-invocation.
- `BulkFlightClient` can observe a terminal stream error, but it cannot recreate the `DoPut` RPC: `ingester-bulk-protocol/src/main/java/org/apache/arrow/flight/BulkFlightClient.java`

## Expected behavior

After a terminal transport failure, the current batch should fail with the transport error. A later `writeNext()` should create a new `DoPut` stream automatically when the endpoint is reachable again, without replaying the failed batch implicitly.

Handling the batch that was in flight at disconnect requires care: its server-side outcome may be unknown, so automatic replay can introduce duplicates unless an idempotency mechanism exists.

At minimum:

- Mark the writer terminal immediately and consistently.
- Do not allow later writes to hang or degrade into per-message timeout.
- Do not leak limiter permits or Arrow allocator memory.
- Expose enough state/error information for callers to recreate the writer safely.
- Document whether unacknowledged batches can be replayed.

## Suggested tests

- Successful batch followed by TCP RST.
- In-flight futures receive the terminal stream error.
- Writes after terminal failure fail immediately and consistently.
- Recovery creates a new `DoPut` stream; whether the implementation also replaces the manager or channel is an implementation choice.
- Failed-writer close releases all Arrow memory.
- Concurrent write racing with stream termination does not miss error fan-out.
- Limiter permits are released when `putNext()` throws synchronously.

Start with BulkWriteClient.java, BulkWriteService.java, and BulkFlightClient.java to trace manager creation, startPut(), and terminal error handling. Run the stated BulkWriteReconnectIT Maven command, then use the suggested reset, in-flight, concurrency, close, and limiter cases to define the current failure behavior. Done means later writes fail or recover consistently without replaying an unacknowledged batch, leaked permits, or Arrow memory.

Written by the indexing model from the issue text.