Filtering & Conditional Listeners

Not every record on a topic deserves a trip through your business logic. Some events are heartbeats, some are types your service doesn’t care about, and some are duplicates you can cheaply discard. Spring for Apache Kafka lets you drop these records before they reach your @KafkaListener method using a RecordFilterStrategy, and lets you subscribe dynamically with topic patterns or gate listeners with SpEL. Filtering at the framework layer keeps your handler code clean and avoids paying serialization and processing costs for messages you’d only throw away.

Filtering with RecordFilterStrategy

A RecordFilterStrategy<K, V> is a single-method interface: return true to filter (discard) a record, false to keep it. You attach it to a ConcurrentKafkaListenerContainerFactory, and the framework consults it for every polled record before invoking your listener. Crucially, filtered records are still committed (the offset advances), so they won’t be redelivered — the consumer simply skips the handler call.

The example below ignores any OrderEvent that is a synthetic test event, so production listeners never see them:

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.adapter.RecordFilterStrategy;

@Configuration
public class FilteringConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderEvent> orderListenerFactory(
            ConsumerFactory<String, OrderEvent> consumerFactory) {

        var factory = new ConcurrentKafkaListenerContainerFactory<String, OrderEvent>();
        factory.setConsumerFactory(consumerFactory);

        // Return true to DROP the record before it reaches the listener.
        factory.setRecordFilterStrategy(record -> record.value().test());

        // Optional: also commit filtered offsets without invoking the listener.
        factory.setAckDiscarded(true);
        return factory;
    }
}

public record OrderEvent(String orderId, String type, boolean test) { }

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class OrderListener {

    @KafkaListener(
        topics = "orders",
        containerFactory = "orderListenerFactory",
        groupId = "order-service")
    public void onOrder(OrderEvent event) {
        // Only non-test events arrive here.
        System.out.println("Processing order " + event.orderId());
    }
}

The ackDiscarded flag matters. When true, the container acknowledges discarded records so their offsets are committed; when false (the default), a filtered record’s offset may not advance, risking redelivery on rebalance. For a pure “drop and forget” filter, set it to true.

Filtering batch listeners

When batchListener is enabled, the same strategy is applied per element. By default Spring removes the filtered records and delivers the surviving list; if every record is filtered, set factory.setBatchListener(true) together with RecordFilterStrategy.filterBatch overrides if you need custom whole-batch logic. Most teams simply reuse the per-record method and let the framework prune the batch.

Subscribing with topic patterns

Instead of naming each topic, a listener can subscribe to a regular expression and pick up matching topics automatically — including ones created after the consumer starts. Use topicPattern instead of topics:

@KafkaListener(
    topicPattern = "orders\\..*",   // matches orders.us, orders.eu, ...
    groupId = "order-service")
public void onRegionalOrder(OrderEvent event) {
    System.out.println("Regional order: " + event.orderId());
}

New topics matching the pattern are discovered on the consumer’s metadata refresh interval, controlled by metadata.max.age.ms (default 5 minutes). Lower it if you need faster pickup of freshly created topics.

Pattern subscriptions are convenient but blunt. A typo or overly broad regex can silently start consuming unrelated topics and shift load onto your group. Prefer explicit topic lists for critical pipelines and reserve patterns for multi-tenant or partitioned-by-region designs.

Conditional listeners with SpEL

@KafkaListener attributes are SpEL-evaluated, so you can enable or configure listeners from properties and the environment. This is ideal for toggling a listener per profile or wiring topic names from config without recompiling.

@KafkaListener(
    topics = "#{@kafkaTopics.orders}",                 // bean property reference
    groupId = "${app.kafka.group-id}",                  // property placeholder
    autoStartup = "${app.kafka.orders.enabled:true}",   // SpEL-driven on/off switch
    concurrency = "${app.kafka.orders.concurrency:3}")
public void onOrder(OrderEvent event) {
    // ...
}

app:
  kafka:
    group-id: order-service
    orders:
      enabled: true
      concurrency: 3

Setting autoStartup to false registers the container but leaves it stopped; you can start it later via the KafkaListenerEndpointRegistry. This pairs well with feature flags and blue-green rollouts.

Choosing an approach

Technique	Decides	Best for
`RecordFilterStrategy`	Per-record, at runtime	Dropping event types, test traffic, or duplicates
`topicPattern`	Which topics to subscribe to	Multi-tenant / per-region topics, dynamic topic sets
SpEL in attributes	Listener config & startup	Profile/feature-flag toggles, externalized config

Best Practices

Keep RecordFilterStrategy logic cheap and side-effect free — it runs for every record on every poll.
Set ackDiscarded(true) for drop filters so filtered offsets commit and records aren’t redelivered after a rebalance.
Filter on cheap signals (headers, a type field) rather than fields buried deep in a payload that force expensive deserialization.
Anchor topic-pattern regexes tightly (e.g. orders\\..*, not .*) and review them, since a broad pattern can pull in unexpected topics.
Lower metadata.max.age.ms only if you genuinely need fast discovery of new topics; it increases metadata traffic.
Use autoStartup with a SpEL flag instead of commenting out listeners, so configuration drives behavior across environments.
Log filtered records at debug level (or count them via a metric) so silent drops remain observable in production.