Introduction

KafkaSpout is the Storm component that consumes records from Kafka and emits them into a topology as tuples. A working example is mostly about getting three pieces right: the spout configuration, the tuple translation, and the topology wiring between the spout and the bolts.

What KafkaSpout Does

At runtime, KafkaSpout acts like a Kafka consumer managed by Storm. It subscribes to one or more topics, polls records, converts each record into a Storm tuple, and tracks offsets according to the spout's processing guarantees.

The practical outcome is simple: Kafka remains the source of events, while Storm handles the real-time processing graph.

A Minimal Working Example

The example below uses Storm's Kafka client package to read strings from a topic named orders and print them in a bolt.

1import org.apache.kafka.clients.consumer.ConsumerConfig;
2import org.apache.kafka.common.serialization.StringDeserializer;
3import org.apache.storm.Config;
4import org.apache.storm.LocalCluster;
5import org.apache.storm.generated.StormTopology;
6import org.apache.storm.kafka.spout.ByTopicRecordTranslator;
7import org.apache.storm.kafka.spout.KafkaSpout;
8import org.apache.storm.kafka.spout.KafkaSpoutConfig;
9import org.apache.storm.topology.OutputFieldsDeclarer;
10import org.apache.storm.topology.TopologyBuilder;
11import org.apache.storm.topology.base.BaseBasicBolt;
12import org.apache.storm.topology.BasicOutputCollector;
13import org.apache.storm.tuple.Fields;
14import org.apache.storm.tuple.Tuple;
15import org.apache.storm.tuple.Values;
16
17public class KafkaSpoutExample {
18    public static void main(String[] args) throws Exception {
19        ByTopicRecordTranslator<String, String> translator =
20            new ByTopicRecordTranslator<>(
21                record -> new Values(record.key(), record.value()),
22                new Fields("key", "value")
23            );
24
25        KafkaSpoutConfig<String, String> spoutConfig = KafkaSpoutConfig.builder("localhost:9092", "orders")
26            .setProp(ConsumerConfig.GROUP_ID_CONFIG, "storm-orders-group")
27            .setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class)
28            .setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class)
29            .setRecordTranslator(translator)
30            .build();
31
32        TopologyBuilder builder = new TopologyBuilder();
33        builder.setSpout("kafka-spout", new KafkaSpout<>(spoutConfig));
34        builder.setBolt("print-bolt", new PrintBolt()).shuffleGrouping("kafka-spout");
35
36        Config config = new Config();
37        StormTopology topology = builder.createTopology();
38
39        try (LocalCluster cluster = new LocalCluster()) {
40            cluster.submitTopology("kafka-spout-example", config, topology);
41            Thread.sleep(30_000);
42        }
43    }
44
45    public static class PrintBolt extends BaseBasicBolt {
46        @Override
47        public void execute(Tuple input, BasicOutputCollector collector) {
48            System.out.println("Received: " + input.getStringByField("value"));
49        }
50
51        @Override
52        public void declareOutputFields(OutputFieldsDeclarer declarer) {
53        }
54    }
55}

This is enough to prove the spout is wired correctly in a local development setup.

Why the Record Translator Matters

Older examples sometimes assume the tuple fields automatically match what the bolt expects. In practice, you should define the tuple fields explicitly with a translator.

If your bolt expects value, the translator must emit a field named value. Otherwise the spout may consume correctly while the bolt fails at runtime with field lookup errors.

Offset and Delivery Semantics

KafkaSpout manages Kafka offsets based on Storm's processing lifecycle. That means reliability depends on both spout configuration and topology behavior. If a bolt fails or tuples are replayed, you should expect at-least-once behavior unless you design a stronger end-to-end strategy yourself.

In real systems, that usually means downstream processing should be idempotent.

For local testing, it also helps to keep the topology small at first: one topic, one spout, one bolt, and a clearly visible side effect such as logging. Once that works, add parsing, aggregation, or persistence one step at a time.

Common Pitfalls

A common mistake is forgetting the translator and then trying to read tuple fields that were never declared.

Another mistake is mixing incompatible Storm and Kafka client examples from different versions. KafkaSpout APIs changed over time, so use examples that match the Storm client library you actually depend on.

A third mistake is testing only the topology and forgetting to produce data into Kafka. If the topic is empty, the topology may look idle even though it is configured correctly.

Summary

• 'KafkaSpout consumes Kafka records and emits them into a Storm topology.'
• A working setup needs a spout config, a record translator, and at least one bolt.
• Explicit field translation prevents runtime tuple-field mismatches.
• Expect at-least-once style behavior unless your downstream design handles duplicates explicitly.
• Match your example code to the Storm Kafka client version you are actually using.