From adaed91df65d1d8d6459ba8114d4afe6bc65e738 Mon Sep 17 00:00:00 2001 From: Hrishabh Gupta Date: Tue, 30 Jun 2026 14:12:55 +0530 Subject: [PATCH 1/7] docs: propose default forwarding to the assigned upstream broker Revises the Router plugin API (proposal 070) so a request on a broker-bound connection forwards to the broker that endpoint represents by default, and a router declares only the API keys it must intercept. Replaces staticRoutes(). Addresses the staticRoutes() gaps in kroxylicious#4177: the lost forward-to-assigned-broker default, the VC-level vs per-connection scope mismatch, and the unpopulatable key->route map. SPI: intercepts(apiKey, ctx) gate (default = bootstrap connections only), shouldDecodeRequest(apiKey, ctx) (default false), and onRequest(apiKey, RequestFrame, ctx). Response decode is lazy via ResponseFrame.body(); no shouldDecodeResponse and no onResponse callback. Route binding is established at connection-creation time, which keeps bound-connection forwarding unambiguous and permits multiple routes to one cluster. Relates-to: kroxylicious/kroxylicious#4177 Assisted-by: Claude Opus 4.8 Signed-off-by: Hrishabh Gupta --- .../118-default-forward-to-assigned-broker.md | 315 ++++++++++++++++++ 1 file changed, 315 insertions(+) create mode 100644 proposals/118-default-forward-to-assigned-broker.md 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 00000000..3fe53622 --- /dev/null +++ b/proposals/118-default-forward-to-assigned-broker.md @@ -0,0 +1,315 @@ +# 000 - 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**, and how much of each message to decode. + +### 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(); + } + + /** + * Whether onRequest() needs the decoded request body. Default false: the request is handed to + * onRequest as an opaque frame unless the router asks for it. The header is always available. + * There is no response counterpart — see "Decode depth" below. + */ + default boolean shouldDecodeRequest(ApiKeys apiKey, RouterContext ctx) { return false; } + + /** Called only when intercepts() is true. The router chooses the destination(s). */ + CompletionStage onRequest(ApiKeys apiKey, RequestFrame frame, RouterContext ctx); + + default void close() {} +} +``` + +Two axes, deliberately separate: + +* **Destination** is decided by `intercepts()` (and, when intercepted, by `onRequest()` calling + `ctx.sendRequest(node, ...)`). When not intercepted, the destination is the bound broker. +* **Decode depth** is "pay only for what you touch," defaulting to nothing: + * **Request** — declarative via `shouldDecodeRequest` (default `false`). `onRequest` receives a + `RequestFrame` whose body is decoded iff the router declared it. It must be declarative because + the request is decoded during the upstream filter pass *before* `onRequest` runs, so it composes + with the filter `DecodePredicate` up front. + * **Response** — **lazy**, no declaration. `ctx.sendRequest(...)` returns a `ResponseFrame` whose + `body()` decodes *on access*. A relay never calls `body()` and pays zero response-decode cost; a + virtualising router calls `body()` and it decodes then. This **widens** 070's `sendRequest` + (`CompletionStage`, always decoded) to `CompletionStage`, and + `respondWith` accepts either a decoded `ApiMessage` or an opaque `ResponseFrame` for relay. + +There is deliberately no `shouldDecodeResponse` method and no separate `onResponse` callback: the +lazy frame expresses "decode the response only if the router reads it" without either. + +### 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 and does not imply decoding. + +### 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)) { + // request body decoded iff shouldDecodeRequest; responses from sendRequest decode lazily on access + dispatchToRouter(apiKey, frame, routerCtx); + } 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) && router.shouldDecodeRequest(k, routerCtx)); + } + + @Override + public boolean shouldDecodeResponse(ApiKeys k, short v) { + return filters.shouldDecodeResponse(k, v); // router responses decode lazily, not here + } +} +``` + +The router does not contribute to the response predicate: responses it fetches via `sendRequest` +decode lazily on access (see _Decode depth_), and responses on the pass-through path are governed by +the route's filter chain. On a bound (data-plane) connection a router that does not override +`intercepts` contributes **zero** decode interest, so those connections get the leanest possible +decode profile. + +### 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 boolean shouldDecodeRequest(ApiKeys k, RouterContext ctx) { return false; } // client-id is in the header + public CompletionStage onRequest(ApiKeys k, RequestFrame f, 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), f.header(), f.rawBody()) + .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(ApiKeys k, RequestFrame f, RouterContext ctx) { + String route = routeForSubject(ctx.authenticatedSubject()); // set by SaslTerminator earlier + return ctx.sendRequest(ctx.anyNode(route), f.header(), f.rawBody()) + .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 boolean shouldDecodeRequest(ApiKeys k, RouterContext ctx) { return needsBody(k); } // false for pinned relays + public CompletionStage onRequest(ApiKeys k, RequestFrame f, RouterContext ctx) { + Map sub = decomposer(k).decompose(f.body(), table, f.apiVersion()); // 1 entry (pinned) or N + var calls = sub.entrySet().stream() + .map(e -> ctx.sendRequest(ctx.anyNode(e.getKey()), f.header(), e.getValue()) + .thenApply(r -> Map.entry(e.getKey(), r))) + .toList(); + return allOf(calls).thenApply(responses -> + ctx.respondWith(decomposer(k).recompose(responses, f.body(), f.apiVersion())).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` and + `shouldDecodeRequest` (no `shouldDecodeResponse`); `onRequest` takes a `RequestFrame`; the dispatch + path gains the bound/unbound gate; the `DecodePredicate` is built per connection. `sendRequest` is + widened to return a `CompletionStage` whose body decodes lazily on access (from + `CompletionStage`), with a matching `respondWith(ResponseFrame)` overload for opaque + relay. +- **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. The `routingMode` metric (`static`/`dynamic`) is preserved: +`intercepts == false` is reported as `static`, `intercepts == true` as `dynamic`. + +## 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); + `HEADER_ONLY` needs a header-decoded-but-body-opaque codec tier that doesn't exist (`DecodePredicate` + is binary per message). +- **`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 + +- **`intercepts` vs `requiresDynamicRouting` naming.** `intercepts` better describes the gate's real + job (the router may intercept to reroute, to fan out, _or_ to rewrite a response without + rerouting). `requiresDynamicRouting` aligns with the `routingMode` metric vocabulary. The metric + name and method name need not match. +- **`RequestFrame`/`ResponseFrame` shape** — `RequestFrame.body()` is present only when + `shouldDecodeRequest` returned true; `ResponseFrame.body()` decodes lazily on first access (raw + bytes retained until then, with the usual `ByteBuf` lifecycle care). Confirm both expose the header + unconditionally. Requires widening 070's `sendRequest` to `CompletionStage` and a + `respondWith(ResponseFrame)` overload. From e0ad1015fe8417d6f920c72d229da152a629600a Mon Sep 17 00:00:00 2001 From: Hrishabh Gupta Date: Thu, 2 Jul 2026 08:52:59 +0530 Subject: [PATCH 2/7] docs: defer decode-depth mechanics to a follow-on proposal Slims the proposal to the intercepts() gate per review feedback: drops shouldDecodeRequest and the RequestFrame/ResponseFrame lazy-decode abstractions, reverting onRequest and sendRequest to their 070 signatures. The DecodePredicate wiring stays, consulting intercepts() so that non-intercepted keys on bound connections are never decoded. Decode-depth optimisation of the interception path is recorded as follow-on work. Also updates the heading to the assigned proposal number (118). Relates-to: kroxylicious/design#118 Assisted-by: Claude Fable 5 Signed-off-by: Hrishabh Gupta --- .../118-default-forward-to-assigned-broker.md | 105 ++++++++---------- 1 file changed, 44 insertions(+), 61 deletions(-) diff --git a/proposals/118-default-forward-to-assigned-broker.md b/proposals/118-default-forward-to-assigned-broker.md index 3fe53622..eded9c5e 100644 --- a/proposals/118-default-forward-to-assigned-broker.md +++ b/proposals/118-default-forward-to-assigned-broker.md @@ -1,4 +1,4 @@ -# 000 - Default forwarding to the assigned upstream broker +# 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, @@ -45,7 +45,7 @@ Three problems with `staticRoutes()`, drawn from ## 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**, and how much of each message to decode. +default; a router declares only what it must **intercept**. ### Route binding is established when the connection is created @@ -87,37 +87,25 @@ interface Router { return ctx.virtualNode().isEmpty(); } - /** - * Whether onRequest() needs the decoded request body. Default false: the request is handed to - * onRequest as an opaque frame unless the router asks for it. The header is always available. - * There is no response counterpart — see "Decode depth" below. - */ - default boolean shouldDecodeRequest(ApiKeys apiKey, RouterContext ctx) { return false; } - /** Called only when intercepts() is true. The router chooses the destination(s). */ - CompletionStage onRequest(ApiKeys apiKey, RequestFrame frame, RouterContext ctx); + CompletionStage onRequest(short apiVersion, ApiKeys apiKey, + RequestHeaderData header, ApiMessage request, RouterContext ctx); default void close() {} } ``` -Two axes, deliberately separate: +`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. -* **Destination** is decided by `intercepts()` (and, when intercepted, by `onRequest()` calling - `ctx.sendRequest(node, ...)`). When not intercepted, the destination is the bound broker. -* **Decode depth** is "pay only for what you touch," defaulting to nothing: - * **Request** — declarative via `shouldDecodeRequest` (default `false`). `onRequest` receives a - `RequestFrame` whose body is decoded iff the router declared it. It must be declarative because - the request is decoded during the upstream filter pass *before* `onRequest` runs, so it composes - with the filter `DecodePredicate` up front. - * **Response** — **lazy**, no declaration. `ctx.sendRequest(...)` returns a `ResponseFrame` whose - `body()` decodes *on access*. A relay never calls `body()` and pays zero response-decode cost; a - virtualising router calls `body()` and it decodes then. This **widens** 070's `sendRequest` - (`CompletionStage`, always decoded) to `CompletionStage`, and - `respondWith` accepts either a decoded `ApiMessage` or an opaque `ResponseFrame` for relay. - -There is deliberately no `shouldDecodeResponse` method and no separate `onResponse` callback: the -lazy frame expresses "decode the response only if the router reads it" without either. +**Deferred: 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, and a relay shouldn't pay to decode +a response it never reads. Those optimisations (a `shouldDecodeRequest` declaration, lazy +request/response frames) are real but separable: they optimise what happens *after* interception and +have no bearing on the default destination. They are left to a follow-on proposal so this one stays +focused on the gate. 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`) @@ -131,7 +119,7 @@ no extra methods: | **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 and does not imply decoding. +entry; it needs no separate code path. ### What the runtime does @@ -146,8 +134,7 @@ class RouteDispatchHandler extends ChannelDuplexHandler { ApiKeys apiKey = frame.apiKey(); { if (router.intercepts(apiKey, routerCtx)) { - // request body decoded iff shouldDecodeRequest; responses from sendRequest decode lazily on access - dispatchToRouter(apiKey, frame, 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()); @@ -171,22 +158,19 @@ class RouteAwareDelegatingDecodePredicate implements DecodePredicate { @Override public boolean shouldDecodeRequest(ApiKeys k, short v) { - return filters.shouldDecodeRequest(k, v) - || (router.intercepts(k, routerCtx) && router.shouldDecodeRequest(k, routerCtx)); + return filters.shouldDecodeRequest(k, v) || router.intercepts(k, routerCtx); } @Override public boolean shouldDecodeResponse(ApiKeys k, short v) { - return filters.shouldDecodeResponse(k, v); // router responses decode lazily, not here + return filters.shouldDecodeResponse(k, v) || router.intercepts(k, routerCtx); } } ``` -The router does not contribute to the response predicate: responses it fetches via `sendRequest` -decode lazily on access (see _Decode depth_), and responses on the pass-through path are governed by -the route's filter chain. On a bound (data-plane) connection a router that does not override -`intercepts` contributes **zero** decode interest, so those connections get the leanest possible -decode profile. +On a bound (data-plane) connection a router that does not override `intercepts` contributes **zero** +decode interest, so those connections get the leanest possible decode profile. Reducing what an +*intercepted* request/response must decode is the follow-on decode-depth work noted above. ### Examples @@ -208,12 +192,12 @@ 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 boolean shouldDecodeRequest(ApiKeys k, RouterContext ctx) { return false; } // client-id is in the header - public CompletionStage onRequest(ApiKeys k, RequestFrame f, RouterContext ctx) { + 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), f.header(), f.rawBody()) + return ctx.sendRequest(ctx.anyNode(route), header, request) .thenApply(r -> ctx.respondWith(r).build()); } } @@ -227,9 +211,10 @@ class ClientIdRouter implements Router { ```java class SubjectRouter implements Router { public boolean intercepts(ApiKeys k, RouterContext ctx) { return ctx.virtualNode().isEmpty(); } - public CompletionStage onRequest(ApiKeys k, RequestFrame f, RouterContext ctx) { + 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), f.header(), f.rawBody()) + return ctx.sendRequest(ctx.anyNode(route), header, request) .thenApply(r -> ctx.respondWith(r).build()); } } @@ -243,15 +228,15 @@ class TopicRouter implements Router { public boolean intercepts(ApiKeys k, RouterContext ctx) { return ctx.virtualNode().isEmpty() || spansClusters(k) || pinned(k); // METADATA…, FIND_COORDINATOR… } - public boolean shouldDecodeRequest(ApiKeys k, RouterContext ctx) { return needsBody(k); } // false for pinned relays - public CompletionStage onRequest(ApiKeys k, RequestFrame f, RouterContext ctx) { - Map sub = decomposer(k).decompose(f.body(), table, f.apiVersion()); // 1 entry (pinned) or N + 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()), f.header(), e.getValue()) + .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, f.body(), f.apiVersion())).build()); + ctx.respondWith(decomposer(k).recompose(responses, request, v)).build()); } } // Data-plane traffic on a broker connection passes through; only cluster-spanning / coordinator @@ -273,12 +258,9 @@ Subject-routing connection walk-through (`cluster0`: `c0b0`,`c0b1`; `cluster1`: ## Affected/not affected projects -- **kroxylicious (runtime):** affected. `staticRoutes()` removed; `Router` gains `intercepts` and - `shouldDecodeRequest` (no `shouldDecodeResponse`); `onRequest` takes a `RequestFrame`; the dispatch - path gains the bound/unbound gate; the `DecodePredicate` is built per connection. `sendRequest` is - widened to return a `CompletionStage` whose body decodes lazily on access (from - `CompletionStage`), with a matching `respondWith(ResponseFrame)` overload for opaque - relay. +- **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. @@ -295,9 +277,8 @@ this revises an in-flight API. The `routingMode` metric (`static`/`dynamic`) is - **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); - `HEADER_ONLY` needs a header-decoded-but-body-opaque codec tier that doesn't exist (`DecodePredicate` - is binary per message). + 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. @@ -308,8 +289,10 @@ this revises an in-flight API. The `routingMode` metric (`static`/`dynamic`) is job (the router may intercept to reroute, to fan out, _or_ to rewrite a response without rerouting). `requiresDynamicRouting` aligns with the `routingMode` metric vocabulary. The metric name and method name need not match. -- **`RequestFrame`/`ResponseFrame` shape** — `RequestFrame.body()` is present only when - `shouldDecodeRequest` returned true; `ResponseFrame.body()` decodes lazily on first access (raw - bytes retained until then, with the usual `ByteBuf` lifecycle care). Confirm both expose the header - unconditionally. Requires widening 070's `sendRequest` to `CompletionStage` and a - `respondWith(ResponseFrame)` overload. + +## Follow-on work (out of scope here) + +- **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, and lazy + request/response frames so a relay doesn't pay to decode a response it never reads. Separable from + the gate; deserves its own proposal. From 8656bb137916a252b8ebce4792a7a66a63669492 Mon Sep 17 00:00:00 2001 From: Hrishabh Gupta Date: Mon, 6 Jul 2026 09:24:25 +0530 Subject: [PATCH 3/7] docs: address review comments on metrics and gate naming Marks the routingMode metric categorisation as revisitable rather than fixed by this proposal. Reworks the gate-naming open question: drops the response-rewrite justification (that belongs to the route filter chain) and the claim that metric and method names need not match, adopting a single vocabulary as the goal while leaving the name choice to review. Relates-to: kroxylicious/design#118 Assisted-by: Claude Fable 5 Signed-off-by: Hrishabh Gupta --- proposals/118-default-forward-to-assigned-broker.md | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/proposals/118-default-forward-to-assigned-broker.md b/proposals/118-default-forward-to-assigned-broker.md index eded9c5e..28fd4b6d 100644 --- a/proposals/118-default-forward-to-assigned-broker.md +++ b/proposals/118-default-forward-to-assigned-broker.md @@ -269,8 +269,9 @@ Subject-routing connection walk-through (`cluster0`: `c0b0`,`c0b1`; `cluster1`: ## Compatibility Proposal [070](./070-routing-api.md) is not yet released, so `staticRoutes()` has no users to break; -this revises an in-flight API. The `routingMode` metric (`static`/`dynamic`) is preserved: -`intercepts == false` is reported as `static`, `intercepts == true` as `dynamic`. +this revises an in-flight API. The gate maps naturally onto the `routingMode` metric +(`intercepts == false` → `static`, `== true` → `dynamic`), but the metric categorisation +is to be revisited once the SPI settles and is not fixed by this proposal. ## Rejected alternatives @@ -285,10 +286,10 @@ this revises an in-flight API. The `routingMode` metric (`static`/`dynamic`) is ## Open questions -- **`intercepts` vs `requiresDynamicRouting` naming.** `intercepts` better describes the gate's real - job (the router may intercept to reroute, to fan out, _or_ to rewrite a response without - rerouting). `requiresDynamicRouting` aligns with the `routingMode` metric vocabulary. The metric - name and method name need not match. +- **Gate naming: `intercepts` vs `requiresDynamicRouting`.** One term should be used everywhere — + SPI, metrics, logs, docs. The gate covers rerouting and fan-out (both routing), which favours + `requiresDynamicRouting` and matches the existing `routingMode` vocabulary; alternatively, keep + `intercepts` and align the metric vocabulary to it when the metrics are revisited. ## Follow-on work (out of scope here) From a3aa45460a4afd8ad407620629b9a4ef2cfc5bf7 Mon Sep 17 00:00:00 2001 From: Hrishabh Gupta Date: Mon, 6 Jul 2026 09:37:05 +0530 Subject: [PATCH 4/7] docs: drop response-side laziness, slim gate-naming open question Response decoding is already driven by virtual node-id translation (kroxylicious#4257), Fetch included, leaving too narrow a case to justify widening sendRequest; the follow-on is now request-side decode depth only. The naming open question no longer ties itself to the unsettled metric vocabulary. Relates-to: kroxylicious/design#118 Assisted-by: Claude Fable 5 Signed-off-by: Hrishabh Gupta --- .../118-default-forward-to-assigned-broker.md | 27 +++++++++---------- 1 file changed, 12 insertions(+), 15 deletions(-) diff --git a/proposals/118-default-forward-to-assigned-broker.md b/proposals/118-default-forward-to-assigned-broker.md index 28fd4b6d..0ff388ed 100644 --- a/proposals/118-default-forward-to-assigned-broker.md +++ b/proposals/118-default-forward-to-assigned-broker.md @@ -99,13 +99,11 @@ interface Router { router, and responses fetched via `sendRequest` return decoded (`CompletionStage`), as 070 specifies. This proposal changes only the gate. -**Deferred: 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, and a relay shouldn't pay to decode -a response it never reads. Those optimisations (a `shouldDecodeRequest` declaration, lazy -request/response frames) are real but separable: they optimise what happens *after* interception and -have no bearing on the default destination. They are left to a follow-on proposal so this one stays -focused on the gate. The gate already delivers the big win by itself — non-intercepted keys are -never decoded at all (see the `DecodePredicate` wiring below). +**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`) @@ -286,14 +284,13 @@ is to be revisited once the SPI settles and is not fixed by this proposal. ## Open questions -- **Gate naming: `intercepts` vs `requiresDynamicRouting`.** One term should be used everywhere — - SPI, metrics, logs, docs. The gate covers rerouting and fan-out (both routing), which favours - `requiresDynamicRouting` and matches the existing `routingMode` vocabulary; alternatively, keep - `intercepts` and align the metric vocabulary to it when the metrics are revisited. +- **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) -- **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, and lazy - request/response frames so a relay doesn't pay to decode a response it never reads. Separable from - the gate; deserves its own proposal. +- **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`.) From 40a78f552d775c419743d1340688d7a4d33c0273 Mon Sep 17 00:00:00 2001 From: Hrishabh Gupta Date: Mon, 6 Jul 2026 09:41:17 +0530 Subject: [PATCH 5/7] docs: drop metric mapping from compatibility section Metrics are unreleased alongside 070, so the mapping is not a compatibility concern; categorisation follows once the SPI naming settles. Relates-to: kroxylicious/design#118 Assisted-by: Claude Fable 5 Signed-off-by: Hrishabh Gupta --- proposals/118-default-forward-to-assigned-broker.md | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/proposals/118-default-forward-to-assigned-broker.md b/proposals/118-default-forward-to-assigned-broker.md index 0ff388ed..1a008751 100644 --- a/proposals/118-default-forward-to-assigned-broker.md +++ b/proposals/118-default-forward-to-assigned-broker.md @@ -267,9 +267,7 @@ Subject-routing connection walk-through (`cluster0`: `c0b0`,`c0b1`; `cluster1`: ## Compatibility Proposal [070](./070-routing-api.md) is not yet released, so `staticRoutes()` has no users to break; -this revises an in-flight API. The gate maps naturally onto the `routingMode` metric -(`intercepts == false` → `static`, `== true` → `dynamic`), but the metric categorisation -is to be revisited once the SPI settles and is not fixed by this proposal. +this revises an in-flight API. ## Rejected alternatives From ecc85266f28351b45fdc0c66803bd28bf2b89f44 Mon Sep 17 00:00:00 2001 From: Hrishabh Gupta Date: Mon, 6 Jul 2026 09:46:02 +0530 Subject: [PATCH 6/7] docs: scope the decode predicate sketch to the request side The gate only changes request dispatch; response decoding stays governed by filters and node-id translation (kroxylicious#4257), so the sketch no longer models shouldDecodeResponse at all. Relates-to: kroxylicious/design#118 Assisted-by: Claude Fable 5 Signed-off-by: Hrishabh Gupta --- proposals/118-default-forward-to-assigned-broker.md | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/proposals/118-default-forward-to-assigned-broker.md b/proposals/118-default-forward-to-assigned-broker.md index 1a008751..bfd75972 100644 --- a/proposals/118-default-forward-to-assigned-broker.md +++ b/proposals/118-default-forward-to-assigned-broker.md @@ -159,16 +159,14 @@ class RouteAwareDelegatingDecodePredicate implements DecodePredicate { return filters.shouldDecodeRequest(k, v) || router.intercepts(k, routerCtx); } - @Override - public boolean shouldDecodeResponse(ApiKeys k, short v) { - return filters.shouldDecodeResponse(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** -decode interest, so those connections get the leanest possible decode profile. Reducing what an -*intercepted* request/response must decode is the follow-on decode-depth work noted above. +request-decode interest. Reducing what an *intercepted* request must decode is the follow-on +decode-depth work noted above. ### Examples From b211b6591f27b7a52c123026bbd902e399f1690e Mon Sep 17 00:00:00 2001 From: Hrishabh Gupta Date: Wed, 5 Aug 2026 11:00:15 +0530 Subject: [PATCH 7/7] docs: rename the gate to shouldIntercept and adopt virtual-node language Names the gate shouldIntercept(apiKey, apiVersion, ctx), taking the apiVersion ask and the should* form the reviewers converged on. Keeping it distinct from DecodePredicate.shouldDecodeRequest matters because the follow-on decode-depth work will let an intercepting router skip the body decode, so the two questions diverge; a paragraph by the predicate wiring says so. Records the YAGNI outcome of the RouteAction discussion in the SPI section and rejected alternatives. Replaces the control-plane/data-plane framing with a plain statement that the bootstrap/node endpoint split is a structural property the SPI depends on. Rewrites the ClientIdRouter comment around where routing actually happens, and switches to virtual-node language throughout, noting that routers compose as a DAG so each sees its own node-id mapping, and that the 1:1 mapping to a physical broker is not promised. Adds the progressive-decode feasibility caveat to the follow-on section, links the proof-of-concept, and notes the pre-existing out-of-band-request defect it surfaced. Drops the settled gate-naming open question. Relates-to: kroxylicious/design#118 Assisted-by: Claude Opus 5 Signed-off-by: Hrishabh Gupta --- .../118-default-forward-to-assigned-broker.md | 164 +++++++++++------- 1 file changed, 100 insertions(+), 64 deletions(-) diff --git a/proposals/118-default-forward-to-assigned-broker.md b/proposals/118-default-forward-to-assigned-broker.md index bfd75972..e4d98eac 100644 --- a/proposals/118-default-forward-to-assigned-broker.md +++ b/proposals/118-default-forward-to-assigned-broker.md @@ -1,7 +1,7 @@ -# 118 - Default forwarding to the assigned upstream broker +# 118 - Default forwarding to the assigned virtual node 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, +node-bound connection is, by default, forwarded to the virtual node that endpoint already represents, and a router declares only the API keys it needs to **intercept**. Replaces `staticRoutes()`. Addresses the gaps raised in @@ -29,14 +29,14 @@ implementation `onRequest()` is not yet wired, so an API key absent from `static 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. +1. **The useful default was lost.** Before routing, every upstream broker was represented by its own + virtual endpoint and traffic to that endpoint _always_ flowed to the 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. + instances are per-connection. A route name cannot express "forward to the implied target virtual + node" — that target (`virtualNode()`) is derived initially from the 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 @@ -44,34 +44,43 @@ Three problems with `staticRoutes()`, drawn from ## Proposal -Invert the model: the runtime forwards a bound connection's traffic to its assigned broker by +Invert the model: the runtime forwards a bound connection's traffic to its assigned virtual node 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 +A connection reaches the proxy on a specific endpoint — a bootstrap endpoint, or a node endpoint for +one `(route, virtual node)`. 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. +per-connection pipeline. Three consequences: + +* A bound (node-endpoint) connection's forward target is unambiguous: it already knows its + `(route, virtual node)` and filter chain, so `shouldIntercept == false` simply forwards there — no + per-request route recovery. +* "Derived initially from the connection" is deliberate: routers compose as a DAG, and each router in + the DAG sees its own virtual node id mapping. Router A may see a request destined for + virtualNodeId 2 while Router B sees the same request destined for virtualNodeId 0. A router reads + `ctx.virtualNode()` to learn which virtual node the request is associated with *at its own position + in the DAG*; it is not a single global identity. Nor is a virtual node a physical broker: the 1:1 + mapping the runtime happens to use today is not something the SPI promises, and future topologies + need not preserve it. * 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 +### Bootstrap and node endpoints are a structural distinction -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: +Kafka makes no protocol distinction between a bootstrap broker and a data broker, but the proxy must, +because the endpoint a connection arrived on is the only signal for whether a default destination +exists. That makes the distinction a structural property the SPI depends on, not an implementation +convenience, and it should be visible in the API: -* **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. +* **bootstrap endpoint** → connection is _unbound_: `RouterContext.virtualNode()` is empty, there is + no default destination, so the router must participate. +* **node endpoint** → connection is _bound_ to one `(route, virtual node)`: the default destination + is that virtual node, so the router need not be consulted at all. ### The SPI @@ -80,14 +89,14 @@ 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). + * frame to the connection's assigned virtual node (virtualNode()) without calling onRequest(). + * Default: intercept only when there is no assigned virtual node (bootstrap connections). */ - default boolean intercepts(ApiKeys apiKey, RouterContext ctx) { + default boolean shouldIntercept(ApiKeys apiKey, short apiVersion, RouterContext ctx) { return ctx.virtualNode().isEmpty(); } - /** Called only when intercepts() is true. The router chooses the destination(s). */ + /** Called only when shouldIntercept() is true. The router chooses the destination(s). */ CompletionStage onRequest(short apiVersion, ApiKeys apiKey, RequestHeaderData header, ApiMessage request, RouterContext ctx); @@ -95,6 +104,14 @@ interface Router { } ``` +`apiVersion` is part of the gate so a router can intercept only the version ranges it cares about, +and so the gate composes directly with `DecodePredicate.shouldDecodeRequest(apiKey, apiVersion)`. + +The gate is a `boolean` rather than a richer action type. A `sealed RouteAction` (forward-to-implicit-node, +intercept, forward-to-named-node, multicast) was considered and rejected as YAGNI: the boolean is +sufficient for every router shape below, and it keeps symmetry with the filter API's `shouldHandle*` +predicates. Richer actions can be introduced when a use case demands them. + `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. @@ -112,9 +129,9 @@ 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` | +| **pass-through** | `shouldIntercept == false` | _(not called; forward to bound virtual node)_ | — | +| **pinned** | `shouldIntercept == true` | send the whole request to one fixed route | `GetTelemetrySubscriptions` | +| **fan-out** | `shouldIntercept == 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. @@ -131,7 +148,7 @@ class RouteDispatchHandler extends ChannelDuplexHandler { RequestFrame frame = (RequestFrame) msg; ApiKeys apiKey = frame.apiKey(); { - if (router.intercepts(apiKey, routerCtx)) { + if (router.shouldIntercept(apiKey, frame.apiVersion(), routerCtx)) { dispatchToRouter(apiKey, frame, routerCtx); // decoded per 070; router picks destination via sendRequest } else { routerCtx.sendRequest(routerCtx.virtualNode(), frame.header(), frame.rawBody()) @@ -156,7 +173,7 @@ class RouteAwareDelegatingDecodePredicate implements DecodePredicate { @Override public boolean shouldDecodeRequest(ApiKeys k, short v) { - return filters.shouldDecodeRequest(k, v) || router.intercepts(k, routerCtx); + return filters.shouldDecodeRequest(k, v) || router.shouldIntercept(k, v, routerCtx); } // shouldDecodeResponse is unchanged: response decoding is governed by filters and the runtime's @@ -164,7 +181,13 @@ class RouteAwareDelegatingDecodePredicate implements DecodePredicate { } ``` -On a bound (data-plane) connection a router that does not override `intercepts` contributes **zero** +The two predicates are deliberately distinct names for distinct questions: `shouldDecodeRequest` asks +whether the runtime must materialise the body, `shouldIntercept` asks whether the router must be +invoked. Interception implies decoding today, which is why the gate is OR'd in here, but the +follow-on decode-depth work will let an intercepting router opt out of the body decode — at which +point conflating the two names would be actively wrong. + +On a bound connection a router that does not override `shouldIntercept` contributes **zero** request-decode interest. Reducing what an *intercepted* request must decode is the follow-on decode-depth work noted above. @@ -172,11 +195,11 @@ decode-depth work noted above. 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. +**Static** — front a single cluster. Never intercepts; bootstrap and node traffic both forward. ```java class StaticRouter implements Router { - public boolean intercepts(ApiKeys k, RouterContext ctx) { return false; } // single route, never + public boolean shouldIntercept(ApiKeys k, short v, RouterContext ctx) { return false; } // single route, never public CompletionStage onRequest(...) { throw new AssertionError("unused"); } } ``` @@ -185,9 +208,11 @@ class StaticRouter implements Router { ```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(); } + // Routing happens on the bootstrap connection (virtualNode() is empty): the client is routed to a + // cluster and gets back a Metadata response describing that cluster's nodes. It then connects to + // the nodes it needs, and those connections are already bound — virtualNode() is present, so they + // need no routing at all and the client-id is never re-examined. + public boolean shouldIntercept(ApiKeys k, short v, 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 @@ -199,14 +224,14 @@ class ClientIdRouter implements Router { } // 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. +// pass-through on bound connections. ``` **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 boolean shouldIntercept(ApiKeys k, short v, 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 @@ -221,7 +246,7 @@ class SubjectRouter implements Router { ```java class TopicRouter implements Router { - public boolean intercepts(ApiKeys k, RouterContext ctx) { + public boolean shouldIntercept(ApiKeys k, short v, RouterContext ctx) { return ctx.virtualNode().isEmpty() || spansClusters(k) || pinned(k); // METADATA…, FIND_COORDINATOR… } public CompletionStage onRequest(short v, ApiKeys k, RequestHeaderData header, @@ -235,8 +260,8 @@ class TopicRouter implements Router { 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). +// Traffic on a bound 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`): @@ -246,22 +271,28 @@ Subject-routing connection walk-through (`cluster0`: `c0b0`,`c0b1`; `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 ----------->| intercepts (bootstrap): expose ONLY cluster1 nodes |<-- 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 + |--- Produce (vID 1) ---->| bound: shouldIntercept==false -> forward to virtualNode()==c1b0 + |--- Fetch (vID 3) ----->| bound: shouldIntercept==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. +- **kroxylicious (runtime):** affected. `staticRoutes()` removed; `Router` gains `shouldIntercept`; + the dispatch path gains the bound/unbound gate; the `DecodePredicate` is built per connection and + consults `shouldIntercept`. `onRequest` and `sendRequest` keep their 070 signatures. +- **Router plugin authors:** affected. The no-op default is "forward to the assigned virtual node," a + safe and usually-correct starting point. - **Filter API and existing filter configurations:** not affected. - **Operator / CRDs:** not affected — SPI/runtime only. +A proof-of-concept of this design exists at +[kroxylicious#4510](https://github.com/kroxylicious/kroxylicious/pull/4510). It surfaced a +pre-existing defect on `main` — out-of-band requests issued by virtual-cluster filters or router +filters misbehave in combination with dynamic routing. That defect is independent of this proposal +and should be tracked and fixed separately, but it will be encountered by whoever implements this. + ## Compatibility Proposal [070](./070-routing-api.md) is not yet released, so `staticRoutes()` has no users to break; @@ -270,23 +301,28 @@ 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). + represented virtual node (route ≠ node). +- **A `sealed RouteAction` return type instead of a `boolean`** — richer actions + (forward-to-implicit-node, forward-to-named-node, multicast) would leave room for extension, but no + current router shape needs them, and the boolean keeps the gate symmetric with the filter API's + `shouldHandle*` predicates. Revisit when a use case arrives. - **`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. + cluster-granular, so it can't name the specific represented virtual node that `sendRequest` (070) + targets by node; loses leader affinity. ## 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`.) +- **Request decode depth on the interception path** — a 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, and is worth registering as an issue now even though it should wait for a concrete use + case. Two caveats for whoever picks it up: + - Progressive decoding is not obviously feasible against today's `RequestData`/`ResponseData` + APIs. It would need the root object to retain a reference to the `ByteBuf` and decode + sub-structs lazily on first access, which is a change to generated Kafka message classes rather + than to the routing SPI. + - 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`.