Posts /

How to bootstrap KTables in Kafka Streams 0.10


One of the most common use-cases in a streaming application is being able to join two different input sources into a new stream. Kafka Streams enables this behavior through a variety of join operations, including join, leftJoin, and outerJoin. Since joins are a stateful operation in Kafka Streams, a state store is utilized every time joins are used. However, there is a known issue (see KAFKA-4113) that, under certain conditions, will prevent the state store from being fully materialized. When this occurs, the following unintended side-effects may be observed:

This article will describe the conditions under which this behavior manifests itself, and proposes some possible solutions for working around this behavior.


Before diving into the possible solutions, let first reproduce the issue so that we can understand the circumstances that lead to the state store not being fully materialized.

Broker setup

First, set up some brokers using the wurstmeister/kafka-docker image.

Create a docker machine

$ docker-machine create --driver virtualbox default
$ eval $(docker-machine env)

Start the Kafka and Zookeeper containers

$ git clone && cd kafka-docker

Place the following configuration in a file named docker-compose-kafka-all.yml.

version: '2'
    image: wurstmeister/zookeeper
      - "2181:2181"
    image: wurstmeister/kafka
      - "9092:9092"
      KAFKA_CREATE_TOPICS: "test:1:1"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      - /var/run/docker.sock:/var/run/docker.sock

Now, start the containers.

$ docker-compose -f docker-compose-kafka-all.yml up -d

If everything goes well, you should be able to telnet to the broker and zookeeper node.

$ telnet $(docker-machine ip default) 9092
$ telnet $(docker-machine ip default) 2181


public void forceKTableBootstrap(String sourceTopic, Deserializer keyDeserializer, Deserializer valueDeserializer) {
        try {
            Map<String, Object> config = getStreamsConfig();
            String groupId = config.get(APPLICATION_ID_CONFIG).toString();
            config.put(GROUP_ID_CONFIG, groupId);
            config.put(ENABLE_AUTO_COMMIT_CONFIG, true);
            config.put(MAX_POLL_RECORDS_CONFIG, 1);
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config, keyDeserializer, valueDeserializer);
            ConsumerRebalanceListener listener = new ConsumerRebalanceListener() {
                public void onPartitionsRevoked(Collection<TopicPartition> partitions) {

                public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                    List<TopicPartition> unInitialized = new ArrayList<>();
                    for (TopicPartition tp : partitions) {
                        OffsetAndMetadata offsetAndMetaData = consumer.committed(tp);
                        if (offsetAndMetaData == null) {
                    if (unInitialized.size() > 0) {
              "Bootstrapping {} state stores for topic: {}", unInitialized.size(), sourceTopic);
                    } else {
              "{} state stores have already been bootstrapped for topic: {}", partitions.size(), sourceTopic);
            consumer.subscribe(Collections.singletonList(sourceTopic), listener);
        } catch (InconsistentGroupProtocolException e) {
  "Inconsistent group protocol while bootstrapping KTable. Restarting");