dekaf: Add connection+buffer size limit, fix network errors being categorized as auth errors

[jshearer]

Tunables

We were seeing issues caused by large numbers of incoming connection requests causing OOMs, so I'm hoping that adding and subsequently tuning these knobs will let us prevent that in the future. It's possible that we'll want to introduce the ability to adjust the chunk cap on a per-materialization basis, but for the moment tuning it globally is good enough.

• MAX_CONNECTIONS limits the total number of connections per Dekaf pod. The limit is enforced by immediately disconnecting new connections if we're over the limit.
• READ_BUFFER_CHUNK_LIMIT allows for configuring the buffer size we ask read_json_lines() to maintain. Previously this was hardcoded to 30, which means we can have up to 4mb buffered per session. If we have 300 sessions, that's ~1.2gb in just ReadJsonLine buffers.

Error codes

We're getting lots of complaints about unexpected authentication errors. This is because we're currently categorizing all errors encountered during sasl_authenticate as SaslAuthenticationFailed, whereas most of the real causes are transient network related errors. We could be even better about breaking down the error variants, but this should be enough to stop the bleeding.

Fixes #2064

---

This change is 

[jshearer]

Doing some testing of this. librdkafka says some things including the following when a connection is rejected:

%6|1744746186.306|FAIL|rdkafka#consumer-1| [thrd:sasl_plaintext://127.0.0.1:22262/1]: sasl_plaintext://127.0.0.1:22262/1: Disconnected while requesting ApiVersion: might be caused by incorrect security.protocol configuration (connecting to a SSL listener?) or broker version is < 0.10 (see api.version.request)

%6|1744749880.880|FAIL|rdkafka#consumer-1| [thrd:GroupCoordinator]: GroupCoordinator: 127.0.0.1:22262: Disconnected (after 0ms in state APIVERSION_QUERY)

%6|1744749903.332|FAIL|rdkafka#consumer-1| [thrd:sasl_plaintext://localhost:22262/bootstrap]: sasl_plaintext://localhost:22262/bootstrap: Disconnected (after 22453ms in state UP)

It works fine, in that the consumer retries and once it's able to establish the connection, consumes data. But I'm somewhat concerned that we're just setting ourselves up for more confused users in the future, asking "Why are you using super old Kafka brokers?" or some such.

The "real" solution is probably to accept all connections, and "poison" the ones that are accepted after the limit is reached. A poisoned session will take its earliest opportunity to return an error code like THROTTLING_QUOTA_EXCEEDED and then close itself.

Edit:

I tried:

• Poisoning a session, causing it to return THROTTLING_QUOTA_EXCEEDED or INTERNAL_SERVER_ERROR from either ApiVersions or either of the SASL handshake steps. This added some complexity, and at no point did the consumer log anything less scary than previously
• Simply refusing to accept() TCP connections if over the limit. This was actually worse as it caused the consumer to exit without retrying

In the end, I think the cleanest solution is what I had originally: disconnecting connections immediately.

[psFried]

LGTM

[psFried]

I think this is OK as it is for now, but I'll suggest that you take a look at tower because I think it might help simplify a lot of this. There's built in support for concurrency limit and load shedding, in addition to generally having a nice design for composing whatever other middleware we might want.

[jshearer]

Ooo, that looks great! I'll take a look, thanks