RPC (dynamic) plugin input: `Ack` ignores `BatchId`/`Error` and always closes the component instead of routing to the acked batch

public/plugin/go/rpcn/rpcn.go's input.Ack doesn't route to the batch it's acking — it discards the request and unconditionally closes the whole component instead.

func (i *input) ReadBatch(ctx context.Context, _ *runtimepb.BatchInputReadRequest) (*runtimepb.BatchInputReadResponse, error) {
	...
	myID := i.batchIDGenerator.Add(1)
	i.acks.Store(myID, ack)
	...
	return &runtimepb.BatchInputReadResponse{BatchId: myID, Batch: proto}, nil
}

func (i *input) Ack(ctx context.Context, _ *runtimepb.BatchInputAckRequest) (*runtimepb.BatchInputAckResponse, error) {
	if i.component == nil {
		return &runtimepb.BatchInputAckResponse{Error: runtimepb.ErrorToProto(service.ErrNotConnected)}, nil
	}
	err := i.component.Close(ctx)
	return &runtimepb.BatchInputAckResponse{Error: runtimepb.ErrorToProto(err)}, nil
}

ReadBatch stores each batch's AckFunc in i.acks, keyed by a generated myID — clearly meant to be looked up later by ID. But Ack's request parameter is discarded entirely (_ *runtimepb.BatchInputAckRequest): it never reads BatchId or Error, and i.acks is never read anywhere in the file. Every Ack call, for any batch, ack or nack, just calls Close on the whole component instead.

The host side already sends both fields on every ack — internal/rpcplugin/input.go:

resp, err := i.client.Ack(ctx, &runtimepb.BatchInputAckRequest{
	BatchId: id,
	Error:   runtimepb.ErrorToProto(err),
})

Consequence: for an input component that relies on the ack/nack outcome (e.g. completing or abandoning a message-queue receive depending on whether the downstream write succeeded), the first ack the engine sends tears down the whole component instead of settling the specific batch it's for. Close() doesn't necessarily leave the component in a state ReadBatch reports as ErrNotConnected afterward (the only error the host reconnects on), so the input can end up permanently stuck rather than cleanly reconnecting.

For context: the official Go RPC plugin template (internal/rpcplugin/golangtemplate/input/main.go) works around this by setting autoRetryNacks: true with a no-op AckFunc. If that's the intended pattern for RPC input plugins today, happy to close this out — otherwise I'm glad to send a PR that routes Ack to the stored closure by BatchId.

Version: v4.107.2 (benthos_version=4.107.2 reported at runtime); confirmed the same code is present on main.