Home » Java » Enterprise Java » Apache Storm: How to configure KafkaBolt with Flux

About Adrianos Dadis

Adrianos Dadis
Adrianos is working as senior software engineer in telcos business domain. Particularly interested in enterprise integration, multi-tier architecture and middleware services. He mainly works with Weblogic, JBoss, Java EE, Spring, Drools, Oracle SOA Suite and various ESBs.

Apache Storm: How to configure KafkaBolt with Flux

Flux in a mini framework that can help us define and deploy a Storm topology.

Flux has various wrappers that help you define the required stream(s) and initialize your Bolts and Spouts (using constructor with or without arguments and call custom configuration methods automatically via reflection).

What you only need to use Flux is to add it as dependency in your “pom.xml”, configure it via a single YAML file (check flux examples) and then use it as main class to deploy your topology in a Storm cluster (or as local test).

In order to initialize a KafkaBolt the following steps are needed:

  1. Define a “topicSelector” via “withTopicSelector” method
  2. Define a “kafkaMapper” via “withTupleToKafkaMapper” method
  3. Define a “kafkaProducerProps” via “withProducerProperties” method
  4. Initialize “org.apache.storm.kafka.bolt.KafkaBolt” with above configuration
  5. Include above KafkaBolt as part of a stream

Minimal Flux configuration example for KafkaBolt:

  - id: "stringScheme"
    className: "org.apache.storm.kafka.StringScheme"

  - id: "stringMultiScheme"
    className: "org.apache.storm.spout.SchemeAsMultiScheme"
      - ref: "stringScheme"

  - id: "zkHosts"
    className: "org.apache.storm.kafka.ZkHosts"
      - "localhost:2181"

  - id: "topicSelector"
    className: "org.apache.storm.kafka.bolt.selector.DefaultTopicSelector"
      - "myTopicName"

  - id: "kafkaMapper"
    className: "org.apache.storm.kafka.bolt.mapper.FieldNameBasedTupleToKafkaMapper"

  - id: "kafkaProducerProps"
    className: "java.util.Properties"
      - name: "put"
          - "bootstrap.servers"
          - "localhost:9092"
      - name: "put"
          - "acks"
          - "1"
      - name: "put"
          - "key.serializer"
          - "org.apache.kafka.common.serialization.StringSerializer"
      - name: "put"
          - "value.serializer"
          - "org.apache.kafka.common.serialization.StringSerializer" 

  - id: "bolt-kafka"
    className: "org.apache.storm.kafka.bolt.KafkaBolt"
    parallelism: 1
      - name: "withProducerProperties"
        args: [ref: "kafkaProducerProps"]
      - name: "withTopicSelector"
        args: [ref: "topicSelector"]
      - name: "withTupleToKafkaMapper"
        args: [ref: "kafkaMapper"]

  - name: "spout --> kafkaBolt"
    from: "spout-1"
    to: "bolt-kafka"
      type: LOCAL_OR_SHUFFLE

For a full working configuration example check this, which can be used like this.

Example command to deploy your topology on Storm:

storm jar target/sentiment-analysis-storm-0.0.1-SNAPSHOT.jar org.apache.storm.flux.Flux --remote --c nimbus.host= src/test/resources/flux/topology_kafka.yaml

Flux configuration for KafkaSpout is already described as official flux example. Flux is a really helpful framework that eliminates required custom code to define and initialize a topology

(0 rating, 0 votes)
You need to be a registered member to rate this.
Start the discussion Views Tweet it!
Do you want to know how to develop your skillset to become a Java Rockstar?
Subscribe to our newsletter to start Rocking right now!
To get you started we give you our best selling eBooks for FREE!
1. JPA Mini Book
2. JVM Troubleshooting Guide
3. JUnit Tutorial for Unit Testing
4. Java Annotations Tutorial
5. Java Interview Questions
6. Spring Interview Questions
7. Android UI Design
and many more ....
I agree to the Terms and Privacy Policy
Notify of

This site uses Akismet to reduce spam. Learn how your comment data is processed.

Inline Feedbacks
View all comments