Designing a Rate Limiter with Kafka Streaming

Designing a Rate Limiter with Kafka Streaming

What Is a Rate Limiter?

A rate limiter controls the number of requests or events a user, service, or resource can process within a given time window. Common use cases include API rate limiting, preventing abuse, managing quotas, and protecting downstream systems from overload. In distributed systems, a centralized rate limiter can become a bottleneck. Apache Kafka, combined with its stream processing capabilities, offers a scalable, fault‑tolerant way to implement rate limiting as part of your data pipeline.

Why Use Kafka for Rate Limiting?

Traditional rate limiters (e.g., token bucket in memory or Redis) work well for single‑node applications, but struggle with distributed deployments. Kafka provides:

• Scalability – Partitioning allows horizontal scaling of rate‑limiting logic.
• Durability – Events are persisted; no data loss during restarts.
• Stream Processing – Kafka Streams or ksqlDB can compute sliding‑window counts with exactly‑once semantics.
• Decoupling – The rate limiter becomes a reusable component in your event‑driven architecture.

Core Architecture

The basic design uses a Kafka topic for incoming events (e.g., API requests). A Kafka Streams application consumes these events, groups them by a key (e.g., user ID or IP address), and computes a count over a sliding time window. If the count exceeds a threshold, the event is either rejected (sent to a dead‑letter topic) or flagged for throttling. The stream processor can also produce a “rate‑limit exceeded” notification back to the caller or to a monitoring system.

How to Use It: Practical Implementation

Below is a complete Kafka Streams implementation in Java that implements a sliding‑window rate limiter. It counts requests per user in 1‑minute windows with a maximum of 10 requests per minute. Events that exceed the limit are sent to a rate-limited-events topic.

1. Maven Dependencies

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>3.6.0</version>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.0</version>
</dependency>

2. Rate Limiter Stream Processor

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.WindowStore;

import java.time.Duration;
import java.util.Properties;

public class KafkaRateLimiter {

    public static final int MAX_REQUESTS = 10;
    public static final long WINDOW_SIZE_MS = 60_000; // 1 minute

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "rate-limiter-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);

        StreamsBuilder builder = new StreamsBuilder();

        // Input topic: user requests (key = userId, value = request payload)
        KStream<String, String> input = builder.stream("incoming-requests");

        // Windowed count per user
        KTable<Windowed<String>, Long> requestCounts = input
                .groupByKey()
                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMillis(WINDOW_SIZE_MS)))
                .count(Materialized.<String, Long, WindowStore<Object, Long>>as("request-counts")
                        .withValueSerde(Serdes.Long()));

        // Join with original stream to decide allow/deny
        KStream<String, String> allowed = input
                .selectKey((key, value) -> key)  // keep key
                .join(requestCounts,
                        (requestValue, count) -> count,
                        Joined.with(Serdes.String(), Serdes.String(), Serdes.Long()))
                .filter((key, count) -> count <= MAX_REQUESTS)
                .mapValues((key, count) -> "ALLOWED");

        KStream<String, String> denied = input
                .selectKey((key, value) -> key)
                .join(requestCounts,
                        (requestValue, count) -> count,
                        Joined.with(Serdes.String(), Serdes.String(), Serdes.Long()))
                .filter((key, count) -> count > MAX_REQUESTS)
                .mapValues((key, count) -> "DENIED");

        allowed.to("allowed-requests");
        denied.to("rate-limited-events");

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

3. Explanation

• Windowed Counting – We use a hopping window of 1 minute (no grace period). The count is stored in a persistent WindowStore.
• Exactly‑Once Semantics – PROCESSING_GUARANTEE_CONFIG ensures no duplicate counts.
• Joining – We join the original stream with the windowed count table to decide if the current request should be allowed. Note: This is a simplified approach; in production you may want to use a Processor or Transformer to avoid the join overhead for every event.
• Output Topics – Allowed requests go to allowed-requests; denied ones go to rate-limited-events for downstream processing (e.g., logging, retry logic).

4. Alternative: Using a Transformer for Lower Latency

A more efficient approach uses a custom Transformer that accesses the state store directly, avoiding a KTable‑KStream join. Below is a snippet:

public class RateLimiterTransformer implements Transformer<String, String, KeyValue<String, String>> {
    private WindowStore<String, Long> stateStore;
    private final int maxRequests;
    private final long windowSizeMs;

    public RateLimiterTransformer(int maxRequests, long windowSizeMs) {
        this.maxRequests = maxRequests;
        this.windowSizeMs = windowSizeMs;
    }

    @Override
    public void init(ProcessorContext context) {
        stateStore = context.getStateStore("request-counts");
    }

    @Override
    public KeyValue<String, String> transform(String key, String value) {
        long windowStart = System.currentTimeMillis() - windowSizeMs;
        long count = 0;
        try (WindowStoreIterator<Long> iter = stateStore.fetch(key, windowStart, System.currentTimeMillis())) {
            while (iter.hasNext()) {
                count += iter.next().value;
            }
        }
        // Add current request (will be committed later)
        stateStore.put(key, count + 1, System.currentTimeMillis());
        if (count < maxRequests) {
            return KeyValue.pair(key, "ALLOWED:" + value);
        } else {
            return KeyValue.pair(key, "DENIED:" + value);
        }
    }

    @Override
    public void close() {}
}

You would attach this transformer to the stream using transform() and provide the state store.

Best Practices

• Choose the Right Window Type – Use sliding windows for smooth rate limiting (e.g., SlidingWindows in newer Kafka Streams versions) or hopping windows for discrete intervals.
• Handle High Cardinality Keys – Partition the input topic by the rate‑limiting key (e.g., user ID) to distribute load. Use a keyed state store per partition.
• Monitor and Alert – Track the number of denied events, lag in the rate‑limiter application, and state store sizes. Export metrics to Prometheus or similar.
• Grace Period and Retention – Set appropriate window retention and grace periods to avoid memory leaks. Use Materialized.withRetention(Duration).
• Idempotency – Use exactly‑once semantics to prevent double counting, especially if you reprocess events after failures.
• Backpressure – If the rate limiter itself becomes a bottleneck, consider pre‑partitioning or using multiple application instances.
• Fallback – For critical paths, provide a circuit breaker or a local in‑memory rate limiter as a fallback if Kafka is unavailable.

Conclusion

Designing a rate limiter with Kafka streaming gives you a durable, scalable, and stream‑native solution that integrates naturally with event‑driven architectures. By leveraging Kafka Streams’ windowed aggregations and state stores, you can enforce rate limits with exactly‑once guarantees while keeping the logic decoupled from individual services. The code examples provided demonstrate both a join‑based and a transformer‑based implementation. With careful consideration of window types, state store sizing, and monitoring, your Kafka‑based rate limiter can handle millions of events per second while protecting downstream resources.