{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/37147003","id":37147003,"diff_hunk":"@@ -0,0 +1,308 @@\n+/**\n+ * Licensed to the Apache Software Foundation (ASF) under one or more\n+ * contributor license agreements.  See the NOTICE file distributed with\n+ * this work for additional information regarding copyright ownership.\n+ * The ASF licenses this file to You under the Apache License, Version 2.0\n+ * (the \"License\"); you may not use this file except in compliance with\n+ * the License.  You may obtain a copy of the License at\n+ * <p>\n+ * http://www.apache.org/licenses/LICENSE-2.0\n+ * <p>\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+package org.apache.kafka.clients.processor.internals;\n+\n+import org.apache.kafka.clients.consumer.Consumer;\n+import org.apache.kafka.clients.consumer.ConsumerRebalanceCallback;\n+import org.apache.kafka.clients.consumer.KafkaConsumer;\n+import org.apache.kafka.clients.processor.PTopology;\n+import org.apache.kafka.clients.processor.ProcessorConfig;\n+import org.apache.kafka.clients.processor.ProcessorProperties;\n+import org.apache.kafka.clients.producer.KafkaProducer;\n+import org.apache.kafka.clients.producer.Producer;\n+import org.apache.kafka.common.KafkaException;\n+import org.apache.kafka.common.MetricName;\n+import org.apache.kafka.common.TopicPartition;\n+import org.apache.kafka.common.metrics.Metrics;\n+import org.apache.kafka.common.metrics.Sensor;\n+import org.apache.kafka.common.metrics.stats.Avg;\n+import org.apache.kafka.common.metrics.stats.Count;\n+import org.apache.kafka.common.metrics.stats.Max;\n+import org.apache.kafka.common.metrics.stats.Rate;\n+import org.apache.kafka.common.serialization.ByteArrayDeserializer;\n+import org.apache.kafka.common.serialization.ByteArraySerializer;\n+import org.apache.kafka.common.serialization.Serializer;\n+import org.apache.kafka.common.utils.SystemTime;\n+import org.apache.kafka.common.utils.Time;\n+import org.apache.kafka.common.utils.Utils;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+import java.io.File;\n+import java.util.ArrayList;\n+import java.util.Collection;\n+import java.util.HashMap;\n+import java.util.HashSet;\n+import java.util.Map;\n+\n+public class KStreamThread extends Thread {\n+\n+    private static final Logger log = LoggerFactory.getLogger(KStreamThread.class);\n+\n+    private final Class<? extends PTopology> topologyClass;\n+    private final ArrayList<StreamGroup> streamGroups = new ArrayList<>();\n+    private final Map<Integer, ProcessorContextImpl> kstreamContexts = new HashMap<>();\n+    private final IngestorImpl ingestor;\n+    private final RecordCollectorImpl collector;\n+    private final ProcessorProperties properties;\n+    private final ProcessorConfig config;\n+    private final Metrics metrics;\n+    private final KafkaStreamingMetrics streamingMetrics;\n+    private final Time time;\n+    private volatile boolean running;\n+    private long lastCommit;\n+    private long nextStateCleaning;\n+    private long recordsProcessed;\n+\n+    protected final ConsumerRebalanceCallback rebalanceCallback = new ConsumerRebalanceCallback() {\n+        @Override\n+        public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> assignment) {\n+            ingestor.init();\n+            addPartitions(assignment);\n+        }\n+\n+        @Override\n+        public void onPartitionsRevoked(Consumer<?, ?> consumer, Collection<TopicPartition> assignment) {\n+            commitAll(time.milliseconds());\n+            removePartitions();\n+            ingestor.clear();\n+        }\n+    };\n+\n+    @SuppressWarnings(\"unchecked\")\n+    public KStreamThread(Class<? extends PTopology> topologyClass, ProcessorProperties properties) throws Exception {\n+        super();\n+\n+        if (properties.timestampExtractor() == null)\n+            throw new NullPointerException(\"timestamp extractor is missing\");\n+\n+        this.metrics = new Metrics();\n+        this.config = new ProcessorConfig(properties.config());\n+        this.topologyClass = topologyClass;\n+\n+        this.properties = properties;\n+        this.streamingMetrics = new KafkaStreamingMetrics();\n+\n+        // build the topology without initialization to get the topics for consumer\n+        PTopology topology = topologyClass.getConstructor().newInstance();\n+        topology.build();\n+\n+        // create the producer and consumer clients\n+        Producer<byte[], byte[]> producer = new KafkaProducer<>(properties.config(), new ByteArraySerializer(), new ByteArraySerializer());\n+        this.collector = new RecordCollectorImpl(producer, (Serializer<Object>) properties.keySerializer(), (Serializer<Object>) properties.valueSerializer());\n+\n+        Consumer<byte[], byte[]> consumer = new KafkaConsumer<>(properties.config(), rebalanceCallback, new ByteArrayDeserializer(), new ByteArrayDeserializer());\n+        this.ingestor = new IngestorImpl(consumer, topology.topics());\n+\n+        this.running = true;\n+        this.lastCommit = 0;\n+        this.nextStateCleaning = Long.MAX_VALUE;\n+        this.recordsProcessed = 0;\n+        this.time = new SystemTime();\n+    }\n+\n+    /**\n+     * Execute the stream processors\n+     */\n+    @Override\n+    public synchronized void run() {\n+        log.info(\"Starting a kstream thread\");\n+        try {\n+            ingestor.open();\n+            runLoop();\n+        } catch (RuntimeException e) {\n+            log.error(\"Uncaught error during processing: \", e);\n+            throw e;\n+        } finally {\n+            shutdown();\n+        }\n+    }\n+\n+    private void shutdown() {\n+        log.info(\"Shutting down a kstream thread\");\n+        commitAll(time.milliseconds());\n+\n+        collector.close();\n+        ingestor.close();\n+        removePartitions();\n+        log.info(\"kstream thread shutdown complete\");\n+    }\n+\n+    /**\n+     * Shutdown this streaming thread.\n+     */\n+    public synchronized void close() {\n+        running = false;","path":"stream/src/main/java/org/apache/kafka/clients/processor/internals/KStreamThread.java","position":150,"original_position":150,"commit_id":"68d055c61c40cb7d6fa8a8db04420b2891daa1ce","original_commit_id":"68d055c61c40cb7d6fa8a8db04420b2891daa1ce","user":{"login":"rhauch","id":84732,"avatar_url":"https://avatars.githubusercontent.com/u/84732?v=3","gravatar_id":"","url":"https://api.github.com/users/rhauch","html_url":"https://github.com/rhauch","followers_url":"https://api.github.com/users/rhauch/followers","following_url":"https://api.github.com/users/rhauch/following{/other_user}","gists_url":"https://api.github.com/users/rhauch/gists{/gist_id}","starred_url":"https://api.github.com/users/rhauch/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rhauch/subscriptions","organizations_url":"https://api.github.com/users/rhauch/orgs","repos_url":"https://api.github.com/users/rhauch/repos","events_url":"https://api.github.com/users/rhauch/events{/privacy}","received_events_url":"https://api.github.com/users/rhauch/received_events","type":"User","site_admin":false},"body":"The `run()` and `close()` methods should not be synchronized. Because they are, then once `run()` is called it will block any other synchronized method, including `close()`, and because `run()` only completes when `close()` is called, `run()` will never complete. In other words, the thread will never stop.\r\n\r\nYou should be able to simply remove the `synchronized` keywords with the current code and maintain thread safety of the `running` volatile boolean field: the only method that reads that field is the private `stillRunning()` (called via private `runLoop()` which is called via public `run()`), while the only method that writes to the field is `close()`.","created_at":"2015-08-16T15:37:38Z","updated_at":"2015-08-16T15:37:38Z","html_url":"https://github.com/apache/kafka/pull/130#discussion_r37147003","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/130","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/37147003"},"html":{"href":"https://github.com/apache/kafka/pull/130#discussion_r37147003"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/130"}}},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/130","id":42085063,"html_url":"https://github.com/apache/kafka/pull/130","diff_url":"https://github.com/apache/kafka/pull/130.diff","patch_url":"https://github.com/apache/kafka/pull/130.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/130","number":130,"state":"open","locked":false,"title":"KIP-28: First patch","user":{"login":"guozhangwang","id":10729597,"avatar_url":"https://avatars.githubusercontent.com/u/10729597?v=3","gravatar_id":"","url":"https://api.github.com/users/guozhangwang","html_url":"https://github.com/guozhangwang","followers_url":"https://api.github.com/users/guozhangwang/followers","following_url":"https://api.github.com/users/guozhangwang/following{/other_user}","gists_url":"https://api.github.com/users/guozhangwang/gists{/gist_id}","starred_url":"https://api.github.com/users/guozhangwang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/guozhangwang/subscriptions","organizations_url":"https://api.github.com/users/guozhangwang/orgs","repos_url":"https://api.github.com/users/guozhangwang/repos","events_url":"https://api.github.com/users/guozhangwang/events{/privacy}","received_events_url":"https://api.github.com/users/guozhangwang/received_events","type":"User","site_admin":false},"body":"Some open questions collected so far on the first patch. Thanks @gwenshap @jkreps @junrao .\r\n\r\n* Topology API: requiring users to instantiate their own Topology class with the overridden build() function is a little awkward. Instead it would be great to let users explicitly build the topology in Main and pass it in as a class:\r\n\r\n```\r\n    Topology myTopology = new TopologyBuilder(defaultDeser)\r\n                                                 .addProcessor(\"my-processor\", MyProcessor.class, new Source(\"my-source\"))\r\n                                                 .addProcessor(\"my-other-processor\", MyOtherProcessor.class, \"my-processor\");\r\n    KafkaStreaming streaming = new KafkaStreaming(config, myTopology);\r\n   streaming.run();\r\n```\r\n\r\nSo the implementation of KStream.filter look instead like this:\r\n```\r\n    public KStream<K, V> filter(Predicate<K, V> predicate) {\r\n        KStreamFilter<K, V> filter = new KStreamFilter<>();\r\n        topology.addProcessor(KStreamFilter.class, new Configs(\"predicate\", predicate));\r\n        return this;\r\n    }\r\n```\r\nThe advantage is that the user code can now get rid of the whole Topology class with the builder. I think the order of execution for that API is quite unintuitive.\r\n\r\n* We can probably move the forward() function from Processor to ProcessorContext, and split ProcessorContext into two classes, one with all the function calls as commit / send / schedule / forward, and another with the metadata function calls as topic / partition / offset / timestamp.\r\n\r\n* Can we hide the Chooser interface from users? In other words, if users can specify the \"time\" on each fetched messages from Kafka, would a hard-coded MinTimestampMessageChooser be sufficient so that we can move TimestampTracker / RecordQueue / Chooser / RecordCollector / etc all to the internal folders?\r\n\r\n* Shall we split the o.a.k.clients into two folders, with o.a.k.clients.processor in stream? Or should we just remove o.a.k.clients.processor and make everything under o.a.k.stream? In addition, currently there is a cyclic dependency between that two, would better to break it in the end state.\r\n\r\n* Consider moving the external dependencies such as RocksDB into a separate jar? For example we can just include a kafka-stream-rocksdb.jar which includes the RocksDBKeyValueStore only, and later on when we deprecate / remove such implementations we can simply remove the jar itself.\r\n\r\n* The way the KeyValueStore is created seems a bit weird. Since this is part of the internal state managed by KafkaProcessorContext, it seems there should be an api to create the KeyValueStore from KafkaProcessorContext, instead of passing context to the constructor of KeyValueStore.\r\n\r\n* Merge ProcessorConfigs with ProcessorProperties.\r\n\r\n* We can potentially remove the processor argument in ProcessorContext.schedule().","created_at":"2015-08-11T01:45:03Z","updated_at":"2015-08-16T15:37:38Z","closed_at":null,"merged_at":null,"merge_commit_sha":"0f15c5896f438b81db91f3dab220c23423cf6975","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/130/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/130/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/130/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/68d055c61c40cb7d6fa8a8db04420b2891daa1ce","head":{"label":"confluentinc:streaming","ref":"streaming","sha":"68d055c61c40cb7d6fa8a8db04420b2891daa1ce","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":37555321,"name":"kafka","full_name":"confluentinc/kafka","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","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/confluentinc/kafka","forks_url":"https://api.github.com/repos/confluentinc/kafka/forks","keys_url":"https://api.github.com/repos/confluentinc/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/confluentinc/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/confluentinc/kafka/teams","hooks_url":"https://api.github.com/repos/confluentinc/kafka/hooks","issue_events_url":"https://api.github.com/repos/confluentinc/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/confluentinc/kafka/events","assignees_url":"https://api.github.com/repos/confluentinc/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/confluentinc/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/confluentinc/kafka/tags","blobs_url":"https://api.github.com/repos/confluentinc/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/confluentinc/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/confluentinc/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/confluentinc/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/confluentinc/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/confluentinc/kafka/languages","stargazers_url":"https://api.github.com/repos/confluentinc/kafka/stargazers","contributors_url":"https://api.github.com/repos/confluentinc/kafka/contributors","subscribers_url":"https://api.github.com/repos/confluentinc/kafka/subscribers","subscription_url":"https://api.github.com/repos/confluentinc/kafka/subscription","commits_url":"https://api.github.com/repos/confluentinc/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/confluentinc/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/confluentinc/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/confluentinc/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/confluentinc/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/confluentinc/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/confluentinc/kafka/merges","archive_url":"https://api.github.com/repos/confluentinc/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/confluentinc/kafka/downloads","issues_url":"https://api.github.com/repos/confluentinc/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/confluentinc/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/confluentinc/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/confluentinc/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/confluentinc/kafka/labels{/name}","releases_url":"https://api.github.com/repos/confluentinc/kafka/releases{/id}","created_at":"2015-06-16T20:48:28Z","updated_at":"2015-08-07T12:33:19Z","pushed_at":"2015-08-15T22:23:51Z","git_url":"git://github.com/confluentinc/kafka.git","ssh_url":"git@github.com:confluentinc/kafka.git","clone_url":"https://github.com/confluentinc/kafka.git","svn_url":"https://github.com/confluentinc/kafka","homepage":null,"size":22087,"stargazers_count":3,"watchers_count":3,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":2,"mirror_url":null,"open_issues_count":2,"forks":2,"open_issues":2,"watchers":3,"default_branch":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"1ed88f0eba860bce4be0269cdd4d38c2b57db7f7","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":2211243,"name":"kafka","full_name":"apache/kafka","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/kafka","description":"Mirror of Apache Kafka","fork":false,"url":"https://api.github.com/repos/apache/kafka","forks_url":"https://api.github.com/repos/apache/kafka/forks","keys_url":"https://api.github.com/repos/apache/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/kafka/teams","hooks_url":"https://api.github.com/repos/apache/kafka/hooks","issue_events_url":"https://api.github.com/repos/apache/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/apache/kafka/events","assignees_url":"https://api.github.com/repos/apache/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/apache/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/apache/kafka/tags","blobs_url":"https://api.github.com/repos/apache/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/kafka/languages","stargazers_url":"https://api.github.com/repos/apache/kafka/stargazers","contributors_url":"https://api.github.com/repos/apache/kafka/contributors","subscribers_url":"https://api.github.com/repos/apache/kafka/subscribers","subscription_url":"https://api.github.com/repos/apache/kafka/subscription","commits_url":"https://api.github.com/repos/apache/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/apache/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/kafka/merges","archive_url":"https://api.github.com/repos/apache/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/kafka/downloads","issues_url":"https://api.github.com/repos/apache/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/apache/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/kafka/labels{/name}","releases_url":"https://api.github.com/repos/apache/kafka/releases{/id}","created_at":"2011-08-15T18:06:16Z","updated_at":"2015-08-16T11:11:00Z","pushed_at":"2015-08-15T08:42:42Z","git_url":"git://github.com/apache/kafka.git","ssh_url":"git@github.com:apache/kafka.git","clone_url":"https://github.com/apache/kafka.git","svn_url":"https://github.com/apache/kafka","homepage":null,"size":29696,"stargazers_count":1801,"watchers_count":1801,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1064,"mirror_url":"git://git.apache.org/kafka.git","open_issues_count":54,"forks":1064,"open_issues":54,"watchers":1801,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/130"},"html":{"href":"https://github.com/apache/kafka/pull/130"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/130"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/130/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/130/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/130/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/68d055c61c40cb7d6fa8a8db04420b2891daa1ce"}}}}