{"action":"created","issue":{"url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/63","labels_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/63/labels{/name}","comments_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/63/comments","events_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/63/events","html_url":"https://github.com/spring-projects/spring-integration-kafka/issues/63","id":106441341,"number":63,"title":"Issue with ConsumerConfiguration.receive method.","user":{"login":"abhijith-prabhakar","id":2000158,"avatar_url":"https://avatars.githubusercontent.com/u/2000158?v=3","gravatar_id":"","url":"https://api.github.com/users/abhijith-prabhakar","html_url":"https://github.com/abhijith-prabhakar","followers_url":"https://api.github.com/users/abhijith-prabhakar/followers","following_url":"https://api.github.com/users/abhijith-prabhakar/following{/other_user}","gists_url":"https://api.github.com/users/abhijith-prabhakar/gists{/gist_id}","starred_url":"https://api.github.com/users/abhijith-prabhakar/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/abhijith-prabhakar/subscriptions","organizations_url":"https://api.github.com/users/abhijith-prabhakar/orgs","repos_url":"https://api.github.com/users/abhijith-prabhakar/repos","events_url":"https://api.github.com/users/abhijith-prabhakar/events{/privacy}","received_events_url":"https://api.github.com/users/abhijith-prabhakar/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":9,"created_at":"2015-09-14T22:30:25Z","updated_at":"2015-10-29T19:21:51Z","closed_at":null,"body":"Hi,\r\n\r\nI see an major issue on ConsumerConfiguration.recieve() method.  If maxMessages is set to a very high value and no of messages in kafka topic is less, then while loop in this method will keep running until maxMessages value is hit.  It will never exit until that many number of messages arrived.  For ex: I have set maxMessages value as 5000 and there are 500 messages put in that topic.  I never get control back since this while loop keeps looping.  The only way I could get back the control is to keep consumer time out very less.  Please let me know if I am missing something.\r\n\r\n```\r\npublic Map<String, Map<Integer, List<Object>>> receive() {\r\n\t\tcount = messageLeftOverTracker.getCurrentCount();\r\n\t\tfinal Object lock = new Object();\r\n\r\n\t\tfinal List<Callable<List<MessageAndMetadata<K, V>>>> tasks = new LinkedList<Callable<List<MessageAndMetadata<K, V>>>>();\r\n\r\n\t\tfor (final List<KafkaStream<K, V>> streams : createConsumerMessageStreams()) {\r\n\t\t\tfor (final KafkaStream<K, V> stream : streams) {\r\n\t\t\t\ttasks.add(new Callable<List<MessageAndMetadata<K, V>>>() {\r\n\t\t\t\t\t@Override\r\n\t\t\t\t\tpublic List<MessageAndMetadata<K, V>> call() throws Exception {\r\n\t\t\t\t\t\tfinal List<MessageAndMetadata<K, V>> rawMessages = new ArrayList<MessageAndMetadata<K, V>>();\r\n\t\t\t\t\t\ttry {\r\n\t\t\t\t\t\t\twhile (count < maxMessages) {\r\n\t\t\t\t\t\t\t\tfinal MessageAndMetadata<K, V> messageAndMetadata = stream.iterator().next();\r\n\t\t\t\t\t\t\t\tsynchronized (lock) {\r\n\t\t\t\t\t\t\t\t\tif (count < maxMessages) {\r\n\t\t\t\t\t\t\t\t\t\trawMessages.add(messageAndMetadata);\r\n\t\t\t\t\t\t\t\t\t\tcount++;\r\n\t\t\t\t\t\t\t\t\t}\r\n\t\t\t\t\t\t\t\t\telse {\r\n\t\t\t\t\t\t\t\t\t\tmessageLeftOverTracker.addMessageAndMetadata(messageAndMetadata);\r\n\t\t\t\t\t\t\t\t\t}\r\n\t\t\t\t\t\t\t\t}\r\n\t\t\t\t\t\t\t}\r\n\t\t\t\t\t\t}\r\n\t\t\t\t\t\tcatch (ConsumerTimeoutException cte) {\r\n\t\t\t\t\t\t\tLOGGER.debug(\"Consumer timed out\");\r\n\t\t\t\t\t\t}\r\n\t\t\t\t\t\treturn rawMessages;\r\n\t\t\t\t\t}\r\n\t\t\t\t});\r\n\t\t\t}\r\n\t\t}\r\n\t\treturn executeTasks(tasks);\r\n\t}\r\n```"},"comment":{"url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/comments/152293048","html_url":"https://github.com/spring-projects/spring-integration-kafka/issues/63#issuecomment-152293048","issue_url":"https://api.github.com/repos/spring-projects/spring-integration-kafka/issues/63","id":152293048,"user":{"login":"artembilan","id":1071590,"avatar_url":"https://avatars.githubusercontent.com/u/1071590?v=3","gravatar_id":"","url":"https://api.github.com/users/artembilan","html_url":"https://github.com/artembilan","followers_url":"https://api.github.com/users/artembilan/followers","following_url":"https://api.github.com/users/artembilan/following{/other_user}","gists_url":"https://api.github.com/users/artembilan/gists{/gist_id}","starred_url":"https://api.github.com/users/artembilan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/artembilan/subscriptions","organizations_url":"https://api.github.com/users/artembilan/orgs","repos_url":"https://api.github.com/users/artembilan/repos","events_url":"https://api.github.com/users/artembilan/events{/privacy}","received_events_url":"https://api.github.com/users/artembilan/received_events","type":"User","site_admin":false},"created_at":"2015-10-29T19:21:51Z","updated_at":"2015-10-29T19:21:51Z","body":"Yes, that's fine. Quoting some guy from the Internet:\r\n\r\n> Never mind – I went through the Kafka high level consumer source code, and I came to a conclusion that it is not possible to create high level consumers on per-partition basis like I asked above. To do that I must use SimpleConsumer. ... On the same time, the trade-off is I do not have control over from which partition each consumer retrieves data (only the SimpleConsumer API gives this level of control).\r\n\r\nAnd an answer for him from Gwen Shapira:\r\n\r\n> True. High level API doesn’t let you choose which partition each consumer reads."}}