{"action":"created","comment":{"url":"https://api.github.com/repos/confluentinc/kafka-rest/pulls/comments/26641506","id":26641506,"diff_hunk":"@@ -0,0 +1,233 @@\n+/**\n+ * Copyright 2015 Confluent Inc.\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+package io.confluent.kafkarest;\n+\n+import io.confluent.kafka.serializers.KafkaAvroDecoder;\n+import io.confluent.kafkarest.converters.AvroConverter;\n+import io.confluent.kafkarest.entities.AvroConsumerRecord;\n+import io.confluent.kafkarest.entities.BinaryConsumerRecord;\n+import io.confluent.kafkarest.entities.ConsumerRecord;\n+import io.confluent.kafkarest.entities.EmbeddedFormat;\n+import io.confluent.kafkarest.simpleconsumerspool.NaiveSimpleConsumerPool;\n+import io.confluent.kafkarest.simpleconsumerspool.SimpleConsumerPool;\n+import io.confluent.kafkarest.simpleconsumerspool.SizeLimitedSimpleConsumerPool;\n+import io.confluent.rest.exceptions.RestException;\n+import io.confluent.rest.exceptions.RestServerErrorException;\n+import kafka.api.PartitionFetchInfo;\n+import kafka.cluster.Broker;\n+import kafka.common.TopicAndPartition;\n+import kafka.consumer.ConsumerConfig;\n+import kafka.javaapi.FetchRequest;\n+import kafka.javaapi.FetchResponse;\n+import kafka.javaapi.message.ByteBufferMessageSet;\n+import kafka.message.MessageAndMetadata;\n+import kafka.message.MessageAndOffset;\n+import kafka.serializer.Decoder;\n+import kafka.serializer.DefaultDecoder;\n+import kafka.utils.VerifiableProperties;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+import javax.ws.rs.core.Response;\n+import java.util.*;\n+import java.util.concurrent.atomic.AtomicInteger;\n+\n+public class SimpleConsumerManager {\n+\n+  private static final Logger log = LoggerFactory.getLogger(SimpleConsumerManager.class);\n+\n+  private final int maxPoolSize;\n+  private final MetadataObserver mdObserver;\n+  private final SimpleConsumerFactory simpleConsumerFactory;\n+\n+  private final Map<Broker, SimpleConsumerPool> simpleConsumersPools;\n+\n+  private AtomicInteger correlationId = new AtomicInteger(0);\n+\n+  private final Decoder<Object> avroDecoder;\n+  private final Decoder<byte[]> binaryDecoder;\n+\n+  public SimpleConsumerManager(final KafkaRestConfig config,\n+                               final MetadataObserver mdObserver,\n+                               final SimpleConsumerFactory simpleConsumerFactory) {\n+\n+    this.mdObserver = mdObserver;\n+    this.simpleConsumerFactory = simpleConsumerFactory;\n+\n+    maxPoolSize = config.getInt(KafkaRestConfig.SIMPLE_CONSUMER_POOL_SIZE_CONFIG);\n+    simpleConsumersPools = new HashMap<Broker, SimpleConsumerPool>();\n+\n+    // Load decoders\n+    Properties props = new Properties();\n+    props.setProperty(\"schema.registry.url\",\n+        config.getString(KafkaRestConfig.SCHEMA_REGISTRY_URL_CONFIG));\n+    avroDecoder = new KafkaAvroDecoder(new VerifiableProperties(props));\n+\n+    binaryDecoder = new DefaultDecoder(new VerifiableProperties());\n+  }\n+\n+  private SimpleConsumerPool createSimpleConsumerPool() {\n+    if (maxPoolSize > 0) {\n+      return new SizeLimitedSimpleConsumerPool(maxPoolSize, simpleConsumerFactory);\n+    }\n+    return new NaiveSimpleConsumerPool(simpleConsumerFactory);\n+  }\n+\n+  private SimpleFetcher getSimpleFetcher(final Broker broker) {\n+    if (!simpleConsumersPools.containsKey(broker)) {\n+      simpleConsumersPools.put(broker, createSimpleConsumerPool());\n+    }\n+    final SimpleConsumerPool pool = simpleConsumersPools.get(broker);\n+    return pool.get(broker.host(), broker.port());\n+  }\n+\n+  public void consume(final String topicName,\n+                      final int partitionId,\n+                      long offset,\n+                      long count,\n+                      final EmbeddedFormat embeddedFormat,\n+                      final ConsumerManager.ReadCallback callback) {\n+\n+    List<ConsumerRecord> records = null;\n+    Exception exception = null;\n+    SimpleFetcher simpleFetcher = null;\n+\n+    try {\n+      final Broker broker = mdObserver.getLeader(topicName, partitionId);\n+      simpleFetcher = getSimpleFetcher(broker);\n+\n+      records = new ArrayList<ConsumerRecord>();\n+\n+      int fetchIterations = 0;\n+      while (count > 0) {\n+        fetchIterations++;\n+        log.debug(\"Simple consumer \" + simpleFetcher.clientId() +\n+            \": fetch \" + fetchIterations + \"; \" + count + \" messages remaining\");\n+\n+        final ByteBufferMessageSet messageAndOffsets =\n+            fetchRecords(topicName, partitionId, offset, simpleFetcher);\n+\n+        for (final MessageAndOffset messageAndOffset : messageAndOffsets) {\n+          records.add(createConsumerRecord(messageAndOffset, topicName, partitionId, embeddedFormat));\n+          count--;\n+          offset++;\n+\n+          // If all requested messages have already been fetched, we can break early\n+          if (count == 0) {\n+            break;\n+          }\n+        }\n+      }\n+\n+    } catch (RestException e) {\n+      exception = e;\n+    } finally {","path":"src/main/java/io/confluent/kafkarest/SimpleConsumerManager.java","position":137,"original_position":137,"commit_id":"dc14821bddcbd377c02166dcb594040902bc6feb","original_commit_id":"dc14821bddcbd377c02166dcb594040902bc6feb","user":{"login":"ewencp","id":47366,"avatar_url":"https://avatars.githubusercontent.com/u/47366?v=3","gravatar_id":"","url":"https://api.github.com/users/ewencp","html_url":"https://github.com/ewencp","followers_url":"https://api.github.com/users/ewencp/followers","following_url":"https://api.github.com/users/ewencp/following{/other_user}","gists_url":"https://api.github.com/users/ewencp/gists{/gist_id}","starred_url":"https://api.github.com/users/ewencp/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ewencp/subscriptions","organizations_url":"https://api.github.com/users/ewencp/orgs","repos_url":"https://api.github.com/users/ewencp/repos","events_url":"https://api.github.com/users/ewencp/events{/privacy}","received_events_url":"https://api.github.com/users/ewencp/received_events","type":"User","site_admin":false},"body":"What about other exceptions that can be thrown by the SimpleConsumer? Or SerializationException? SimpleConsumer internally catches all Throwables and will eventually rethrow if retries continue to fail.\r\n\r\nAs it is, an exception isn't set even if one was thrown and records is still empty (since `exception` is only set for RestExceptions). What we want to do is make sure we always convert to an exception that inherits from RestException because that guarantees it will be correctly handled, set the HTTP status, and return a JSON error payload with a more specific error information than the HTTP status provides and a (hopefully) useful message. Check out the translation that ConsumerManager.readTopic does in the callback for worker.readTopic for an example. There are probably ways we could refine this when there are specific errors we can detect (e.g. connectivity issues with the broker?), but the code there at least makes it so we should always have a response that is formatted to our standards.","created_at":"2015-03-18T06:47:02Z","updated_at":"2015-03-18T06:47:02Z","html_url":"https://github.com/confluentinc/kafka-rest/pull/66#discussion_r26641506","pull_request_url":"https://api.github.com/repos/confluentinc/kafka-rest/pulls/66","_links":{"self":{"href":"https://api.github.com/repos/confluentinc/kafka-rest/pulls/comments/26641506"},"html":{"href":"https://github.com/confluentinc/kafka-rest/pull/66#discussion_r26641506"},"pull_request":{"href":"https://api.github.com/repos/confluentinc/kafka-rest/pulls/66"}}},"pull_request":{"url":"https://api.github.com/repos/confluentinc/kafka-rest/pulls/66","id":31325159,"html_url":"https://github.com/confluentinc/kafka-rest/pull/66","diff_url":"https://github.com/confluentinc/kafka-rest/pull/66.diff","patch_url":"https://github.com/confluentinc/kafka-rest/pull/66.patch","issue_url":"https://api.github.com/repos/confluentinc/kafka-rest/issues/66","number":66,"state":"open","locked":false,"title":"Add support for SimpleConsumer-like access (Issue  #26)","user":{"login":"HenriJ","id":594657,"avatar_url":"https://avatars.githubusercontent.com/u/594657?v=3","gravatar_id":"","url":"https://api.github.com/users/HenriJ","html_url":"https://github.com/HenriJ","followers_url":"https://api.github.com/users/HenriJ/followers","following_url":"https://api.github.com/users/HenriJ/following{/other_user}","gists_url":"https://api.github.com/users/HenriJ/gists{/gist_id}","starred_url":"https://api.github.com/users/HenriJ/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/HenriJ/subscriptions","organizations_url":"https://api.github.com/users/HenriJ/orgs","repos_url":"https://api.github.com/users/HenriJ/repos","events_url":"https://api.github.com/users/HenriJ/events{/privacy}","received_events_url":"https://api.github.com/users/HenriJ/received_events","type":"User","site_admin":false},"body":"See issue #26 for previous discussion","created_at":"2015-03-17T11:07:03Z","updated_at":"2015-03-18T06:47:02Z","closed_at":null,"merged_at":null,"merge_commit_sha":"e63815a5f42d4b380ce232926cd0b019090921a6","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/confluentinc/kafka-rest/pulls/66/commits","review_comments_url":"https://api.github.com/repos/confluentinc/kafka-rest/pulls/66/comments","review_comment_url":"https://api.github.com/repos/confluentinc/kafka-rest/pulls/comments{/number}","comments_url":"https://api.github.com/repos/confluentinc/kafka-rest/issues/66/comments","statuses_url":"https://api.github.com/repos/confluentinc/kafka-rest/statuses/dc14821bddcbd377c02166dcb594040902bc6feb","head":{"label":"HenriJ:master","ref":"master","sha":"dc14821bddcbd377c02166dcb594040902bc6feb","user":{"login":"HenriJ","id":594657,"avatar_url":"https://avatars.githubusercontent.com/u/594657?v=3","gravatar_id":"","url":"https://api.github.com/users/HenriJ","html_url":"https://github.com/HenriJ","followers_url":"https://api.github.com/users/HenriJ/followers","following_url":"https://api.github.com/users/HenriJ/following{/other_user}","gists_url":"https://api.github.com/users/HenriJ/gists{/gist_id}","starred_url":"https://api.github.com/users/HenriJ/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/HenriJ/subscriptions","organizations_url":"https://api.github.com/users/HenriJ/orgs","repos_url":"https://api.github.com/users/HenriJ/repos","events_url":"https://api.github.com/users/HenriJ/events{/privacy}","received_events_url":"https://api.github.com/users/HenriJ/received_events","type":"User","site_admin":false},"repo":{"id":31652177,"name":"kafka-rest","full_name":"HenriJ/kafka-rest","owner":{"login":"HenriJ","id":594657,"avatar_url":"https://avatars.githubusercontent.com/u/594657?v=3","gravatar_id":"","url":"https://api.github.com/users/HenriJ","html_url":"https://github.com/HenriJ","followers_url":"https://api.github.com/users/HenriJ/followers","following_url":"https://api.github.com/users/HenriJ/following{/other_user}","gists_url":"https://api.github.com/users/HenriJ/gists{/gist_id}","starred_url":"https://api.github.com/users/HenriJ/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/HenriJ/subscriptions","organizations_url":"https://api.github.com/users/HenriJ/orgs","repos_url":"https://api.github.com/users/HenriJ/repos","events_url":"https://api.github.com/users/HenriJ/events{/privacy}","received_events_url":"https://api.github.com/users/HenriJ/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/HenriJ/kafka-rest","description":"REST Proxy for Kafka","fork":true,"url":"https://api.github.com/repos/HenriJ/kafka-rest","forks_url":"https://api.github.com/repos/HenriJ/kafka-rest/forks","keys_url":"https://api.github.com/repos/HenriJ/kafka-rest/keys{/key_id}","collaborators_url":"https://api.github.com/repos/HenriJ/kafka-rest/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/HenriJ/kafka-rest/teams","hooks_url":"https://api.github.com/repos/HenriJ/kafka-rest/hooks","issue_events_url":"https://api.github.com/repos/HenriJ/kafka-rest/issues/events{/number}","events_url":"https://api.github.com/repos/HenriJ/kafka-rest/events","assignees_url":"https://api.github.com/repos/HenriJ/kafka-rest/assignees{/user}","branches_url":"https://api.github.com/repos/HenriJ/kafka-rest/branches{/branch}","tags_url":"https://api.github.com/repos/HenriJ/kafka-rest/tags","blobs_url":"https://api.github.com/repos/HenriJ/kafka-rest/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/HenriJ/kafka-rest/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/HenriJ/kafka-rest/git/refs{/sha}","trees_url":"https://api.github.com/repos/HenriJ/kafka-rest/git/trees{/sha}","statuses_url":"https://api.github.com/repos/HenriJ/kafka-rest/statuses/{sha}","languages_url":"https://api.github.com/repos/HenriJ/kafka-rest/languages","stargazers_url":"https://api.github.com/repos/HenriJ/kafka-rest/stargazers","contributors_url":"https://api.github.com/repos/HenriJ/kafka-rest/contributors","subscribers_url":"https://api.github.com/repos/HenriJ/kafka-rest/subscribers","subscription_url":"https://api.github.com/repos/HenriJ/kafka-rest/subscription","commits_url":"https://api.github.com/repos/HenriJ/kafka-rest/commits{/sha}","git_commits_url":"https://api.github.com/repos/HenriJ/kafka-rest/git/commits{/sha}","comments_url":"https://api.github.com/repos/HenriJ/kafka-rest/comments{/number}","issue_comment_url":"https://api.github.com/repos/HenriJ/kafka-rest/issues/comments{/number}","contents_url":"https://api.github.com/repos/HenriJ/kafka-rest/contents/{+path}","compare_url":"https://api.github.com/repos/HenriJ/kafka-rest/compare/{base}...{head}","merges_url":"https://api.github.com/repos/HenriJ/kafka-rest/merges","archive_url":"https://api.github.com/repos/HenriJ/kafka-rest/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/HenriJ/kafka-rest/downloads","issues_url":"https://api.github.com/repos/HenriJ/kafka-rest/issues{/number}","pulls_url":"https://api.github.com/repos/HenriJ/kafka-rest/pulls{/number}","milestones_url":"https://api.github.com/repos/HenriJ/kafka-rest/milestones{/number}","notifications_url":"https://api.github.com/repos/HenriJ/kafka-rest/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/HenriJ/kafka-rest/labels{/name}","releases_url":"https://api.github.com/repos/HenriJ/kafka-rest/releases{/id}","created_at":"2015-03-04T10:51:31Z","updated_at":"2015-03-17T11:04:34Z","pushed_at":"2015-03-17T11:04:33Z","git_url":"git://github.com/HenriJ/kafka-rest.git","ssh_url":"git@github.com:HenriJ/kafka-rest.git","clone_url":"https://github.com/HenriJ/kafka-rest.git","svn_url":"https://github.com/HenriJ/kafka-rest","homepage":null,"size":749,"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":"confluentinc:master","ref":"master","sha":"7609dd3db4a80b9c5aecb69ff80fa208421ce2d1","user":{"login":"confluentinc","id":9439498,"avatar_url":"https://avatars.githubusercontent.com/u/9439498?v=3","gravatar_id":"","url":"https://api.github.com/users/confluentinc","html_url":"https://github.com/confluentinc","followers_url":"https://api.github.com/users/confluentinc/followers","following_url":"https://api.github.com/users/confluentinc/following{/other_user}","gists_url":"https://api.github.com/users/confluentinc/gists{/gist_id}","starred_url":"https://api.github.com/users/confluentinc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/confluentinc/subscriptions","organizations_url":"https://api.github.com/users/confluentinc/orgs","repos_url":"https://api.github.com/users/confluentinc/repos","events_url":"https://api.github.com/users/confluentinc/events{/privacy}","received_events_url":"https://api.github.com/users/confluentinc/received_events","type":"Organization","site_admin":false},"repo":{"id":26873147,"name":"kafka-rest","full_name":"confluentinc/kafka-rest","owner":{"login":"confluentinc","id":9439498,"avatar_url":"https://avatars.githubusercontent.com/u/9439498?v=3","gravatar_id":"","url":"https://api.github.com/users/confluentinc","html_url":"https://github.com/confluentinc","followers_url":"https://api.github.com/users/confluentinc/followers","following_url":"https://api.github.com/users/confluentinc/following{/other_user}","gists_url":"https://api.github.com/users/confluentinc/gists{/gist_id}","starred_url":"https://api.github.com/users/confluentinc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/confluentinc/subscriptions","organizations_url":"https://api.github.com/users/confluentinc/orgs","repos_url":"https://api.github.com/users/confluentinc/repos","events_url":"https://api.github.com/users/confluentinc/events{/privacy}","received_events_url":"https://api.github.com/users/confluentinc/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/confluentinc/kafka-rest","description":"REST Proxy for Kafka","fork":false,"url":"https://api.github.com/repos/confluentinc/kafka-rest","forks_url":"https://api.github.com/repos/confluentinc/kafka-rest/forks","keys_url":"https://api.github.com/repos/confluentinc/kafka-rest/keys{/key_id}","collaborators_url":"https://api.github.com/repos/confluentinc/kafka-rest/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/confluentinc/kafka-rest/teams","hooks_url":"https://api.github.com/repos/confluentinc/kafka-rest/hooks","issue_events_url":"https://api.github.com/repos/confluentinc/kafka-rest/issues/events{/number}","events_url":"https://api.github.com/repos/confluentinc/kafka-rest/events","assignees_url":"https://api.github.com/repos/confluentinc/kafka-rest/assignees{/user}","branches_url":"https://api.github.com/repos/confluentinc/kafka-rest/branches{/branch}","tags_url":"https://api.github.com/repos/confluentinc/kafka-rest/tags","blobs_url":"https://api.github.com/repos/confluentinc/kafka-rest/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/confluentinc/kafka-rest/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/confluentinc/kafka-rest/git/refs{/sha}","trees_url":"https://api.github.com/repos/confluentinc/kafka-rest/git/trees{/sha}","statuses_url":"https://api.github.com/repos/confluentinc/kafka-rest/statuses/{sha}","languages_url":"https://api.github.com/repos/confluentinc/kafka-rest/languages","stargazers_url":"https://api.github.com/repos/confluentinc/kafka-rest/stargazers","contributors_url":"https://api.github.com/repos/confluentinc/kafka-rest/contributors","subscribers_url":"https://api.github.com/repos/confluentinc/kafka-rest/subscribers","subscription_url":"https://api.github.com/repos/confluentinc/kafka-rest/subscription","commits_url":"https://api.github.com/repos/confluentinc/kafka-rest/commits{/sha}","git_commits_url":"https://api.github.com/repos/confluentinc/kafka-rest/git/commits{/sha}","comments_url":"https://api.github.com/repos/confluentinc/kafka-rest/comments{/number}","issue_comment_url":"https://api.github.com/repos/confluentinc/kafka-rest/issues/comments{/number}","contents_url":"https://api.github.com/repos/confluentinc/kafka-rest/contents/{+path}","compare_url":"https://api.github.com/repos/confluentinc/kafka-rest/compare/{base}...{head}","merges_url":"https://api.github.com/repos/confluentinc/kafka-rest/merges","archive_url":"https://api.github.com/repos/confluentinc/kafka-rest/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/confluentinc/kafka-rest/downloads","issues_url":"https://api.github.com/repos/confluentinc/kafka-rest/issues{/number}","pulls_url":"https://api.github.com/repos/confluentinc/kafka-rest/pulls{/number}","milestones_url":"https://api.github.com/repos/confluentinc/kafka-rest/milestones{/number}","notifications_url":"https://api.github.com/repos/confluentinc/kafka-rest/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/confluentinc/kafka-rest/labels{/name}","releases_url":"https://api.github.com/repos/confluentinc/kafka-rest/releases{/id}","created_at":"2014-11-19T17:34:24Z","updated_at":"2015-03-17T22:31:51Z","pushed_at":"2015-03-17T16:49:24Z","git_url":"git://github.com/confluentinc/kafka-rest.git","ssh_url":"git@github.com:confluentinc/kafka-rest.git","clone_url":"https://github.com/confluentinc/kafka-rest.git","svn_url":"https://github.com/confluentinc/kafka-rest","homepage":null,"size":1355,"stargazers_count":14,"watchers_count":14,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":4,"mirror_url":null,"open_issues_count":15,"forks":4,"open_issues":15,"watchers":14,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/confluentinc/kafka-rest/pulls/66"},"html":{"href":"https://github.com/confluentinc/kafka-rest/pull/66"},"issue":{"href":"https://api.github.com/repos/confluentinc/kafka-rest/issues/66"},"comments":{"href":"https://api.github.com/repos/confluentinc/kafka-rest/issues/66/comments"},"review_comments":{"href":"https://api.github.com/repos/confluentinc/kafka-rest/pulls/66/comments"},"review_comment":{"href":"https://api.github.com/repos/confluentinc/kafka-rest/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/confluentinc/kafka-rest/pulls/66/commits"},"statuses":{"href":"https://api.github.com/repos/confluentinc/kafka-rest/statuses/dc14821bddcbd377c02166dcb594040902bc6feb"}}}}