{"action":"created","comment":{"url":"https://api.github.com/repos/druid-io/druid/pulls/comments/49364650","id":49364650,"diff_hunk":"@@ -0,0 +1,518 @@\n+/*\n+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one\n+ * or more contributor license agreements. See the NOTICE file\n+ * distributed with this work for additional information\n+ * regarding copyright ownership. Metamarkets licenses this file\n+ * to you under the Apache License, Version 2.0 (the\n+ * \"License\"); you may not use this file except in compliance\n+ * with the License. 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,\n+ * software distributed under the License is distributed on an\n+ * \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n+ * KIND, either express or implied. See the License for the\n+ * specific language governing permissions and limitations\n+ * under the License.\n+ */\n+package io.druid.firehose.rocketmq;\n+\n+import com.alibaba.rocketmq.client.Validators;\n+import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;\n+import com.alibaba.rocketmq.client.consumer.MessageQueueListener;\n+import com.alibaba.rocketmq.client.consumer.PullResult;\n+import com.alibaba.rocketmq.client.consumer.store.OffsetStore;\n+import com.alibaba.rocketmq.client.consumer.store.ReadOffsetType;\n+import com.alibaba.rocketmq.client.exception.MQBrokerException;\n+import com.alibaba.rocketmq.client.exception.MQClientException;\n+import com.alibaba.rocketmq.common.ServiceThread;\n+import com.alibaba.rocketmq.common.message.MessageExt;\n+import com.alibaba.rocketmq.common.message.MessageQueue;\n+import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;\n+import com.alibaba.rocketmq.remoting.exception.RemotingException;\n+import com.fasterxml.jackson.annotation.JsonCreator;\n+import com.fasterxml.jackson.annotation.JsonProperty;\n+import com.google.common.collect.Sets;\n+import com.metamx.common.parsers.ParseException;\n+import io.druid.data.input.ByteBufferInputRowParser;\n+import io.druid.data.input.Firehose;\n+import io.druid.data.input.FirehoseFactory;\n+import io.druid.data.input.InputRow;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+import java.io.IOException;\n+import java.nio.ByteBuffer;\n+import java.util.*;\n+import java.util.concurrent.ConcurrentHashMap;\n+import java.util.concurrent.ConcurrentSkipListSet;\n+import java.util.concurrent.CountDownLatch;\n+\n+public class RocketMQFirehoseFactory implements FirehoseFactory<ByteBufferInputRowParser> {\n+\n+    private static final Logger LOGGER = LoggerFactory.getLogger(RocketMQFirehoseFactory.class);\n+\n+    /**\n+     * Passed in configuration for consumer client.\n+     * This provides an approach to override default values defined in {@link com.alibaba.rocketmq.common.MixAll}.\n+     */\n+    @JsonProperty\n+    private final Properties consumerProps;\n+\n+    /**\n+     * Consumer group. It's required.\n+     */\n+    @JsonProperty(required = true)\n+    private final String consumerGroup;\n+\n+    /**\n+     * Topics to consume. It's required.\n+     */\n+    @JsonProperty(required = true)\n+    private final List<String> feed;\n+\n+    /**\n+     * Pull batch size. It's optional.\n+     */\n+    @JsonProperty\n+    private final String pullBatchSize;\n+\n+    /**\n+     * Store messages that are fetched from brokers but not yet delivered to druid via fire hose.\n+     */\n+    private ConcurrentHashMap<MessageQueue, ConcurrentSkipListSet<MessageExt>> messageQueueTreeSetMap = new ConcurrentHashMap<>();\n+\n+    /**\n+     * Store message consuming status.\n+     */\n+    private ConcurrentHashMap<MessageQueue, ConcurrentSkipListSet<Long>> windows = new ConcurrentHashMap<>();\n+\n+    /**\n+     * Default pull batch size.\n+     */\n+    private static final int DEFAULT_PULL_BATCH_SIZE = 32;\n+\n+    @JsonCreator\n+    public RocketMQFirehoseFactory(@JsonProperty(\"consumerProps\") Properties consumerProps,\n+                                   @JsonProperty(\"consumerGroup\") String consumerGroup,\n+                                   @JsonProperty(\"feed\") List<String> feed,\n+                                   @JsonProperty(\"pullBatchSize\") String pullBatchSize) {\n+        this.consumerProps = consumerProps;\n+        this.pullBatchSize = pullBatchSize;\n+        for (Map.Entry<Object, Object> configItem : this.consumerProps.entrySet()) {\n+            System.setProperty(configItem.getKey().toString(), configItem.getValue().toString());\n+        }\n+        this.consumerGroup = consumerGroup;\n+        this.feed = feed;\n+    }\n+\n+    /**\n+     * Check if there are locally pending messages to consume.\n+     * @return true if there are some; false otherwise.\n+     */\n+    private boolean hasMessagesPending() {\n+\n+        for (ConcurrentHashMap.Entry<MessageQueue, ConcurrentSkipListSet<MessageExt>> entry : messageQueueTreeSetMap.entrySet()) {\n+            if (!entry.getValue().isEmpty()) {\n+                return true;\n+            }\n+        }\n+\n+        return false;\n+    }\n+\n+    @Override\n+    public Firehose connect(ByteBufferInputRowParser byteBufferInputRowParser) throws IOException, ParseException {\n+\n+        Set<String> newDimExclus = Sets.union(\n+                byteBufferInputRowParser.getParseSpec().getDimensionsSpec().getDimensionExclusions(),\n+                Sets.newHashSet(\"feed\")\n+        );\n+\n+        final ByteBufferInputRowParser theParser = byteBufferInputRowParser.withParseSpec(\n+                byteBufferInputRowParser.getParseSpec()\n+                        .withDimensionsSpec(\n+                                byteBufferInputRowParser.getParseSpec()\n+                                        .getDimensionsSpec()\n+                                        .withDimensionExclusions(\n+                                                newDimExclus\n+                                        )\n+                        )\n+        );\n+\n+        /**\n+         * Topic-Queue mapping.\n+         */\n+        final ConcurrentHashMap<String, Set<MessageQueue>> topicQueueMap;\n+\n+        /**\n+         * Default Pull-style client for RocketMQ.\n+         */\n+        final DefaultMQPullConsumer defaultMQPullConsumer;\n+        final DruidPullMessageService pullMessageService;\n+\n+        messageQueueTreeSetMap.clear();\n+        windows.clear();\n+\n+        try {\n+            defaultMQPullConsumer = new DefaultMQPullConsumer(this.consumerGroup);\n+            defaultMQPullConsumer.setMessageModel(MessageModel.CLUSTERING);\n+            topicQueueMap = new ConcurrentHashMap<>();\n+\n+            pullMessageService = new DruidPullMessageService(defaultMQPullConsumer);\n+            for (String topic : feed) {\n+                Validators.checkTopic(topic);\n+                topicQueueMap.put(topic, defaultMQPullConsumer.fetchSubscribeMessageQueues(topic));\n+            }\n+            DruidMessageQueueListener druidMessageQueueListener =\n+                    new DruidMessageQueueListener(Sets.newHashSet(feed), topicQueueMap, defaultMQPullConsumer);\n+            defaultMQPullConsumer.setMessageQueueListener(druidMessageQueueListener);\n+            defaultMQPullConsumer.start();\n+            pullMessageService.start();\n+        } catch (MQClientException e) {\n+            LOGGER.error(\"Failed to start DefaultMQPullConsumer\", e);\n+            throw new IOException(\"Failed to start RocketMQ client\", e);\n+        }\n+\n+        return new Firehose() {\n+\n+            @Override\n+            public boolean hasMore() {\n+                boolean hasMore = false;\n+                DruidPullRequest earliestPullRequest = null;\n+\n+                for (Map.Entry<String, Set<MessageQueue>> entry : topicQueueMap.entrySet()) {\n+                    for (MessageQueue messageQueue : entry.getValue()) {\n+                        if (messageQueueTreeSetMap.keySet().contains(messageQueue)\n+                                && !messageQueueTreeSetMap.get(messageQueue).isEmpty()) {\n+                            hasMore = true;\n+                        } else {\n+                            try {\n+                                long offset = defaultMQPullConsumer.fetchConsumeOffset(messageQueue, false);\n+                                int batchSize = (null == pullBatchSize || pullBatchSize.isEmpty()) ?\n+                                        DEFAULT_PULL_BATCH_SIZE:Integer.parseInt(pullBatchSize);\n+\n+                                DruidPullRequest newPullRequest = new DruidPullRequest(messageQueue, null, offset,\n+                                        batchSize, !hasMessagesPending());\n+\n+                                // notify pull message service to pull messages from brokers.\n+                                pullMessageService.putRequest(newPullRequest);\n+\n+                                // set the earliest pull in case we need to block.\n+                                if (null == earliestPullRequest) {\n+                                    earliestPullRequest = newPullRequest;\n+                                }\n+                            } catch (MQClientException e) {\n+                                LOGGER.error(\"Failed to fetch consume offset for queue: {}\", entry.getKey());\n+                            }\n+                        }\n+                    }\n+                }\n+\n+                // Block only when there is no locally pending messages.\n+                if (!hasMore && null != earliestPullRequest) {\n+                    try {\n+                        earliestPullRequest.getCountDownLatch().await();\n+                        hasMore = true;\n+                    } catch (InterruptedException e) {\n+                        LOGGER.error(\"CountDownLatch await got interrupted\", e);\n+                    }\n+                }\n+                return hasMore;\n+            }\n+\n+            @Override\n+            public InputRow nextRow() {\n+                for (Map.Entry<MessageQueue, ConcurrentSkipListSet<MessageExt>> entry : messageQueueTreeSetMap.entrySet()) {\n+                    if (!entry.getValue().isEmpty()) {\n+                        MessageExt message = entry.getValue().pollFirst();\n+                        InputRow inputRow = theParser.parse(ByteBuffer.wrap(message.getBody()));\n+\n+                        if (!windows.keySet().contains(entry.getKey())) {\n+                            windows.put(entry.getKey(), new ConcurrentSkipListSet<Long>());\n+                        }\n+                        windows.get(entry.getKey()).add(message.getQueueOffset());\n+                        return inputRow;\n+                    }\n+                }\n+\n+                // should never happen.\n+                throw new RuntimeException(\"Unexpected Fatal Error! There should have been one row available.\");\n+            }\n+\n+            @Override\n+            public Runnable commit() {\n+                return new Runnable() {\n+                    @Override\n+                    public void run() {\n+                        OffsetStore offsetStore = defaultMQPullConsumer.getOffsetStore();\n+                        Set<MessageQueue> updated = new HashSet<>();\n+                        // calculate offsets according to consuming windows.\n+                        for (ConcurrentHashMap.Entry<MessageQueue, ConcurrentSkipListSet<Long>> entry : windows.entrySet()) {\n+                            while (!entry.getValue().isEmpty()) {\n+\n+                                long offset = offsetStore.readOffset(entry.getKey(), ReadOffsetType.MEMORY_FIRST_THEN_STORE);\n+                                if (offset + 1 > entry.getValue().first()) {\n+                                    entry.getValue().pollFirst();\n+                                } else if (offset + 1 == entry.getValue().first()) {\n+                                    entry.getValue().pollFirst();\n+                                    offsetStore.updateOffset(entry.getKey(), offset + 1, true);\n+                                    updated.add(entry.getKey());\n+                                } else {\n+                                    break;\n+                                }\n+\n+                            }\n+                        }\n+                        offsetStore.persistAll(updated);\n+                    }\n+                };\n+            }\n+\n+            @Override\n+            public void close() throws IOException {\n+                defaultMQPullConsumer.shutdown();\n+                pullMessageService.shutdown(false);\n+            }\n+        };\n+    }\n+\n+\n+    /**\n+     * Pull request.\n+     */\n+    final class DruidPullRequest {\n+        private final MessageQueue messageQueue;\n+        private final String tag;\n+        private final long nextBeginOffset;\n+        private final int pullBatchSize;\n+        private final boolean longPull;\n+        private final CountDownLatch countDownLatch;\n+\n+        public DruidPullRequest(final MessageQueue messageQueue,\n+                                final String tag,\n+                                final long nextBeginOffset,\n+                                final int pullBatchSize,\n+                                final boolean useLongPull) {\n+            this.messageQueue = messageQueue;\n+            this.tag = (null == tag ? \"*\" : tag);\n+            this.nextBeginOffset = nextBeginOffset;\n+            this.pullBatchSize = pullBatchSize;\n+            this.longPull = useLongPull;\n+            countDownLatch  = new CountDownLatch(1);\n+        }\n+\n+        public MessageQueue getMessageQueue() {\n+            return messageQueue;\n+        }\n+\n+        public long getNextBeginOffset() {\n+            return nextBeginOffset;\n+        }\n+\n+        public String getTag() {\n+            return tag;\n+        }\n+\n+        public int getPullBatchSize() {\n+            return pullBatchSize;\n+        }\n+\n+        public boolean isLongPull() {\n+            return longPull;\n+        }\n+\n+        public CountDownLatch getCountDownLatch() {\n+            return countDownLatch;\n+        }\n+    }\n+\n+\n+    /**\n+     * Pull message service for druid.\n+     *\n+     * <strong>Note: this is a single thread service.</strong>\n+     */\n+    final class DruidPullMessageService extends ServiceThread {\n+\n+        private volatile List<DruidPullRequest> requestsWrite = new ArrayList<>();\n+        private volatile List<DruidPullRequest> requestsRead = new ArrayList<>();\n+\n+        private final DefaultMQPullConsumer defaultMQPullConsumer;\n+\n+        public DruidPullMessageService(final DefaultMQPullConsumer defaultMQPullConsumer) {\n+            this.defaultMQPullConsumer = defaultMQPullConsumer;\n+        }\n+\n+        public void putRequest(final DruidPullRequest request) {\n+            synchronized (this) {\n+                this.requestsWrite.add(request);\n+                if (!hasNotified) {\n+                    hasNotified = true;\n+                    notify();\n+                }\n+            }\n+        }\n+\n+        private void swapRequests() {\n+            List<DruidPullRequest> tmp = requestsWrite;\n+            requestsWrite = requestsRead;\n+            requestsRead = tmp;\n+        }\n+\n+        @Override\n+        public String getServiceName() {\n+            return getClass().getSimpleName();\n+        }\n+\n+        /**\n+         * Core message pulling logic code goes here.\n+         */\n+        private void doPull() {\n+            for (DruidPullRequest pullRequest : requestsRead) {\n+                PullResult pullResult;\n+                try {\n+                    if (!pullRequest.isLongPull()) {\n+                        pullResult = defaultMQPullConsumer.pull(\n+                                pullRequest.getMessageQueue(),\n+                                pullRequest.getTag(),\n+                                pullRequest.getNextBeginOffset(),\n+                                pullRequest.getPullBatchSize());\n+                    } else {\n+                        pullResult = defaultMQPullConsumer.pullBlockIfNotFound(\n+                                pullRequest.getMessageQueue(),\n+                                pullRequest.getTag(),\n+                                pullRequest.getNextBeginOffset(),\n+                                pullRequest.getPullBatchSize()\n+                        );\n+                    }\n+\n+                    switch(pullResult.getPullStatus()) {\n+                        case FOUND:\n+                            // Handle pull result.\n+                            if (!messageQueueTreeSetMap.keySet().contains(pullRequest.getMessageQueue())) {\n+                                messageQueueTreeSetMap.putIfAbsent(pullRequest.getMessageQueue(),\n+                                        new ConcurrentSkipListSet<>(new MessageComparator()));\n+                            }\n+                            messageQueueTreeSetMap.get(pullRequest.getMessageQueue()).addAll(pullResult.getMsgFoundList());\n+                            break;\n+\n+                        case NO_NEW_MSG:\n+                        case NO_MATCHED_MSG:\n+                            break;\n+\n+                        case OFFSET_ILLEGAL:\n+                            LOGGER.error(\"Bad Pull Request: Offset is illegal. Offset used: {}\",\n+                                    pullRequest.getNextBeginOffset());\n+                            break;\n+\n+                        default:\n+                            break;\n+                    }\n+                } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {\n+                    LOGGER.error(\"Failed to pull message from broker.\", e);\n+                } finally {\n+                    pullRequest.getCountDownLatch().countDown();\n+                }\n+\n+            }\n+            requestsRead.clear();\n+        }\n+\n+        /**\n+         * Thread looping entry.\n+         */\n+        @Override\n+        public void run() {\n+            LOGGER.info(getServiceName() + \" starts.\");\n+            while (!isStoped()) {\n+                waitForRunning(0);\n+                doPull();\n+            }\n+\n+            // in case this service is shutdown gracefully without interruption.","path":"extensions/druid-rocketmq/src/main/java/io/druid/firehose/rocketmq/RocketMQFirehoseFactory.java","position":434,"original_position":434,"commit_id":"abe134bef699c0f301dfbf862cca57eb8c7ba59a","original_commit_id":"abe134bef699c0f301dfbf862cca57eb8c7ba59a","user":{"login":"drcrallen","id":8213081,"avatar_url":"https://avatars.githubusercontent.com/u/8213081?v=3","gravatar_id":"","url":"https://api.github.com/users/drcrallen","html_url":"https://github.com/drcrallen","followers_url":"https://api.github.com/users/drcrallen/followers","following_url":"https://api.github.com/users/drcrallen/following{/other_user}","gists_url":"https://api.github.com/users/drcrallen/gists{/gist_id}","starred_url":"https://api.github.com/users/drcrallen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/drcrallen/subscriptions","organizations_url":"https://api.github.com/users/drcrallen/orgs","repos_url":"https://api.github.com/users/drcrallen/repos","events_url":"https://api.github.com/users/drcrallen/events{/privacy}","received_events_url":"https://api.github.com/users/drcrallen/received_events","type":"User","site_admin":false},"body":"Why is this needed?","created_at":"2016-01-11T19:18:23Z","updated_at":"2016-01-11T19:18:23Z","html_url":"https://github.com/druid-io/druid/pull/2231#discussion_r49364650","pull_request_url":"https://api.github.com/repos/druid-io/druid/pulls/2231","_links":{"self":{"href":"https://api.github.com/repos/druid-io/druid/pulls/comments/49364650"},"html":{"href":"https://github.com/druid-io/druid/pull/2231#discussion_r49364650"},"pull_request":{"href":"https://api.github.com/repos/druid-io/druid/pulls/2231"}}},"pull_request":{"url":"https://api.github.com/repos/druid-io/druid/pulls/2231","id":55409799,"html_url":"https://github.com/druid-io/druid/pull/2231","diff_url":"https://github.com/druid-io/druid/pull/2231.diff","patch_url":"https://github.com/druid-io/druid/pull/2231.patch","issue_url":"https://api.github.com/repos/druid-io/druid/issues/2231","number":2231,"state":"open","locked":false,"title":"Add druid-rocketmq module","user":{"login":"lizhanhui","id":641819,"avatar_url":"https://avatars.githubusercontent.com/u/641819?v=3","gravatar_id":"","url":"https://api.github.com/users/lizhanhui","html_url":"https://github.com/lizhanhui","followers_url":"https://api.github.com/users/lizhanhui/followers","following_url":"https://api.github.com/users/lizhanhui/following{/other_user}","gists_url":"https://api.github.com/users/lizhanhui/gists{/gist_id}","starred_url":"https://api.github.com/users/lizhanhui/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/lizhanhui/subscriptions","organizations_url":"https://api.github.com/users/lizhanhui/orgs","repos_url":"https://api.github.com/users/lizhanhui/repos","events_url":"https://api.github.com/users/lizhanhui/events{/privacy}","received_events_url":"https://api.github.com/users/lizhanhui/received_events","type":"User","site_admin":false},"body":"Re-base on top of lastest master branch of druid-io repo, aka, d0b10c29d7542249c804a3f8d7f3b9f178981b08","created_at":"2016-01-08T00:17:22Z","updated_at":"2016-01-11T19:18:23Z","closed_at":null,"merged_at":null,"merge_commit_sha":"f3d8aa1639aa431f750a4f8aa0427b26b28fc736","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/druid-io/druid/pulls/2231/commits","review_comments_url":"https://api.github.com/repos/druid-io/druid/pulls/2231/comments","review_comment_url":"https://api.github.com/repos/druid-io/druid/pulls/comments{/number}","comments_url":"https://api.github.com/repos/druid-io/druid/issues/2231/comments","statuses_url":"https://api.github.com/repos/druid-io/druid/statuses/abe134bef699c0f301dfbf862cca57eb8c7ba59a","head":{"label":"lizhanhui:pull_request","ref":"pull_request","sha":"abe134bef699c0f301dfbf862cca57eb8c7ba59a","user":{"login":"lizhanhui","id":641819,"avatar_url":"https://avatars.githubusercontent.com/u/641819?v=3","gravatar_id":"","url":"https://api.github.com/users/lizhanhui","html_url":"https://github.com/lizhanhui","followers_url":"https://api.github.com/users/lizhanhui/followers","following_url":"https://api.github.com/users/lizhanhui/following{/other_user}","gists_url":"https://api.github.com/users/lizhanhui/gists{/gist_id}","starred_url":"https://api.github.com/users/lizhanhui/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/lizhanhui/subscriptions","organizations_url":"https://api.github.com/users/lizhanhui/orgs","repos_url":"https://api.github.com/users/lizhanhui/repos","events_url":"https://api.github.com/users/lizhanhui/events{/privacy}","received_events_url":"https://api.github.com/users/lizhanhui/received_events","type":"User","site_admin":false},"repo":{"id":26576350,"name":"data_druid","full_name":"lizhanhui/data_druid","owner":{"login":"lizhanhui","id":641819,"avatar_url":"https://avatars.githubusercontent.com/u/641819?v=3","gravatar_id":"","url":"https://api.github.com/users/lizhanhui","html_url":"https://github.com/lizhanhui","followers_url":"https://api.github.com/users/lizhanhui/followers","following_url":"https://api.github.com/users/lizhanhui/following{/other_user}","gists_url":"https://api.github.com/users/lizhanhui/gists{/gist_id}","starred_url":"https://api.github.com/users/lizhanhui/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/lizhanhui/subscriptions","organizations_url":"https://api.github.com/users/lizhanhui/orgs","repos_url":"https://api.github.com/users/lizhanhui/repos","events_url":"https://api.github.com/users/lizhanhui/events{/privacy}","received_events_url":"https://api.github.com/users/lizhanhui/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/lizhanhui/data_druid","description":"Real²time Exploratory Analytics on Large Datasets","fork":true,"url":"https://api.github.com/repos/lizhanhui/data_druid","forks_url":"https://api.github.com/repos/lizhanhui/data_druid/forks","keys_url":"https://api.github.com/repos/lizhanhui/data_druid/keys{/key_id}","collaborators_url":"https://api.github.com/repos/lizhanhui/data_druid/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/lizhanhui/data_druid/teams","hooks_url":"https://api.github.com/repos/lizhanhui/data_druid/hooks","issue_events_url":"https://api.github.com/repos/lizhanhui/data_druid/issues/events{/number}","events_url":"https://api.github.com/repos/lizhanhui/data_druid/events","assignees_url":"https://api.github.com/repos/lizhanhui/data_druid/assignees{/user}","branches_url":"https://api.github.com/repos/lizhanhui/data_druid/branches{/branch}","tags_url":"https://api.github.com/repos/lizhanhui/data_druid/tags","blobs_url":"https://api.github.com/repos/lizhanhui/data_druid/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/lizhanhui/data_druid/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/lizhanhui/data_druid/git/refs{/sha}","trees_url":"https://api.github.com/repos/lizhanhui/data_druid/git/trees{/sha}","statuses_url":"https://api.github.com/repos/lizhanhui/data_druid/statuses/{sha}","languages_url":"https://api.github.com/repos/lizhanhui/data_druid/languages","stargazers_url":"https://api.github.com/repos/lizhanhui/data_druid/stargazers","contributors_url":"https://api.github.com/repos/lizhanhui/data_druid/contributors","subscribers_url":"https://api.github.com/repos/lizhanhui/data_druid/subscribers","subscription_url":"https://api.github.com/repos/lizhanhui/data_druid/subscription","commits_url":"https://api.github.com/repos/lizhanhui/data_druid/commits{/sha}","git_commits_url":"https://api.github.com/repos/lizhanhui/data_druid/git/commits{/sha}","comments_url":"https://api.github.com/repos/lizhanhui/data_druid/comments{/number}","issue_comment_url":"https://api.github.com/repos/lizhanhui/data_druid/issues/comments{/number}","contents_url":"https://api.github.com/repos/lizhanhui/data_druid/contents/{+path}","compare_url":"https://api.github.com/repos/lizhanhui/data_druid/compare/{base}...{head}","merges_url":"https://api.github.com/repos/lizhanhui/data_druid/merges","archive_url":"https://api.github.com/repos/lizhanhui/data_druid/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/lizhanhui/data_druid/downloads","issues_url":"https://api.github.com/repos/lizhanhui/data_druid/issues{/number}","pulls_url":"https://api.github.com/repos/lizhanhui/data_druid/pulls{/number}","milestones_url":"https://api.github.com/repos/lizhanhui/data_druid/milestones{/number}","notifications_url":"https://api.github.com/repos/lizhanhui/data_druid/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/lizhanhui/data_druid/labels{/name}","releases_url":"https://api.github.com/repos/lizhanhui/data_druid/releases{/id}","created_at":"2014-11-13T07:33:52Z","updated_at":"2014-11-17T03:21:39Z","pushed_at":"2016-01-11T03:50:34Z","git_url":"git://github.com/lizhanhui/data_druid.git","ssh_url":"git@github.com:lizhanhui/data_druid.git","clone_url":"https://github.com/lizhanhui/data_druid.git","svn_url":"https://github.com/lizhanhui/data_druid","homepage":"http://druid.io","size":98972,"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":"druid-io:master","ref":"master","sha":"d0b10c29d7542249c804a3f8d7f3b9f178981b08","user":{"login":"druid-io","id":4203474,"avatar_url":"https://avatars.githubusercontent.com/u/4203474?v=3","gravatar_id":"","url":"https://api.github.com/users/druid-io","html_url":"https://github.com/druid-io","followers_url":"https://api.github.com/users/druid-io/followers","following_url":"https://api.github.com/users/druid-io/following{/other_user}","gists_url":"https://api.github.com/users/druid-io/gists{/gist_id}","starred_url":"https://api.github.com/users/druid-io/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/druid-io/subscriptions","organizations_url":"https://api.github.com/users/druid-io/orgs","repos_url":"https://api.github.com/users/druid-io/repos","events_url":"https://api.github.com/users/druid-io/events{/privacy}","received_events_url":"https://api.github.com/users/druid-io/received_events","type":"Organization","site_admin":false},"repo":{"id":6358188,"name":"druid","full_name":"druid-io/druid","owner":{"login":"druid-io","id":4203474,"avatar_url":"https://avatars.githubusercontent.com/u/4203474?v=3","gravatar_id":"","url":"https://api.github.com/users/druid-io","html_url":"https://github.com/druid-io","followers_url":"https://api.github.com/users/druid-io/followers","following_url":"https://api.github.com/users/druid-io/following{/other_user}","gists_url":"https://api.github.com/users/druid-io/gists{/gist_id}","starred_url":"https://api.github.com/users/druid-io/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/druid-io/subscriptions","organizations_url":"https://api.github.com/users/druid-io/orgs","repos_url":"https://api.github.com/users/druid-io/repos","events_url":"https://api.github.com/users/druid-io/events{/privacy}","received_events_url":"https://api.github.com/users/druid-io/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/druid-io/druid","description":"Column oriented distributed data store ideal for powering interactive applications","fork":false,"url":"https://api.github.com/repos/druid-io/druid","forks_url":"https://api.github.com/repos/druid-io/druid/forks","keys_url":"https://api.github.com/repos/druid-io/druid/keys{/key_id}","collaborators_url":"https://api.github.com/repos/druid-io/druid/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/druid-io/druid/teams","hooks_url":"https://api.github.com/repos/druid-io/druid/hooks","issue_events_url":"https://api.github.com/repos/druid-io/druid/issues/events{/number}","events_url":"https://api.github.com/repos/druid-io/druid/events","assignees_url":"https://api.github.com/repos/druid-io/druid/assignees{/user}","branches_url":"https://api.github.com/repos/druid-io/druid/branches{/branch}","tags_url":"https://api.github.com/repos/druid-io/druid/tags","blobs_url":"https://api.github.com/repos/druid-io/druid/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/druid-io/druid/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/druid-io/druid/git/refs{/sha}","trees_url":"https://api.github.com/repos/druid-io/druid/git/trees{/sha}","statuses_url":"https://api.github.com/repos/druid-io/druid/statuses/{sha}","languages_url":"https://api.github.com/repos/druid-io/druid/languages","stargazers_url":"https://api.github.com/repos/druid-io/druid/stargazers","contributors_url":"https://api.github.com/repos/druid-io/druid/contributors","subscribers_url":"https://api.github.com/repos/druid-io/druid/subscribers","subscription_url":"https://api.github.com/repos/druid-io/druid/subscription","commits_url":"https://api.github.com/repos/druid-io/druid/commits{/sha}","git_commits_url":"https://api.github.com/repos/druid-io/druid/git/commits{/sha}","comments_url":"https://api.github.com/repos/druid-io/druid/comments{/number}","issue_comment_url":"https://api.github.com/repos/druid-io/druid/issues/comments{/number}","contents_url":"https://api.github.com/repos/druid-io/druid/contents/{+path}","compare_url":"https://api.github.com/repos/druid-io/druid/compare/{base}...{head}","merges_url":"https://api.github.com/repos/druid-io/druid/merges","archive_url":"https://api.github.com/repos/druid-io/druid/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/druid-io/druid/downloads","issues_url":"https://api.github.com/repos/druid-io/druid/issues{/number}","pulls_url":"https://api.github.com/repos/druid-io/druid/pulls{/number}","milestones_url":"https://api.github.com/repos/druid-io/druid/milestones{/number}","notifications_url":"https://api.github.com/repos/druid-io/druid/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/druid-io/druid/labels{/name}","releases_url":"https://api.github.com/repos/druid-io/druid/releases{/id}","created_at":"2012-10-23T19:08:07Z","updated_at":"2016-01-11T17:27:38Z","pushed_at":"2016-01-11T18:56:57Z","git_url":"git://github.com/druid-io/druid.git","ssh_url":"git@github.com:druid-io/druid.git","clone_url":"https://github.com/druid-io/druid.git","svn_url":"https://github.com/druid-io/druid","homepage":"http://druid.io","size":121894,"stargazers_count":2347,"watchers_count":2347,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":550,"mirror_url":null,"open_issues_count":280,"forks":550,"open_issues":280,"watchers":2347,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/druid-io/druid/pulls/2231"},"html":{"href":"https://github.com/druid-io/druid/pull/2231"},"issue":{"href":"https://api.github.com/repos/druid-io/druid/issues/2231"},"comments":{"href":"https://api.github.com/repos/druid-io/druid/issues/2231/comments"},"review_comments":{"href":"https://api.github.com/repos/druid-io/druid/pulls/2231/comments"},"review_comment":{"href":"https://api.github.com/repos/druid-io/druid/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/druid-io/druid/pulls/2231/commits"},"statuses":{"href":"https://api.github.com/repos/druid-io/druid/statuses/abe134bef699c0f301dfbf862cca57eb8c7ba59a"}}}}