6acf985cc1
JIRA: https://jira.spring.io/browse/INTEXT-123 https://jira.spring.io/browse/INTEXT-124 https://jira.spring.io/browse/INTEXT-125 The implementation has the following components: - Configuration - allows retrieving a seed broker set either from a preconfigured list or from Zookeeper - ConnectionFactory - manages and caches connections to a set of brokers - Connection - 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) - KafkaInboundChannelAdapter - channel adapter using KMLC as underlying implementation - AbstractDecodingMessageListener - utility base class for a MessageListener implementation that decodes the payload and key Tests: - Single and multi-broker configurations - Partition subset retrieval - Replicated sets - Compression - Starting offsets (with reset on wrong offset) - Error handling (servers dropping and leader changes) TO DO: - namespace support Remove warnings Cleanup Use ZookeeperConnect for configuring Zookeeper Fixes, including: - Extracted Connection and ConnectionFactory interfaces - Made fields final where necessary - Fixed Javadoc - Added assertions - Corrected headers Updated copyright to 2015 Make connectivity parameters: timeouts, buffer size, minimum fetch size, fetch timeout configurable via ConnectionFactory. Remove sleep in KafkaMessageListenerContainer Add javadocs. Addressing PR comments: - renamed KafkaInboundChannelAdapter to KafkaMessageDrivenChannelAdapter - removed unnecessary setters - close ZkClient properly - fix version Must use TopicAndPartition as key Log messages too via LoggingErrorHandler Newline - Make `MessageStoreOffsetManager` implement `Closeable` and `Flushable` - Fix the configuration of `ErrorHandler` We don't really need this initialization - it's premature optimization Use ':' instead of ' ' as separator for the key Fixes: - Use 'getPhase' from superclass - more cleanup Defer to the superclass MessageBuilder factory and keep SI_FATAL_WHEN_NO_BEANFACTORY happy Fix warnings in tests Changes to support testing against external brokers - refactored the rules so that an abstraction can be used - topic name is configurable - added (Ignored) TestSingleBrokerExternal that tests against an external broker - refactored OutboundTest to use the new consumer Properly catch TopicExistsException Do not hold state for MetadataStoreOffsetManager In KafkaMessageListenerContainer FetcherTasks will block if they have no partitions to listen to (to do - use separate monitors per task) Updated test for offset reset to latest to wait until an actual refresh event takes place, thus preventing the test from not completing in case of a race condition KafkaMessageDrivenChannelAdapter and KafkaMessageListenerContainer now throw new TopicNotFoundException if the topic does not exist on the broker Renamed test classes to use the Test suffix. Cleanup Test if the component is still running when exiting the wait block Serialization warnings Addressing the last round of PR comments Test configuration cleanup The final polishing: code style, JavaDocs, renaming for test classes to the finish with `*Tests` suffix
Spring Integration 
Code of Conduct
Please see our Code of conduct.
Reporting Security Vulnerabilities
Please see our Security policy.
Checking out and Building
To check out the project and build from the source, do the following:
git clone git://github.com/spring-projects/spring-integration.git
cd spring-integration
./gradlew build
NOTE: While Spring Integration runs with Java SE 6 or higher, a Java 8 compiler is required to build the project.
If you encounter out of memory errors during the build, increase an available heap and permgen for Gradle:
GRADLE_OPTS='-XX:MaxPermSize=1024m -Xmx1024m'
To build and install jars into your local Maven cache:
./gradlew install
To build api Javadoc (results will be in build/api):
./gradlew api
To build reference documentation (results will be in build/reference):
./gradlew reference
To build complete distribution including -dist, -docs, and -schema zip files (results will be in build/distributions)
./gradlew dist
Using Eclipse
To generate Eclipse metadata (.classpath and .project files), do the following:
./gradlew eclipse
Once complete, you may then import the projects into Eclipse as usual:
File -> Import -> Existing projects into workspace
Browse to the 'spring-integration' root directory. All projects should import free of errors.
Using IntelliJ IDEA
To generate IDEA metadata (.iml and .ipr files), do the following:
./gradlew idea
Resources
For more information, please visit the Spring Integration website at: https://projects.spring.io/spring-integration