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
Actually, this class is completely wrong and never should have made it to the PR in this state.

The intent here was not to publish on a channel, but rather to publish to a configurable DLQ topic. The reason why I stayed away from the `MessagePublishingErrorHandler` was that I was trying to avoid tying the `KafkaMessageListenerContainer` to Spring Integration (although I suppose it could depend on the standard Messaging abstraction in Spring 4 somehow). As for the Kafka `ErrorHandler`'s signature itself, my intent was to force any handler of an exception thrown at `KafkaMessageListenerContainer` level to process the original message too (which the `org.springframework.util.ErrorHandler` doesn't do, it only passes the exception). I suppose we could fall back to using the standard `org.springframework.util.ErrorHandler` and then make sure that the wrap the original message and exception, but I'm not a fan of downcasting all that much. 

In any case, the intent here is to have an injectable ErrorHandler that publishes the original message to a DLQ topic (so that we can inject that into the MessageListenerContainer via namespace  support for example), except that the pushed version her doesn't do any of it. I'll provide the correct implementation with the next revision.