I have a Kafka topic with 3 Partitons and 3 replications.How does my Storm Topology will accept the data from these 3 Partitions and Replication
<p>I have Created a Kafka topic with the 3 Partitions and 3 Replications </p>

<p><strong>$KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic test</strong></p>

<p>and Wrote the Kafka-Producer Program to get the data from API and it works fine. My data is stored in the 3 partitions and 3 Replications successfully.</p>

<p>now I have to process the data which is in the Partitioned Kafka-logs. I have a Storm Topology. How do my Topology will accept the data from the logs as an input.</p>

<p>Initially I have created a Topic with 1 Partition,1 Replication and my storm topology works fine.But when I kept the Partition as 3 and replication is 3 then the data is not coming in to the Storm-Topology. How to resolve it.</p>

<p>my topology code</p>

<pre><code> public static void main(String[] args) throws Exception{

            Map map = Maps.newHashMap();
            map.put("dataSourceClassName", "org.postgresql.ds.PGSimpleDataSource");
            map.put("dataSource.url","jdbc:postgresql://localhost:5432/test?user=postgres");

            ConnectionProvider cp = new MyConnectionProvider(map);

            String argument = args[0];
            Config conf = new Config();
            conf.put(JDBC_CONF, map);
            conf.setDebug(true);
            conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 3);
            //set the number of workers
            conf.setNumWorkers(3);

            TopologyBuilder builder = new TopologyBuilder();

           //Setup Kafka spout
            BrokerHosts hosts = new ZkHosts("localhost:2181");
            String topic = "test";
            String zkRoot = "";
            String consumerGroupId = "group1";
            SpoutConfig spoutConfig = new SpoutConfig(hosts, topic, zkRoot, consumerGroupId);
            spoutConfig.scheme = new RawMultiScheme();
            spoutConfig.scheme = new SchemeAsMultiScheme(new Search_Parser());
            KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
            builder.setSpout("KafkaSpout", kafkaSpout);
</code></pre>

<p><strong>Help Me on This Issue</strong></p>

<p>Thanks in Advance </p>