Skip to content

EXPEDIA 2025-11-11 Tier 3

Read original ↗

Expedia — Colocating Input Partitions with Kafka Streams When Consuming Multiple Topics: Sub-Topology Matters!

Summary

Expedia debugs an in-production Kafka Streams application that consumed from two topics with identical partition counts and similar key strategies, expected Kafka Streams to route same-index partitions (Topic-1 partition 0 and Topic-2 partition 0) to the same instance for a shared in-memory Guava cache to be effective, and instead found identical keys processed on different instances — the cache was useless and the external API behind it was being called redundantly. Root cause: the two topics lived in two implicitly-separate sub-topologies (no shared operator / state / sink between them), and Kafka Streams only guarantees partition colocation across topics that are consumed within the same sub-topology. Fix: replace the local Guava cache with a Kafka Streams state store and attach that state store to both processing branches. The shared state-store dependency forced the two sub-topologies to merge into a single unified topology, which in turn made Kafka Streams assign same-index partitions across both topics to the same task — restoring partition colocation, cache locality, and the single-API-call-per-key guarantee.

Key takeaways

  1. Partition colocation in Kafka Streams is a sub-topology-scoped guarantee, not a topic-count-scoped one. The rule: Kafka Streams only guarantees partition colocation across topics if those topics are consumed within the same sub-topology and have the same number of partitions. Same partition count + similar keying across two topics is necessary but not sufficient — sub-topology co-membership is the missing condition. (Source: this post)

  2. Sub-topologies form implicitly from the way a Kafka Streams application is constructed. If two topic consumers have no shared operator or state store between them, Kafka Streams treats them as entirely separate execution graphs — their partitions are assigned independently, and the partition assignor makes no attempt to land same-index partitions on the same instance. Developers building streaming apps "from the top down" typically don't realise their app has been split in two until production behavior contradicts expectations. (Source: this post)

  3. The concrete production symptom was cache thrashing: identical keys from the two topics were being processed on different Kafka Streams instances, each instance maintained its own private Guava cache, and the external key-transformation API was being called redundantly for every copy of every key. The application logic was correct; the architecture assumed colocation that the runtime did not provide. (Source: this post)

  4. Expectations that drove the original design were reasonable but incomplete. "Two topics with the same partition count and the same key strategy → same-index partitions go to the same instance" is a reasonable heuristic and matches how partition-level keying works within a single topic. It breaks across topics unless Kafka Streams has a structural reason to assign them together. (Source: this post)

  5. The architectural fix is a shared state store attached to both processing branches. Replacing the in-memory Guava cache with a Kafka Streams state store — a distributed, fault-tolerant storage abstraction — and attaching it to both branches introduced a common dependency between them. Kafka Streams detected this dependency and merged the two sub-topologies into a single unified topology. (Source: this post)

  6. Once merged, partition assignment follows the unified topology. Kafka Streams assigned same-index partitions from both topics to the same task, identical keys were consistently routed to the same instance, and the shared state store (playing the cache role) absorbed the single-API-call-per-key discipline. Three named outcomes: partition colocation achieved, identical keys always routed to the same instance, external API calls significantly reduced. (Source: this post)

  7. General principle: in Kafka Streams, topology design directly influences partition assignment behavior. The topology is not just a dataflow description — it is the primary input to the partition assignor. When cross-topic coordination is needed without an explicit join (cache-sharing, cross-topic stats, dedup across topics), the developer has to deliberately introduce a shared operator or state store to force sub-topology unification. (Source: this post)

  8. Shared state store as a structural lever, not just a storage primitive. Using a Kafka Streams state store purely to unify topology (not to persist anything across restarts) is a pattern in its own right. The post explicitly names it: "Using a shared state store as a unifying element is a powerful pattern — not only for sharing data, but also for influencing execution behavior in a distributed Kafka Streams application." (Source: this post)

Systems

  • systems/kafka-streams — the framework whose partition-assignment behavior the post is debugging. Topology (Topology + KafkaStreams runtime) is the central abstraction; sub-topologies are automatically derived from the topology's operator graph. The partition assignor (StreamsPartitionAssignor) takes the sub-topology structure as a primary input when assigning tasks.
  • systems/kafka — the messaging substrate; topics, partitions, and keyed producers are the primitives Kafka Streams consumes. Partition count + key-strategy parity between the two topics is Expedia's initial (necessary-but-not-sufficient) condition for expecting colocation.

Concepts

  • concepts/sub-topology — the implicit subgraph unit that Kafka Streams derives from a topology; inputs are assigned together; the load-bearing concept in the post — two topics in different sub-topologies cannot be colocated.
  • concepts/partition-colocation — property that same-index partitions across multiple topics are assigned to the same processing instance/task; enables cache-locality and single-shard-style scoping across topics.
  • concepts/cache-locality — identical keys arrive at the same processing node consistently, so a local cache sees the same keyspace across requests; the property Expedia was trying to preserve and lost.

Patterns

  • patterns/shared-state-store-as-topology-unifier — attach a Kafka Streams state store to two otherwise-disjoint processing branches specifically to force Kafka Streams to merge their sub-topologies into a unified topology, thereby restoring partition-colocation guarantees. The state store may or may not be load-bearing as storage; its structural role is topology-shape enforcement.

Operational numbers

None disclosed. The post is a production-debugging narrative with qualitative outcomes ("external API calls reduced significantly", "improving overall system performance and resilience"). No throughput, no API-cost baseline, no latency distribution, no partition/instance counts, no cache-hit-rate delta.

Caveats

  • No benchmarks or before/after numbers. The qualitative shape of the fix is clear; the magnitude is not.
  • No discussion of the state-store implementation choice. Kafka Streams offers RocksDB-backed, in-memory, and custom state stores with different fault-tolerance and latency profiles. The post doesn't say which Expedia used or whether the changelog topic was considered a cost axis.
  • Not a join use case — explicitly. The post is careful to note "we were not combining records across topics, just sharing a cache across both input streams". Kafka Streams does guarantee co-partitioning across topics in KStream.join() — the post's contribution is pointing out that the same guarantee does not extend to cross-topic cache-sharing without an explicit unifier.
  • No detail on partition-assignor mechanics. The post cites the rule but doesn't walk through StreamsPartitionAssignor's implementation or the co-partitioning check it performs (same partition count, same source-topic grouping via StreamsPartitionAssignor.copartitionGroups). Readers wanting to verify the claim against source code have to go to Kafka's codebase directly.
  • Single-application case study. Expedia's pipeline shape (two topics / same-key transform / external API / in-memory cache) is the worked example. The pattern generalises, but the post doesn't enumerate other shapes (e.g. N-topic dedup, cross-topic rate limiting, shared-stats aggregators) that would benefit from the same fix.

Raw source

Last updated · 200 distilled / 1,178 read