My `Kafka Producer` is producing messages at the rate of about .. `350 mb` per `30 seconds`.. 

`Kafka Setup`:

--> `1 Zookeeper instance`

--> `3 Kafka Brokers`

--> `1 Java Producer`

--> `1 Java Consumer`

This is how I created `Topic and broker partitions`:

`bin/kafka-topics.sh --create --zookeeper 10.10.1.5:2181 --replication-factor 1 --partitions 8 --topic test`

Rest of the configuration is as follows..

`Producer Code`:

```
KeyedMessage<String, byte[]> publishData = new KeyedMessage<String, byte[]>(this.topic, data);
producer.send(publishData);

```

Here, `data` is a `5000` length `byte[]`.

`Producer Config`:

```
batch.size = 200
producer.type = async
sflow-topic = test
connect.timeout.ms = 10000
request.required.acks = 0
zk.connect = 10.10.1.5:2181
serializer.class = kafka.serializer.DefaultEncoder
partitioner.class = kafka.producer.DefaultPartitioner
metadata.broker.list = 10.10.1.5:9092,10.10.1.6:9092,10.10.1.7:9092

```

I can see my producer working just fine.. Problem is with the consumer consuming the messages.. Even if the consumer is lagging behind, I do not see my messages consumed (and eventually process, and inserted in a DB) in the same/equal pace.. Also, I ran few tests on the consumer where I found that, not all the messages are consumed by my consumer.. Not sure why :(

`Consumer Code`:

```
 public class FlowConsumer {
    private final String topic;
    private final ExecutorService threadPool;
    private final ConsumerConnector consumer;
    private static AppProperties appProperties;
    private final ExecutorService processDataThreadPool;

    public FlowConsumer() throws Exception {
        /**
         * Load properties configuration for flowLog4j.properties.
         */
        appProperties = AppProperties.loadConfiguration();

        /** Assign the flow-topic.. */
        this.topic = appProperties.getString(AppConstants.FLOW_TOPIC);
        logger.fatal("Topic : "+topic);

        /** Initialize the thread pool to consume kafka byte[] streams.. */
        this.threadPool = Executors.newFixedThreadPool(20);

        /** Initialize the thread pool for processing kafka byte[] messages.. */
        this.processDataThreadPool = Executors.newFixedThreadPool(100);

        /** Fetch the Consumer Config, by reading the Flow.properties file.. */
        this.consumer = kafka.consumer.Consumer.createJavaConsumerConnector(ConsumerPropertyConfig.getConsumerConfig(appProperties));

        logger.fatal("Consumer : "+consumer);

        //new Thread(new Consumer()).start();
        threadPool.submit(new Consumer());
    }

    public void shutdown() {
        if (consumer != null) consumer.shutdown();
        if (threadPool != null) threadPool.shutdown();
        if (processDataThreadPool!= null) processDataThreadPool.shutdown();
    }

    private class Consumer implements Runnable {

        public Consumer() {
            logger.fatal("Started Consumer Thread!");
        }

        @Override
        public void run() {
            Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
            Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
            List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
            for (final KafkaStream<byte[], byte[]> kafkaStream : streams) {
                for (MessageAndMetadata<byte[], byte[]> messageAndMetadata : kafkaStream) {
                    processDataThreadPool.submit(new FlowServiceImpl(messageAndMetadata.message()));
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        FlowConsumer consumer = new FlowConsumer();

        /*try {
            Thread.sleep(10000);
        } catch (InterruptedException ie) {

        }
        consumer.shutdown();*/
    }
}

```

`Consumer Config`:

```
group.id = group1
flow-topic = test
auto.offset.reset = smallest
auto.commit.interval.ms = 2000
zookeeper.connect = 10.10.1.5:2181
zookeeper.sync.time.ms = 2000
zookeeper.session.timeout.ms = 2000
zookeeper.connection.timeout.ms = 6000

```

`Question 1`: 

For 3 brokers, can/should I create more than `3` partitions? I read that more partitions means, I can add more parallelism to my consumer? But how, by using more consumer thread on a single consumer? Or by having 3 consumer instances, having 1 thread each?

`Question 2`:

Is my Java consumer config code `correct/wrong`? 

Can anyone please tell me what am I doing wrong in here?