{"action":"created","comment":{"url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/pulls/comments/22802905","id":22802905,"diff_hunk":"@@ -0,0 +1,62 @@\n+/*\n+ * Copyright 2015 the original author or authors.\n+ *\n+ * Licensed under the Apache License, Version 2.0 (the \"License\");\n+ * you may not use this file except in compliance with the License.\n+ * You may obtain a copy of the License at\n+ *\n+ * http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+\n+\n+package org.springframework.integration.kafka.listener;\n+\n+import org.springframework.beans.BeansException;\n+import org.springframework.context.ApplicationContext;\n+import org.springframework.context.ApplicationContextAware;\n+import org.springframework.integration.kafka.core.KafkaMessage;\n+import org.springframework.integration.kafka.listener.MessageListener;\n+import org.springframework.integration.support.MessageBuilder;\n+import org.springframework.integration.support.channel.BeanFactoryChannelResolver;\n+import org.springframework.messaging.MessageChannel;\n+import org.springframework.messaging.core.DestinationResolver;\n+\n+/**\n+ * @author Marius Bogoevici\n+ */\n+public class ChannelSendingMessageListener implements MessageListener, ApplicationContextAware {\n+\n+\tprivate String channelName;\n+\n+\tprivate MessageChannel messageChannel;\n+\n+\tpublic String getChannelName() {\n+\t\treturn channelName;\n+\t}\n+\n+\tpublic void setChannelName(String channelName) {\n+\t\tthis.channelName = channelName;\n+\t}\n+\n+\tDestinationResolver<MessageChannel> destinationResolver;\n+\n+\t@Override\n+\tpublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {\n+\t\tthis.destinationResolver = new BeanFactoryChannelResolver(applicationContext);\n+\t\tmessageChannel = destinationResolver.resolveDestination(channelName);\n+\t}\n+\n+\t@Override\n+\tpublic void onMessage(KafkaMessage message) {\n+\t\tbyte b[] = new byte[message.getMessage().payloadSize()];\n+\t\tmessage.getMessage().payload().get(b);\n+\t\tmessageChannel.send(MessageBuilder.withPayload(new String(b)).build());\n+\t\tSystem.out.println(\"Received \" + new String(b) + \" from partition \" + message.getMetadata().getPartition());","path":"src/test/java/org/springframework/integration/kafka/listener/ChannelSendingMessageListener.java","position":60,"original_position":60,"commit_id":"8f61fad6e535bb722c73613a4ecb36fc3e5ab20c","original_commit_id":"8f61fad6e535bb722c73613a4ecb36fc3e5ab20c","user":{"login":"mbogoevici","id":159684,"avatar_url":"https://avatars.githubusercontent.com/u/159684?v=3","gravatar_id":"","url":"https://api.github.com/users/mbogoevici","html_url":"https://github.com/mbogoevici","followers_url":"https://api.github.com/users/mbogoevici/followers","following_url":"https://api.github.com/users/mbogoevici/following{/other_user}","gists_url":"https://api.github.com/users/mbogoevici/gists{/gist_id}","starred_url":"https://api.github.com/users/mbogoevici/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mbogoevici/subscriptions","organizations_url":"https://api.github.com/users/mbogoevici/orgs","repos_url":"https://api.github.com/users/mbogoevici/repos","events_url":"https://api.github.com/users/mbogoevici/events{/privacy}","received_events_url":"https://api.github.com/users/mbogoevici/received_events","type":"User","site_admin":false},"body":"Ugh. Slipped.","created_at":"2015-01-12T16:49:23Z","updated_at":"2015-01-12T16:49:23Z","html_url":"https://github.com/spring-projects/spring-integration-kafka/pull/12#discussion_r22802905","pull_request_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/pulls/12","_links":{"self":{"href":"https://api.github.com/repos/spring-projects/spring-integration-kafka/pulls/comments/22802905"},"html":{"href":"https://github.com/spring-projects/spring-integration-kafka/pull/12#discussion_r22802905"},"pull_request":{"href":"https://api.github.com/repos/spring-projects/spring-integration-kafka/pulls/12"}}},"pull_request":{"url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/pulls/12","id":27107160,"html_url":"https://github.com/spring-projects/spring-integration-kafka/pull/12","diff_url":"https://github.com/spring-projects/spring-integration-kafka/pull/12.diff","patch_url":"https://github.com/spring-projects/spring-integration-kafka/pull/12.patch","issue_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/12","number":12,"state":"open","locked":false,"title":"INTEXT-123-124-125 Initial implementation of Kafka ChannelAdapter using ...","user":{"login":"mbogoevici","id":159684,"avatar_url":"https://avatars.githubusercontent.com/u/159684?v=3","gravatar_id":"","url":"https://api.github.com/users/mbogoevici","html_url":"https://github.com/mbogoevici","followers_url":"https://api.github.com/users/mbogoevici/followers","following_url":"https://api.github.com/users/mbogoevici/following{/other_user}","gists_url":"https://api.github.com/users/mbogoevici/gists{/gist_id}","starred_url":"https://api.github.com/users/mbogoevici/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mbogoevici/subscriptions","organizations_url":"https://api.github.com/users/mbogoevici/orgs","repos_url":"https://api.github.com/users/mbogoevici/repos","events_url":"https://api.github.com/users/mbogoevici/events{/privacy}","received_events_url":"https://api.github.com/users/mbogoevici/received_events","type":"User","site_admin":false},"body":"...the SimpleConsumer API\r\n\r\nThe implementation has the following components:\r\n\r\n- Configuration - allows retrieving a seed broker set either from a preconfigured list or from Zookeeper\r\n- ConnectionFactory - manages and caches connections to a set of brokers\r\n- Connection - handles low-level SimpleConsumer API calls and converts them to internal objects such as KafkaMessage and KafkaMessageBus\r\n- KafkaTemplate - implements higher-level read operations on a Kafka broker\r\n- 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)\r\n- 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)\r\n- KafkaInboundChannelAdapter - channel adapter using KMLC as underlying implementation\r\n- AbstractDecodingMessageListener - utility base class for a MessageListener implementation that decodes the payload and key\r\n\r\nTests:\r\n- Single and multi-broker configurations\r\n- Partition subset retrieval\r\n- Replicated sets\r\n- Compression\r\n- Starting offsets (with reset on wrong offset)\r\n- Error handling (servers dropping and leader changes)\r\n\r\nTO DO:\r\n- namespace support","created_at":"2015-01-09T15:35:51Z","updated_at":"2015-01-12T16:49:23Z","closed_at":null,"merged_at":null,"merge_commit_sha":"1e856a0a81eee890f4853950fdd1f4c79113f096","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/pulls/12/commits","review_comments_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/pulls/12/comments","review_comment_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/pulls/comments/{number}","comments_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/12/comments","statuses_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/statuses/8f61fad6e535bb722c73613a4ecb36fc3e5ab20c","head":{"label":"mbogoevici:INTEXT-123-124-125","ref":"INTEXT-123-124-125","sha":"8f61fad6e535bb722c73613a4ecb36fc3e5ab20c","user":{"login":"mbogoevici","id":159684,"avatar_url":"https://avatars.githubusercontent.com/u/159684?v=3","gravatar_id":"","url":"https://api.github.com/users/mbogoevici","html_url":"https://github.com/mbogoevici","followers_url":"https://api.github.com/users/mbogoevici/followers","following_url":"https://api.github.com/users/mbogoevici/following{/other_user}","gists_url":"https://api.github.com/users/mbogoevici/gists{/gist_id}","starred_url":"https://api.github.com/users/mbogoevici/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mbogoevici/subscriptions","organizations_url":"https://api.github.com/users/mbogoevici/orgs","repos_url":"https://api.github.com/users/mbogoevici/repos","events_url":"https://api.github.com/users/mbogoevici/events{/privacy}","received_events_url":"https://api.github.com/users/mbogoevici/received_events","type":"User","site_admin":false},"repo":{"id":28195714,"name":"spring-integration-kafka","full_name":"mbogoevici/spring-integration-kafka","owner":{"login":"mbogoevici","id":159684,"avatar_url":"https://avatars.githubusercontent.com/u/159684?v=3","gravatar_id":"","url":"https://api.github.com/users/mbogoevici","html_url":"https://github.com/mbogoevici","followers_url":"https://api.github.com/users/mbogoevici/followers","following_url":"https://api.github.com/users/mbogoevici/following{/other_user}","gists_url":"https://api.github.com/users/mbogoevici/gists{/gist_id}","starred_url":"https://api.github.com/users/mbogoevici/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mbogoevici/subscriptions","organizations_url":"https://api.github.com/users/mbogoevici/orgs","repos_url":"https://api.github.com/users/mbogoevici/repos","events_url":"https://api.github.com/users/mbogoevici/events{/privacy}","received_events_url":"https://api.github.com/users/mbogoevici/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/mbogoevici/spring-integration-kafka","description":"","fork":true,"url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka","forks_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/forks","keys_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/teams","hooks_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/hooks","issue_events_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/issues/events{/number}","events_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/events","assignees_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/assignees{/user}","branches_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/branches{/branch}","tags_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/tags","blobs_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/languages","stargazers_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/stargazers","contributors_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/contributors","subscribers_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/subscribers","subscription_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/subscription","commits_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/issues/comments/{number}","contents_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/contents/{+path}","compare_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/merges","archive_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/downloads","issues_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/issues{/number}","pulls_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/labels{/name}","releases_url":"https://api.github.com/repos/mbogoevici/spring-integration-kafka/releases{/id}","created_at":"2014-12-18T18:19:08Z","updated_at":"2014-12-18T18:19:09Z","pushed_at":"2015-01-12T15:33:36Z","git_url":"git://github.com/mbogoevici/spring-integration-kafka.git","ssh_url":"git@github.com:mbogoevici/spring-integration-kafka.git","clone_url":"https://github.com/mbogoevici/spring-integration-kafka.git","svn_url":"https://github.com/mbogoevici/spring-integration-kafka","homepage":null,"size":493,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"spring-projects:master","ref":"master","sha":"6ab8ec33f18125e72b87d15683e48ea2cba17382","user":{"login":"spring-projects","id":317776,"avatar_url":"https://avatars.githubusercontent.com/u/317776?v=3","gravatar_id":"","url":"https://api.github.com/users/spring-projects","html_url":"https://github.com/spring-projects","followers_url":"https://api.github.com/users/spring-projects/followers","following_url":"https://api.github.com/users/spring-projects/following{/other_user}","gists_url":"https://api.github.com/users/spring-projects/gists{/gist_id}","starred_url":"https://api.github.com/users/spring-projects/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/spring-projects/subscriptions","organizations_url":"https://api.github.com/users/spring-projects/orgs","repos_url":"https://api.github.com/users/spring-projects/repos","events_url":"https://api.github.com/users/spring-projects/events{/privacy}","received_events_url":"https://api.github.com/users/spring-projects/received_events","type":"Organization","site_admin":false},"repo":{"id":26547822,"name":"spring-integration-kafka","full_name":"spring-projects/spring-integration-kafka","owner":{"login":"spring-projects","id":317776,"avatar_url":"https://avatars.githubusercontent.com/u/317776?v=3","gravatar_id":"","url":"https://api.github.com/users/spring-projects","html_url":"https://github.com/spring-projects","followers_url":"https://api.github.com/users/spring-projects/followers","following_url":"https://api.github.com/users/spring-projects/following{/other_user}","gists_url":"https://api.github.com/users/spring-projects/gists{/gist_id}","starred_url":"https://api.github.com/users/spring-projects/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/spring-projects/subscriptions","organizations_url":"https://api.github.com/users/spring-projects/orgs","repos_url":"https://api.github.com/users/spring-projects/repos","events_url":"https://api.github.com/users/spring-projects/events{/privacy}","received_events_url":"https://api.github.com/users/spring-projects/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/spring-projects/spring-integration-kafka","description":"","fork":false,"url":"https://api.github.com/repos/spring-projects/spring-integration-kafka","forks_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/forks","keys_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/teams","hooks_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/hooks","issue_events_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/events{/number}","events_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/events","assignees_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/assignees{/user}","branches_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/branches{/branch}","tags_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/tags","blobs_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/languages","stargazers_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/stargazers","contributors_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/contributors","subscribers_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/subscribers","subscription_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/subscription","commits_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/comments/{number}","contents_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/contents/{+path}","compare_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/merges","archive_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/downloads","issues_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues{/number}","pulls_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/labels{/name}","releases_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/releases{/id}","created_at":"2014-11-12T17:35:29Z","updated_at":"2015-01-10T22:53:37Z","pushed_at":"2015-01-08T20:04:20Z","git_url":"git://github.com/spring-projects/spring-integration-kafka.git","ssh_url":"git@github.com:spring-projects/spring-integration-kafka.git","clone_url":"https://github.com/spring-projects/spring-integration-kafka.git","svn_url":"https://github.com/spring-projects/spring-integration-kafka","homepage":null,"size":556,"stargazers_count":6,"watchers_count":6,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":6,"mirror_url":null,"open_issues_count":3,"forks":6,"open_issues":3,"watchers":6,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/spring-projects/spring-integration-kafka/pulls/12"},"html":{"href":"https://github.com/spring-projects/spring-integration-kafka/pull/12"},"issue":{"href":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/12"},"comments":{"href":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/12/comments"},"review_comments":{"href":"https://api.github.com/repos/spring-projects/spring-integration-kafka/pulls/12/comments"},"review_comment":{"href":"https://api.github.com/repos/spring-projects/spring-integration-kafka/pulls/comments/{number}"},"commits":{"href":"https://api.github.com/repos/spring-projects/spring-integration-kafka/pulls/12/commits"},"statuses":{"href":"https://api.github.com/repos/spring-projects/spring-integration-kafka/statuses/8f61fad6e535bb722c73613a4ecb36fc3e5ab20c"}}}}