Malformed or non-Kafka broker response can cause OOM in protocol.decodeArray

[2212881810]

What version of kafka-go are you using?

Observed with github.com/segmentio/kafka-go v0.4.48.

I also checked v0.4.51, and the relevant decoder path still appears to allocate arrays directly from the decoded length:

• protocol.decodeArray calls makeArray(elemType, int(n))
• makeArray calls reflect.MakeSlice(..., n, n) or unsafe_NewArray(..., n)

What version of Kafka are you testing against?

Apache Kafka 3.9.0.

What happened?

A Kafka broker was misconfigured with advertised.listeners=PLAINTEXT://localhost:9092.

The kafka-go writer was bootstrapped from a remote address, for example:

writer := &kafka.Writer{
    Addr:  kafka.TCP("10.x.x.x:9092"),
    Topic: "test-topic",
}

The initial metadata request succeeded, but the broker metadata advertised localhost:9092.

As a result, kafka-go later tried to connect to localhost:9092 from the client host. In our environment that port was not Kafka; it was another local service. The bytes
returned by that non-Kafka service were then decoded as a Kafka response.

Instead of returning a protocol/connection error, the process crashed with OOM.

Relevant stack:

fatal error: runtime: out of memory

github.com/segmentio/kafka-go/protocol.makeArray
github.com/segmentio/kafka-go/protocol.(*decoder).decodeArray
github.com/segmentio/kafka-go/protocol.(*Decoder).Decode
github.com/segmentio/kafka-go/protocol.ReadResponse
github.com/segmentio/kafka-go.(*Transport).RoundTrip
github.com/segmentio/kafka-go.(*connGroup).connect
github.com/segmentio/kafka-go.(*Writer).WriteMessages

### Why this looks like a kafka-go robustness bug

The broker configuration was definitely wrong, but a malformed/non-Kafka response should not be able to make the client allocate an unbounded slice and crash the process.

In protocol/decode.go, decodeArray reads an int32 length from the response and allocates before validating whether that many elements can fit in the remaining frame:

func (d *decoder) decodeArray(v value, elemType reflect.Type, decodeElem decodeFunc) {
    if n := d.readInt32(); n < 0 {
        v.setArray(array{})
    } else {
        a := makeArray(elemType, int(n))
        for i := 0; i < int(n) && d.remain > 0; i++ {
            decodeElem(d, a.index(i))
        }
        v.setArray(a)
    }
}

makeArray then allocates directly from n.

### Expected behavior

If the response is malformed, truncated, or not a Kafka protocol response, kafka-go should return an error such as invalid response / malformed response / unexpected EOF.

It should not allocate memory proportional to an untrusted length field before validating it.

### Possible fix direction

Before allocating arrays, validate the decoded array length against one or more bounds, for example:

- remaining response frame bytes
- minimum element encoded size where known
- configured maximum response size
- a sane protocol-level maximum

If the length is impossible or unreasonable for the remaining frame, return a decode error instead of allocating.

### Additional context

There is an old PR that protected invalid header counts from causing huge allocations:

https://github.com/segmentio/kafka-go/pull/319

This issue seems similar in spirit, but affects general protocol array decoding rather than only message headers.