diff --git a/proposals/118-default-forward-to-assigned-broker.md b/proposals/118-default-forward-to-assigned-broker.md new file mode 100644 index 0000000..bfd7597 --- /dev/null +++ b/proposals/118-default-forward-to-assigned-broker.md @@ -0,0 +1,292 @@ +# 118 - Default forwarding to the assigned upstream broker + +Revise the `Router` plugin API (proposal [070](./070-routing-api.md)) so that a request on a +broker-bound connection is, by default, forwarded to the broker that endpoint already represents, +and a router declares only the API keys it needs to **intercept**. Replaces `staticRoutes()`. + +Addresses the gaps raised in +[kroxylicious#4177](https://github.com/kroxylicious/kroxylicious/issues/4177). + +## Current situation + +Proposal [070](./070-routing-api.md) introduced the `Router` plugin with `staticRoutes()`: + +```java +interface Router { + CompletionStage onRequest(short apiVersion, ApiKeys apiKey, + RequestHeaderData header, ApiMessage request, RouterContext context); + + default Map staticRoutes() { return Map.of(); } +} +``` + +`staticRoutes()` maps `ApiKeys -> route name`; those keys are forwarded as opaque frames to the +named route, bypassing `onRequest()`. Every other key is dynamically routed. In the current +implementation `onRequest()` is not yet wired, so an API key absent from `staticRoutes()` throws. + +## Motivation + +Three problems with `staticRoutes()`, drawn from +[kroxylicious#4177](https://github.com/kroxylicious/kroxylicious/issues/4177): + +1. **The useful default was lost.** Before routing, every broker was a virtual endpoint and traffic + to that endpoint _always_ flowed to the upstream broker it represents (`b0.proxy -> b0.cluster0`, + `b10000.proxy -> b0.cluster1`). `staticRoutes()` makes a router opt _into_ forwarding by listing + API keys, instead of inheriting that default. + +2. **Scope mismatch.** `staticRoutes()` is a fixed `ApiKeys -> route name` map, but `Router` + instances are per-connection. A route name cannot express "forward to the broker _this connection_ + represents" — that target (`virtualNode()`) is only known per connection. + +3. **No router can populate the map usefully.** For a pass-through router it means enumerating ~80 + keys; for client-id, subject, or topic routers it is empty or near-empty, because nothing can be + pinned to a route purely by API key. The map either lists everything (bad DX) or returns `{}`. + +## Proposal + +Invert the model: the runtime forwards a bound connection's traffic to its assigned broker by +default; a router declares only what it must **intercept**. + +### Route binding is established when the connection is created + +A connection reaches the proxy on a specific endpoint — a bootstrap endpoint, or a broker endpoint +for one `(route, broker)`. The runtime binds the connection to that route and builds its upstream +filter chain at connection-creation time, exactly as the non-router proxy already builds a +per-connection pipeline. Two consequences: + +* A bound (broker-endpoint) connection's forward target is unambiguous: it already knows its + `(route, broker)` and filter chain, so `intercepts == false` simply forwards there — no per-request + route recovery. +* This does **not** require one route per cluster. Two routes fronting the same cluster are two + distinct endpoints; a connection's route is fixed by the endpoint it arrived on, not by decoding + the request. The only thing out of scope is a single endpoint representing *several* routes at once + (070's shared node-id mapping, opt-in via `RouterFactoryContext.allowSharedClusterTargets()`), + which would require per-request decomposition. + +### Control plane vs data plane + +Kafka makes no protocol distinction between a bootstrap broker and a data broker. The router world +must, because the connection's endpoint is the only signal for whether a default broker exists: + +* **bootstrap endpoint** → connection is _unbound_ (`RouterContext.virtualNode()` is empty) → the + **control plane**: discovery, `METADATA`, coordinator lookup, the routing decisions. +* **broker endpoint** → connection is _bound_ to one `(route, broker)` → the **data plane**: + data-plane traffic to a known broker. + +### The SPI + +```java +interface Router { + + /** + * Whether the router must be invoked for this request. When false the runtime forwards the + * frame to the connection's assigned broker (virtualNode()) without calling onRequest(). + * Default: intercept only when there is no assigned broker (bootstrap connections). + */ + default boolean intercepts(ApiKeys apiKey, RouterContext ctx) { + return ctx.virtualNode().isEmpty(); + } + + /** Called only when intercepts() is true. The router chooses the destination(s). */ + CompletionStage onRequest(short apiVersion, ApiKeys apiKey, + RequestHeaderData header, ApiMessage request, RouterContext ctx); + + default void close() {} +} +``` + +`onRequest` keeps 070's signature unchanged: an intercepted request is decoded and handed to the +router, and responses fetched via `sendRequest` return decoded (`CompletionStage`), as +070 specifies. This proposal changes only the gate. + +**Deferred: request decode depth on the interception path.** A router that intercepts but only needs +the header (client-id, subject) shouldn't force a full body decode. That optimisation is real but +separable: it optimises what happens *after* interception and has no bearing on the default +destination, so it is left to a follow-on proposal. The gate already delivers the big win by +itself — non-intercepted keys are never decoded at all (see the `DecodePredicate` wiring below). + +### Three request shapes (one `onRequest`) + +Requests fall into three shapes; the two intercepted ones are both expressed through `onRequest` — +no extra methods: + +| shape | gate | onRequest does | example | +|---|---|---|---| +| **pass-through** | `intercepts == false` | _(not called; forward to bound broker)_ | — | +| **pinned** | `intercepts == true` | send the whole request to one fixed route | `GetTelemetrySubscriptions` | +| **fan-out** | `intercepts == true` | decompose, send to each route, recompose | `Metadata` | + +Pinned is the degenerate case of fan-out where the decomposition yields a single `(route, request)` +entry; it needs no separate code path. + +### What the runtime does + +Per request: + +```java +class RouteDispatchHandler extends ChannelDuplexHandler { + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) { + RequestFrame frame = (RequestFrame) msg; + ApiKeys apiKey = frame.apiKey(); + { + if (router.intercepts(apiKey, routerCtx)) { + dispatchToRouter(apiKey, frame, routerCtx); // decoded per 070; router picks destination via sendRequest + } else { + routerCtx.sendRequest(routerCtx.virtualNode(), frame.header(), frame.rawBody()) + .thenAccept(r -> routerCtx.respondWith(r).build()); + } + } + } +} +``` + +Because bound-ness is per connection and known at connection setup, the runtime builds a +per-connection `DecodePredicate` (070 already permits this — the predicate "can depend on which +back-end cluster they're connected to"): + +```java +// Existing DelegatingDecodePredicate or new RouteAwareDelegatingDecodePredicate +class RouteAwareDelegatingDecodePredicate implements DecodePredicate { + + private final DecodePredicate filters; // DecodePredicate.forFilters(...) + private final Router router; + private final RouterContext routerCtx; + + @Override + public boolean shouldDecodeRequest(ApiKeys k, short v) { + return filters.shouldDecodeRequest(k, v) || router.intercepts(k, routerCtx); + } + + // shouldDecodeResponse is unchanged: response decoding is governed by filters and the runtime's + // node-id translation (kroxylicious#4257), neither of which this proposal touches. +} +``` + +On a bound (data-plane) connection a router that does not override `intercepts` contributes **zero** +request-decode interest. Reducing what an *intercepted* request must decode is the follow-on +decode-depth work noted above. + +### Examples + +The four canonical routers form a ramp, and every one relies on the same default gate. + +**Static** — front a single cluster. Never intercepts; bootstrap and broker traffic both forward. + +```java +class StaticRouter implements Router { + public boolean intercepts(ApiKeys k, RouterContext ctx) { return false; } // single route, never + public CompletionStage onRequest(...) { throw new AssertionError("unused"); } +} +``` + +**ClientId-based** — key is in the header; route chosen on the bootstrap connection. + +```java +class ClientIdRouter implements Router { + // on a bound connection, the client-id is already known and the route is fixed; no need to intercept. + // dev can additional add `ApiVersionsRequestFilter` or decode ApiVersionRequest to validate clientId once in connection lifetime. + public boolean intercepts(ApiKeys k, RouterContext ctx) { return ctx.virtualNode().isEmpty(); } + public CompletionStage onRequest(short v, ApiKeys k, RequestHeaderData header, + ApiMessage request, RouterContext ctx) { + // ctx.clientId() — or some other way: the router is connection-local, so it can capture the + // client-id when it intercepts the ApiVersions request and carry it on the router object. + String route = routeFor(ctx.clientId()); + return ctx.sendRequest(ctx.anyNode(route), header, request) + .thenApply(r -> ctx.respondWith(r).build()); + } +} +// Endpoint/identity consistency (client-id route == bound endpoint route) is a connection-scoped +// check; it belongs on the bound filter chain, NOT in onRequest — keeping it out preserves +// data-plane pass-through. +``` + +**Subject-based** — key comes from context; auth is already terminated on the VC chain. + +```java +class SubjectRouter implements Router { + public boolean intercepts(ApiKeys k, RouterContext ctx) { return ctx.virtualNode().isEmpty(); } + public CompletionStage onRequest(short v, ApiKeys k, RequestHeaderData header, + ApiMessage request, RouterContext ctx) { + String route = routeForSubject(ctx.authenticatedSubject()); // set by SaslTerminator earlier + return ctx.sendRequest(ctx.anyNode(route), header, request) + .thenApply(r -> ctx.respondWith(r).build()); + } +} +// API_VERSIONS + SASL are handled by VC filters before the router, so onRequest never sees anonymous. +``` + +**Topic-based** — key is in the body; fan-out for cluster-spanning APIs, pin for coordinator APIs. + +```java +class TopicRouter implements Router { + public boolean intercepts(ApiKeys k, RouterContext ctx) { + return ctx.virtualNode().isEmpty() || spansClusters(k) || pinned(k); // METADATA…, FIND_COORDINATOR… + } + public CompletionStage onRequest(short v, ApiKeys k, RequestHeaderData header, + ApiMessage request, RouterContext ctx) { + Map sub = decomposer(k).decompose(request, table, v); // 1 entry (pinned) or N + var calls = sub.entrySet().stream() + .map(e -> ctx.sendRequest(ctx.anyNode(e.getKey()), header, e.getValue()) + .thenApply(r -> Map.entry(e.getKey(), r))) + .toList(); + return allOf(calls).thenApply(responses -> + ctx.respondWith(decomposer(k).recompose(responses, request, v)).build()); + } +} +// Data-plane traffic on a broker connection passes through; only cluster-spanning / coordinator +// APIs reach onRequest, where the router fans out with ctx.sendRequest (070 contract). +``` + +Subject-routing connection walk-through (`cluster0`: `c0b0`,`c0b1`; `cluster1`: `c1b0`,`c1b1`): + +``` + Client Proxy / Router cluster0 cluster1 + | | virtualNode() empty (bootstrap) c1b0 c1b1 + | ApiVersions/SASL | handled by VC filters (not the router) + | |==== subject "alice" -> cluster1 ==== + |--- Metadata ----------->| intercepts (bootstrap): expose ONLY cluster1 brokers + |<-- Metadata ------------| (vID c1b0=1, c1b1=3 ; mapping V = id + S*t, S=2) + |--- Produce (vID 1) ---->| bound: intercepts==false -> forward to virtualNode()==c1b0 + |--- Fetch (vID 3) ----->| bound: intercepts==false -> forward to virtualNode()==c1b1 +``` + +## Affected/not affected projects + +- **kroxylicious (runtime):** affected. `staticRoutes()` removed; `Router` gains `intercepts`; the + dispatch path gains the bound/unbound gate; the `DecodePredicate` is built per connection and + consults `intercepts`. `onRequest` and `sendRequest` keep their 070 signatures. +- **Router plugin authors:** affected. The no-op default is "forward to the assigned broker," a safe + and usually-correct starting point. +- **Filter API and existing filter configurations:** not affected. +- **Operator / CRDs:** not affected — SPI/runtime only. + +## Compatibility + +Proposal [070](./070-routing-api.md) is not yet released, so `staticRoutes()` has no users to break; +this revises an in-flight API. + +## Rejected alternatives + +- **Keep `staticRoutes()`** — no router can populate it usefully and a route name can't name the + represented broker (route ≠ node). +- **`RoutingMode` enum `{PASS_THROUGH, HEADER_ONLY, FULL_MESSAGE}`** — conflates destination with + decode depth; `PASS_THROUGH` has no destination on bootstrap (so it needs the bound gate anyway). + Decode depth is deferred to a follow-on proposal in any case. +- **`staticRoute(apiKey, ctx)` / `staticRouteForApiKey(apiKey)` returning a route** — a route is + cluster-granular, so it can't name the specific represented broker that `sendRequest` (070) targets + by node; loses leader affinity. + +## Open questions + +- **Gate naming: `intercepts` vs `requiresDynamicRouting`.** Whichever term wins should be used + consistently across SPI, metrics, logs, and docs. + +## Follow-on work (out of scope here) + +- **Request decode depth on the interception path** — a `shouldDecodeRequest`-style declaration so a + router that routes on the header (client-id, subject) doesn't force a body decode. Separable from + the gate; deserves its own proposal. (Response-side laziness was considered and dropped: virtual + node-id translation already decodes the node-reference-bearing responses, `Fetch` included, leaving + too narrow a case to justify widening `sendRequest`.)