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
Second, yes, you are right - we can further increase the concurrency of the fetch process, by using multiple fetch tasks for a single broker (each with a number of partitions allocated to them, and potentially reaching the 1-to-1 situation you are describing). I had that in mind with the current design of the FetchTask, it's just an addition I had considered for later. As the SimpleConsumer is deemed to be thread-safe, we can reuse the instance. 

On this point, two remarks: first, overpartitioning is quite common (it's deemed to be good practice to have a large number of partitions). Secondly, the typical bottleneck in this scenario is the NIC, so increasing the concurrency of the fetch may not yield so many benefits as running multiple instances of the listener/channel adapter on separate machines (each with a subset of partitions). This is the goal to be achieved by Spring XD, when we would create multiple instances of the channel adapter, each configured with a partition subset. Nonetheless, having multiple fetchers per broker is something desirable, and worth adding (as per your comment), so I'll add that as an enhancement. 

As you mention "our" consumer. We do not operate with the consumer group abstraction at all. The ChannelAdapter is *the* consumer for a set of partitions (we can configure a set of topics but that is, fundamentally, shorthand for "all partitions in given topics"), so the fetch request will always include only the partitions we are interested in. The further division that happens in the `ConcurrentMessageListenerDispatcher` is just for allowing the concurrent processing of messages, while keeping the ordering within a partition.