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
Yes, this would replace the bean created by the current namespace element (although we don't have namespace support for this yet) 

> On Dec 20, 2014, at 14:16, Artem Bilan <notifications@github.com> wrote:
> 
> In src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundChannelAdapter.java:
> 
> > +		super.onInit();
> > +	}
> > +
> > +	@Override
> > +	protected void doStart() {
> > +		this.messageListenerContainer.start();
> > +	}
> > +
> > +	@Override
> > +	protected void doStop() {
> > +		this.messageListenerContainer.stop();
> > +	}
> > +
> > +	@Override
> > +	public String getComponentType() {
> > +		return "amqp:inbound-channel-adapter";
> kafka:inbound-channel-adapter. Although we already use this name from KafkaHighLevelConsumerMessageSource and I guess that one isn't correct.
> 
> —
> Reply to this email directly or view it on GitHub.
>