{"action":"created","comment":{"url":"https://api.github.com/repos/druid-io/druid/pulls/comments/60469308","id":60469308,"diff_hunk":"@@ -0,0 +1,326 @@\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+\n+package io.druid.query.lookup;\n+\n+import com.fasterxml.jackson.annotation.JacksonInject;\n+import com.fasterxml.jackson.annotation.JsonCreator;\n+import com.fasterxml.jackson.annotation.JsonProperty;\n+import com.fasterxml.jackson.annotation.JsonTypeName;\n+import com.google.common.base.Preconditions;\n+import com.google.common.primitives.Longs;\n+import com.google.common.util.concurrent.FutureCallback;\n+import com.google.common.util.concurrent.Futures;\n+import com.google.common.util.concurrent.ListenableFuture;\n+import com.google.common.util.concurrent.ListeningExecutorService;\n+import com.google.common.util.concurrent.MoreExecutors;\n+import com.metamx.common.IAE;\n+import com.metamx.common.ISE;\n+import com.metamx.common.StringUtils;\n+import com.metamx.common.logger.Logger;\n+import io.druid.concurrent.Execs;\n+import io.druid.query.extraction.MapLookupExtractor;\n+import io.druid.server.namespace.cache.NamespaceExtractionCacheManager;\n+import kafka.consumer.ConsumerConfig;\n+import kafka.consumer.ConsumerIterator;\n+import kafka.consumer.KafkaStream;\n+import kafka.consumer.Whitelist;\n+import kafka.javaapi.consumer.ConsumerConnector;\n+import kafka.message.MessageAndMetadata;\n+import kafka.serializer.Decoder;\n+\n+import javax.annotation.Nullable;\n+import javax.validation.constraints.NotNull;\n+import java.nio.ByteBuffer;\n+import java.util.List;\n+import java.util.Map;\n+import java.util.Properties;\n+import java.util.UUID;\n+import java.util.concurrent.CancellationException;\n+import java.util.concurrent.atomic.AtomicBoolean;\n+import java.util.concurrent.atomic.AtomicLong;\n+import java.util.concurrent.atomic.AtomicReference;\n+import java.util.regex.Pattern;\n+\n+@JsonTypeName(\"kafka\")\n+public class KafkaLookupExtractorFactory implements LookupExtractorFactory\n+{\n+  private static final Logger LOG = new Logger(KafkaLookupExtractorFactory.class);\n+  static final Decoder<String> DEFAULT_STRING_DECODER = new Decoder<String>()\n+  {\n+    @Override\n+    public String fromBytes(byte[] bytes)\n+    {\n+      return StringUtils.fromUtf8(bytes);\n+    }\n+  };\n+\n+  private final Object startStopLock = new Object();\n+  private final ListeningExecutorService executorService;\n+  private final AtomicLong doubleEventCount = new AtomicLong(0L);\n+  private final NamespaceExtractionCacheManager cacheManager;\n+  private final String factoryId = UUID.randomUUID().toString();\n+  private final AtomicReference<Map<String, String>> mapRef = new AtomicReference<>(null);\n+\n+  private AtomicBoolean started = new AtomicBoolean(false);\n+  private ListenableFuture<?> future = null;\n+  private ConsumerConnector consumerConnector;\n+\n+  @JsonProperty\n+  private final String kafkaTopic;\n+\n+  @JsonProperty\n+  private final Map<String, String> kafkaProperties;\n+\n+  @JsonCreator\n+  public KafkaLookupExtractorFactory(\n+      @JacksonInject NamespaceExtractionCacheManager cacheManager,\n+      @NotNull @JsonProperty(value = \"kafkaTopic\", required = true) final String kafkaTopic,\n+      @NotNull @JsonProperty(value = \"kafkaProperties\", required = true) final Map<String, String> kafkaProperties\n+  )\n+  {\n+    this.kafkaTopic = Preconditions.checkNotNull(kafkaTopic, \"kafkaTopic required\");\n+    this.kafkaProperties = Preconditions.checkNotNull(kafkaProperties, \"kafkaProperties required\");\n+    executorService = MoreExecutors.listeningDecorator(Execs.singleThreaded(\n+        \"kafka-factory-\" + kafkaTopic + \"-%s\",\n+        Thread.MIN_PRIORITY\n+    ));\n+    this.cacheManager = cacheManager;\n+  }\n+\n+  public String getKafkaTopic()\n+  {\n+    return kafkaTopic;\n+  }\n+\n+  public Map<String, String> getKafkaProperties()\n+  {\n+    return kafkaProperties;\n+  }\n+\n+  @Override\n+  public boolean start()\n+  {\n+    synchronized (startStopLock) {\n+      if (started.get()) {\n+        LOG.warn(\"Already started, not starting again\");\n+        return false;","path":"extensions-core/kafka-extraction-namespace/src/main/java/io/druid/query/lookup/KafkaLookupExtractorFactory.java","position":123,"original_position":123,"commit_id":"90f4e4d82ab0a3ec0291d97363d0f8b8da208369","original_commit_id":"90f4e4d82ab0a3ec0291d97363d0f8b8da208369","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":"```\r\n@return true if start successfully started the {@link LookupExtractor}\r\n```\r\n\r\nas per javadoc.","created_at":"2016-04-20T19:08:34Z","updated_at":"2016-04-20T19:08:34Z","html_url":"https://github.com/druid-io/druid/pull/2800#discussion_r60469308","pull_request_url":"https://api.github.com/repos/druid-io/druid/pulls/2800","_links":{"self":{"href":"https://api.github.com/repos/druid-io/druid/pulls/comments/60469308"},"html":{"href":"https://github.com/druid-io/druid/pull/2800#discussion_r60469308"},"pull_request":{"href":"https://api.github.com/repos/druid-io/druid/pulls/2800"}}},"pull_request":{"url":"https://api.github.com/repos/druid-io/druid/pulls/2800","id":65557749,"html_url":"https://github.com/druid-io/druid/pull/2800","diff_url":"https://github.com/druid-io/druid/pull/2800.diff","patch_url":"https://github.com/druid-io/druid/pull/2800.patch","issue_url":"https://api.github.com/repos/druid-io/druid/issues/2800","number":2800,"state":"open","locked":false,"title":"[QTL] Move kafka-extraction-namespace to the Lookup framework.","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":"The major changes to the PR are as follows:\r\n\r\nThe kafka extraction namespace stuff now works off the lookup extractor factory framework. It still uses the caching mechanisms in the main lookup extension, but uses some random UUID in the cache instead of trying to keep some sort of relationship to the lookup name (which is enforced by the Lookup framework in core).\r\n\r\nThis means the kafka-extraction-namespace must go through the main lookup framework.","created_at":"2016-04-07T00:33:32Z","updated_at":"2016-04-20T19:08:34Z","closed_at":null,"merged_at":null,"merge_commit_sha":"f42810d9b7a021c864cbbb682d75ef30faf15144","assignee":null,"milestone":{"url":"https://api.github.com/repos/druid-io/druid/milestones/18","html_url":"https://github.com/druid-io/druid/milestones/0.9.1","labels_url":"https://api.github.com/repos/druid-io/druid/milestones/18/labels","id":1526819,"number":18,"title":"0.9.1","description":"","creator":{"login":"fjy","id":428325,"avatar_url":"https://avatars.githubusercontent.com/u/428325?v=3","gravatar_id":"","url":"https://api.github.com/users/fjy","html_url":"https://github.com/fjy","followers_url":"https://api.github.com/users/fjy/followers","following_url":"https://api.github.com/users/fjy/following{/other_user}","gists_url":"https://api.github.com/users/fjy/gists{/gist_id}","starred_url":"https://api.github.com/users/fjy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/fjy/subscriptions","organizations_url":"https://api.github.com/users/fjy/orgs","repos_url":"https://api.github.com/users/fjy/repos","events_url":"https://api.github.com/users/fjy/events{/privacy}","received_events_url":"https://api.github.com/users/fjy/received_events","type":"User","site_admin":false},"open_issues":25,"closed_issues":183,"state":"open","created_at":"2016-01-20T19:04:22Z","updated_at":"2016-04-19T23:28:58Z","due_on":null,"closed_at":null},"commits_url":"https://api.github.com/repos/druid-io/druid/pulls/2800/commits","review_comments_url":"https://api.github.com/repos/druid-io/druid/pulls/2800/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/2800/comments","statuses_url":"https://api.github.com/repos/druid-io/druid/statuses/90f4e4d82ab0a3ec0291d97363d0f8b8da208369","head":{"label":"metamx:kafkaLookupOnly","ref":"kafkaLookupOnly","sha":"90f4e4d82ab0a3ec0291d97363d0f8b8da208369","user":{"login":"metamx","id":299329,"avatar_url":"https://avatars.githubusercontent.com/u/299329?v=3","gravatar_id":"","url":"https://api.github.com/users/metamx","html_url":"https://github.com/metamx","followers_url":"https://api.github.com/users/metamx/followers","following_url":"https://api.github.com/users/metamx/following{/other_user}","gists_url":"https://api.github.com/users/metamx/gists{/gist_id}","starred_url":"https://api.github.com/users/metamx/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/metamx/subscriptions","organizations_url":"https://api.github.com/users/metamx/orgs","repos_url":"https://api.github.com/users/metamx/repos","events_url":"https://api.github.com/users/metamx/events{/privacy}","received_events_url":"https://api.github.com/users/metamx/received_events","type":"Organization","site_admin":false},"repo":{"id":28248872,"name":"druid","full_name":"metamx/druid","owner":{"login":"metamx","id":299329,"avatar_url":"https://avatars.githubusercontent.com/u/299329?v=3","gravatar_id":"","url":"https://api.github.com/users/metamx","html_url":"https://github.com/metamx","followers_url":"https://api.github.com/users/metamx/followers","following_url":"https://api.github.com/users/metamx/following{/other_user}","gists_url":"https://api.github.com/users/metamx/gists{/gist_id}","starred_url":"https://api.github.com/users/metamx/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/metamx/subscriptions","organizations_url":"https://api.github.com/users/metamx/orgs","repos_url":"https://api.github.com/users/metamx/repos","events_url":"https://api.github.com/users/metamx/events{/privacy}","received_events_url":"https://api.github.com/users/metamx/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/metamx/druid","description":"Real²time Exploratory Analytics on Large Datasets","fork":true,"url":"https://api.github.com/repos/metamx/druid","forks_url":"https://api.github.com/repos/metamx/druid/forks","keys_url":"https://api.github.com/repos/metamx/druid/keys{/key_id}","collaborators_url":"https://api.github.com/repos/metamx/druid/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/metamx/druid/teams","hooks_url":"https://api.github.com/repos/metamx/druid/hooks","issue_events_url":"https://api.github.com/repos/metamx/druid/issues/events{/number}","events_url":"https://api.github.com/repos/metamx/druid/events","assignees_url":"https://api.github.com/repos/metamx/druid/assignees{/user}","branches_url":"https://api.github.com/repos/metamx/druid/branches{/branch}","tags_url":"https://api.github.com/repos/metamx/druid/tags","blobs_url":"https://api.github.com/repos/metamx/druid/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/metamx/druid/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/metamx/druid/git/refs{/sha}","trees_url":"https://api.github.com/repos/metamx/druid/git/trees{/sha}","statuses_url":"https://api.github.com/repos/metamx/druid/statuses/{sha}","languages_url":"https://api.github.com/repos/metamx/druid/languages","stargazers_url":"https://api.github.com/repos/metamx/druid/stargazers","contributors_url":"https://api.github.com/repos/metamx/druid/contributors","subscribers_url":"https://api.github.com/repos/metamx/druid/subscribers","subscription_url":"https://api.github.com/repos/metamx/druid/subscription","commits_url":"https://api.github.com/repos/metamx/druid/commits{/sha}","git_commits_url":"https://api.github.com/repos/metamx/druid/git/commits{/sha}","comments_url":"https://api.github.com/repos/metamx/druid/comments{/number}","issue_comment_url":"https://api.github.com/repos/metamx/druid/issues/comments{/number}","contents_url":"https://api.github.com/repos/metamx/druid/contents/{+path}","compare_url":"https://api.github.com/repos/metamx/druid/compare/{base}...{head}","merges_url":"https://api.github.com/repos/metamx/druid/merges","archive_url":"https://api.github.com/repos/metamx/druid/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/metamx/druid/downloads","issues_url":"https://api.github.com/repos/metamx/druid/issues{/number}","pulls_url":"https://api.github.com/repos/metamx/druid/pulls{/number}","milestones_url":"https://api.github.com/repos/metamx/druid/milestones{/number}","notifications_url":"https://api.github.com/repos/metamx/druid/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/metamx/druid/labels{/name}","releases_url":"https://api.github.com/repos/metamx/druid/releases{/id}","deployments_url":"https://api.github.com/repos/metamx/druid/deployments","created_at":"2014-12-19T22:30:25Z","updated_at":"2016-04-07T13:15:24Z","pushed_at":"2016-04-20T19:00:25Z","git_url":"git://github.com/metamx/druid.git","ssh_url":"git@github.com:metamx/druid.git","clone_url":"https://github.com/metamx/druid.git","svn_url":"https://github.com/metamx/druid","homepage":"http://druid.io","size":113014,"stargazers_count":60,"watchers_count":60,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":17,"mirror_url":null,"open_issues_count":1,"forks":17,"open_issues":1,"watchers":60,"default_branch":"master"}},"base":{"label":"druid-io:master","ref":"master","sha":"234125878a8561d258642d9179d7ca2cb6a7d822","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}","deployments_url":"https://api.github.com/repos/druid-io/druid/deployments","created_at":"2012-10-23T19:08:07Z","updated_at":"2016-04-20T15:11:42Z","pushed_at":"2016-04-20T19:00:25Z","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":129180,"stargazers_count":3086,"watchers_count":3086,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":674,"mirror_url":null,"open_issues_count":411,"forks":674,"open_issues":411,"watchers":3086,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/druid-io/druid/pulls/2800"},"html":{"href":"https://github.com/druid-io/druid/pull/2800"},"issue":{"href":"https://api.github.com/repos/druid-io/druid/issues/2800"},"comments":{"href":"https://api.github.com/repos/druid-io/druid/issues/2800/comments"},"review_comments":{"href":"https://api.github.com/repos/druid-io/druid/pulls/2800/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/2800/commits"},"statuses":{"href":"https://api.github.com/repos/druid-io/druid/statuses/90f4e4d82ab0a3ec0291d97363d0f8b8da208369"}}}}