From adaed91df65d1d8d6459ba8114d4afe6bc65e738 Mon Sep 17 00:00:00 2001 From: Hrishabh Gupta Date: Tue, 30 Jun 2026 14:12:55 +0530 Subject: [PATCH 1/6] 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 0000000..3fe5362 --- /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/6] 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 3fe5362..eded9c5 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/6] 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 eded9c5..28fd4b6 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/6] 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 28fd4b6..0ff388e 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/6] 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 0ff388e..1a00875 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/6] 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 1a00875..bfd7597 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