Package io.kroxylicious.proxy.topology

Interface TopologyService

public interface TopologyService
Opt-in topology cache for routers that need leader, coordinator, broker, or topic ID information.

Routers that need topology information obtain a TopologyService from RouterFactoryContext.topologyService() during RouterFactory.initialize(io.kroxylicious.proxy.router.RouterFactoryContext, C). The runtime creates the underlying cache on first request. Routers that never call topologyService() pay no cost.

Discovery methods

The three discovery methods — leaders(java.util.Map<java.lang.String, java.util.Set<java.lang.String>>), coordinators(java.lang.String, byte, java.util.Set<java.lang.String>), topicNames(java.lang.String, java.util.Set<org.apache.kafka.common.Uuid>) — are async and return self-contained result objects. They may send requests internally to warm the cache (METADATA for leaders and topic names, FIND_COORDINATOR for coordinators). The results are valid immediately upon completion and do not require further cache queries.

Cache population

The cache is populated as a side effect of responses flowing through the routing pipeline:

  • METADATA responses populate partition leaders, replicas, ISR, broker info, and topicId→name mappings.
  • FIND_COORDINATOR responses populate coordinator mappings, using request-side context (keyType, key) carried by the runtime's PendingResponse.

By the time a discovery method's CompletionStage completes, the cache is guaranteed to reflect the response.

Cache scope

The cache is shared per router level (not per connection), so all connections through the same router level share the same topology view. The cache is thread-safe.

Authorization and cache scope

The topology cache is not scoped by authenticated subject. When Kafka brokers filter METADATA responses based on ACLs, different connections may receive different subsets of topics. Because the cache is shared, it converges toward the union of all connections' views: each METADATA response adds or updates entries for topics present in the response, but does not remove entries for topics absent from the response.

This is safe because the cache is a routing optimization, not an access-control mechanism:

  • The cache stores where to send requests (partition leaders, coordinator nodes, broker addresses), not whether a client is authorized.
  • Backend brokers enforce ACLs on every request. A client that resolves a leader from the cache but lacks permission to produce or fetch will receive an authorization error from the broker, exactly as it would without the proxy.
  • Cache misses trigger discovery requests on the current connection's sender, which carries that connection's authentication context. A low-privilege connection's filtered METADATA response adds a subset of entries without corrupting entries populated by other connections.

Router authors must not use the topology cache for authorization decisions. The cache may contain entries populated by connections with different privileges. Routers should use the cache only for routing (leader selection, coordinator discovery, broker address resolution) and rely on backend broker ACL enforcement for access control. If a router needs to make authorization decisions, it should use RouterContext.authenticatedSubject() and consult an external authorization service.

Cache consistency and staleness

The cache uses additive semantics: responses add or update entries but never remove entries for topics absent from the response. The cache does not observe topic lifecycle events (deletion, recreation) independently — it only learns about cluster state from responses that flow through the proxy. Stale entries can therefore persist, for example partition leaders for a topic that has been deleted and recreated with different partition assignments.

Staleness is safe because the cache is a routing optimization, not an authoritative source of cluster state. Stale routing decisions (including incorrect fan-out grouping) produce broker error codes (NOT_LEADER_OR_FOLLOWER, UNKNOWN_TOPIC_OR_PARTITION), never silent misdirection or data corruption — the broker validates every request against its own authoritative state. Routers should treat these errors as staleness indicators and call invalidateRoute(java.lang.String) to trigger cache repopulation from subsequent responses.

  • Method Details

    • leaders

      CompletionStage<PartitionLeaders> leaders(Map<String,Set<String>> topicsByRoute)
      Discovers partition leaders for the given topics on the given routes, sending METADATA requests as needed. Uncached topics are batched into one METADATA request per route.

      The returned PartitionLeaders is a self-contained snapshot — callers should use it directly rather than querying the cache separately.

      Parameters:
      topicsByRoute - map from route name to set of topic names
      Returns:
      a stage that completes with the discovered leaders

    • coordinators

      CompletionStage<Coordinators> coordinators(String route, byte keyType, Set<String> keys)
      Discovers coordinators for the given keys on the given route, sending METADATA (if needed) then FIND_COORDINATOR. Supports batched lookup matching the FIND_COORDINATOR v4+ protocol.

      The returned Coordinators is a self-contained snapshot — callers should use it directly rather than querying the cache separately.

      Parameters:
      route - the route name
      keyType - 0 for group, 1 for transaction
      keys - the group or transaction IDs to discover
      Returns:
      a stage that completes with the discovered coordinators

    • topicNames

      CompletionStage<Map<org.apache.kafka.common.Uuid,String>> topicNames(String route, Set<org.apache.kafka.common.Uuid> topicIds)
      Resolves topic IDs to topic names for a given route, batching cache misses into a single METADATA request.

      The route parameter is required because per-route filter chains can transform topic names differently — the same cluster-level topic ID can map to different router-visible names on different routes.

      Returns a map containing an entry for each topic ID that was successfully resolved. Topic IDs that could not be resolved (e.g. deleted topics) are absent from the returned map.

      Parameters:
      route - the route to resolve names for
      topicIds - the topic IDs to resolve
      Returns:
      a stage that completes with the resolved mappings

    • partitionInfo

      Optional<PartitionInfo> partitionInfo(String topicName, int partitionIndex)
      Returns full partition info (leader, replicas, ISR) for a topic-partition, or empty if not cached.

      This is a supplementary lookup for use cases like follower-fetch / AZ-aware routing where the router needs replica and rack information beyond what PartitionLeaders provides. If this returns empty, the router can fall back to the leader from PartitionLeaders.

      Parameters:
      topicName - the topic name
      partitionIndex - the partition index
      Returns:
      the partition info, or empty if not cached

    • brokerInfo

      Optional<BrokerInfo> brokerInfo(VirtualNode node)
      Returns broker metadata (host, port, rack) for a virtual node, or empty if not cached.
      Parameters:
      node - the virtual node
      Returns:
      the broker info, or empty if not cached

    • invalidateRoute

      void invalidateRoute(String route)
      Coarse invalidation: clears all partition info, coordinators, broker info, and topic ID to name mappings for a route.

      Topic ID to name mappings for the route are also cleared, because per-route filter chains can present different names for the same underlying cluster topic.

      Called by the router when it observes staleness indicators (e.g. NOT_LEADER_OR_FOLLOWER, NOT_COORDINATOR) in responses. No background refresh is fired — the client drives the refresh via its own METADATA request, which repopulates the cache as a side effect.

      Parameters:
      route - the route to invalidate