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
Thanks for spotting it :) Generally I'm following the same rule, but this one slipped through. 

> On Dec 20, 2014, at 14:27, Artem Bilan <notifications@github.com> wrote:
> 
> In src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java:
> 
> > +	@Override
> > +	public void stop() {
> > +		this.stop(null);
> > +	}
> > +
> > +	@Override
> > +	public boolean isRunning() {
> > +		return this.running;
> > +	}
> > +
> > +	@Override
> > +	public int getPhase() {
> > +		return 0;
> > +	}
> > +
> > +	static class GetPartitionsForTopic extends CheckedFunction<String, Iterable<Partition>> {
> Bla-bla-blaFunction. Yes, I see that it is a package protected, but we tipically follow with the name from super in Spring.
> 
> —
> Reply to this email directly or view it on GitHub.
>