XD-2357-2391 Initial implementation of Kafka ChannelAdapter using the SimpleConsumer API

XD-2357-2391 Initial implementation of Kafka ChannelAdapter using the SimpleConsumer API

The implementation has the following components:

- KafkaConnectionFactory - manages and caches connections to a set of brokers
- KafkaBrokerConnection - handles low-level SimpleConsumer API calls and converts them to internal objects such as KafkaMessage and KafkaMessageBus
- KafkaTemplate - implements higher-level read operations on a Kafka broker
- OffsetManager - stores (potentially in a persistent fashion) offsets for a group of consumers, configures/resets initial offsets as necessary - can be configured to start at an arbitrary offset in a partition (e.g. replay since offset 100), or relative to a given timestamp (e.g. replay since Monday)
- KafkaMessageListenerContainer - retrieves messages from a given broker for an arbitrary set of partitions or topics (i.e. all partitions in the topics), invoking a MessageListener. Concurrency is adjustable, and allows processing multiple sets of partitions in parallel (while preserving ordering within a partition). Can poll multiple brokers (each on a parallel thread)
- AbstractDecodingMessageListener - utility base class for a MessageListener implementation that decodes the payload and key
- KafkaInboundChannelAdapter delegating to the KafkaMessageListenerContainer

Tests:
- Single and multi-broker configurations
- Partition subset retrieval
- Replicated sets
- Compression

TO DO:
- add tests for arbitrary start offsets
- add support for leader change failover
- add support for (optionally) retrieving configuration/leader information from Zookeeper
- namespace support
One last question (for today). While testing, I see we have one `SimpleConsumer` receiving data for 5 partitions and then distribute that work to one or more `QueueingMessageListenerInvoker` (always sending data for a particular partition to the same invoker).

I can see that for a large number of partitions this might be a good idea but, for a small number of partitions, would it not be better to have a `SimpleConsumer` for each partition?

It's also not clear to me if, say, we have 2 consumers in the consumer group; my understanding is one will get 3 partitions and the other 2; what happens when a consumer (in that case) asks for data from all 5 partitions? Again, with large partitions, can't we somehow only include "our" partitions in the fetch request?

Also, isn't it the case that a leader rebalance might move the leader to a new broker? We only seem to look for the leader during container `start()`.