{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/205549444","pull_request_review_id":140842723,"id":205549444,"node_id":"MDI0OlB1bGxSZXF1ZXN0UmV2aWV3Q29tbWVudDIwNTU0OTQ0NA==","diff_hunk":"@@ -0,0 +1,361 @@\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+ *\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+\n+package kafka.tools\n+\n+import java.io._\n+import java.nio.ByteBuffer\n+import java.nio.charset.StandardCharsets.UTF_8\n+import java.time.Duration\n+import java.util.{Properties, Random}\n+\n+import joptsimple.OptionParser\n+import kafka.log.Log\n+import kafka.utils._\n+import org.apache.kafka.clients.admin.NewTopic\n+import org.apache.kafka.clients.{CommonClientConfigs, admin}\n+import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}\n+import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}\n+import org.apache.kafka.common.config.TopicConfig\n+import org.apache.kafka.common.record.FileRecords\n+import org.apache.kafka.common.utils.Utils\n+\n+import scala.collection.JavaConverters._\n+\n+/**\n+ * This is a torture test that runs against an existing broker\n+ *\n+ * Here is how it works:\n+ *\n+ * It produces a series of specially formatted messages to one or more partitions. Each message it produces\n+ * it logs out to a text file. The messages have a limited set of keys, so there is duplication in the key space.\n+ *\n+ * The broker will clean its log as the test runs.\n+ *\n+ * When the specified number of messages have been produced we create a consumer and consume all the messages in the topic\n+ * and write that out to another text file.\n+ *\n+ * Using a stable unix sort we sort both the producer log of what was sent and the consumer log of what was retrieved by the message key.\n+ * Then we compare the final message in both logs for each key. If this final message is not the same for all keys we\n+ * print an error and exit with exit code 1, otherwise we print the size reduction and exit with exit code 0.\n+ */\n+object TestLogCleaning {\n+\n+  def main(args: Array[String]) {\n+    val parser = new OptionParser(false)\n+    val numMessagesOpt = parser.accepts(\"messages\", \"The number of messages to send or consume.\")\n+      .withRequiredArg\n+      .describedAs(\"count\")\n+      .ofType(classOf[java.lang.Long])\n+      .defaultsTo(Long.MaxValue)\n+    val messageCompressionOpt = parser.accepts(\"compression-type\", \"message compression type\")\n+      .withOptionalArg()\n+      .describedAs(\"compressionType\")\n+      .ofType(classOf[java.lang.String])\n+      .defaultsTo(\"none\")\n+    val numDupsOpt = parser.accepts(\"duplicates\", \"The number of duplicates for each key.\")\n+      .withRequiredArg\n+      .describedAs(\"count\")\n+      .ofType(classOf[java.lang.Integer])\n+      .defaultsTo(5)\n+    val brokerOpt = parser.accepts(\"broker\", \"Url to connect to.\")\n+      .withRequiredArg\n+      .describedAs(\"url\")\n+      .ofType(classOf[String])\n+    val topicsOpt = parser.accepts(\"topics\", \"The number of topics to test.\")\n+      .withRequiredArg\n+      .describedAs(\"count\")\n+      .ofType(classOf[java.lang.Integer])\n+      .defaultsTo(1)\n+    val percentDeletesOpt = parser.accepts(\"percent-deletes\", \"The percentage of updates that are deletes.\")\n+      .withRequiredArg\n+      .describedAs(\"percent\")\n+      .ofType(classOf[java.lang.Integer])\n+      .defaultsTo(0)\n+    val sleepSecsOpt = parser.accepts(\"sleep\", \"Time to sleep between production and consumption.\")\n+      .withRequiredArg\n+      .describedAs(\"ms\")\n+      .ofType(classOf[java.lang.Integer])\n+      .defaultsTo(0)\n+    val dumpOpt = parser.accepts(\"dump\", \"Dump the message contents of a topic partition that contains test data from this test to standard out.\")\n+      .withRequiredArg\n+      .describedAs(\"directory\")\n+      .ofType(classOf[String])\n+\n+    val options = parser.parse(args: _*)\n+\n+    if (args.length == 0)\n+      CommandLineUtils.printUsageAndDie(parser, \"An integration test for log cleaning.\")\n+\n+    if (options.has(dumpOpt)) {\n+      dumpLog(new File(options.valueOf(dumpOpt)))\n+      Exit.exit(0)\n+    }\n+\n+    CommandLineUtils.checkRequiredArgs(parser, options, brokerOpt, numMessagesOpt)\n+\n+    // parse options\n+    val messages = options.valueOf(numMessagesOpt).longValue\n+    val compressionType = options.valueOf(messageCompressionOpt)\n+    val percentDeletes = options.valueOf(percentDeletesOpt).intValue\n+    val dups = options.valueOf(numDupsOpt).intValue\n+    val brokerUrl = options.valueOf(brokerOpt)\n+    val topicCount = options.valueOf(topicsOpt).intValue\n+    val sleepSecs = options.valueOf(sleepSecsOpt).intValue\n+\n+    val testId = new Random().nextInt(Int.MaxValue)\n+    val topics = (0 until topicCount).map(\"log-cleaner-test-\" + testId + \"-\" + _).toArray\n+    createTopics(brokerUrl, topics.toSeq)\n+\n+    println(\"Producing %d messages..to topics %s\".format(messages, topics.mkString(\",\")))\n+    val producedDataFile = produceMessages(brokerUrl, topics, messages, compressionType, dups, percentDeletes)\n+    println(\"Sleeping for %d seconds...\".format(sleepSecs))\n+    Thread.sleep(sleepSecs * 1000)\n+    println(\"Consuming messages...\")\n+    val consumedDataFile = consumeMessages(brokerUrl, topics)\n+\n+    val producedLines = lineCount(producedDataFile)\n+    val consumedLines = lineCount(consumedDataFile)\n+    val reduction = 1.0 - consumedLines.toDouble / producedLines.toDouble\n+    println(\"%d rows of data produced, %d rows of data consumed (%.1f%% reduction).\".format(producedLines, consumedLines, 100 * reduction))\n+\n+    println(\"De-duplicating and validating output files...\")\n+    validateOutput(producedDataFile, consumedDataFile)\n+    Utils.delete(producedDataFile)\n+    Utils.delete(consumedDataFile)\n+    //if you change this line, we need to update test_log_compaction_tool.py system test\n+    println(\"Data verification is completed\")\n+  }\n+\n+  def createTopics(brokerUrl: String, topics: Seq[String]): Unit = {\n+    val adminConfig = new Properties\n+    adminConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerUrl)\n+    val adminClient = admin.AdminClient.create(adminConfig)\n+\n+    val topicConfigs = Map(TopicConfig.CLEANUP_POLICY_CONFIG -> TopicConfig.CLEANUP_POLICY_COMPACT)\n+    val newTopics = topics.map(name => new NewTopic(name, 1, 1).configs(topicConfigs.asJava)).asJava\n+    adminClient.createTopics(newTopics).all.get()\n+\n+    TestUtils.waitUntilTrue(() => {\n+        val newTopics = adminClient.listTopics.names.get()\n+        topics.forall(topicName => newTopics.contains(topicName))\n+    }, \"timed out waiting for topics\")\n+\n+    adminClient.close()\n+  }\n+\n+  def dumpLog(dir: File) {\n+    require(dir.exists, \"Non-existent directory: \" + dir.getAbsolutePath)\n+    for (file <- dir.list.sorted; if file.endsWith(Log.LogFileSuffix)) {\n+      val fileRecords = FileRecords.open(new File(dir, file))\n+      for (entry <- fileRecords.records.asScala) {\n+        val key = readString(entry.key)\n+        val content =\n+          if (!entry.hasValue)\n+            null\n+          else\n+            readString(entry.value)\n+        println(\"offset = %s, key = %s, content = %s\".format(entry.offset, key, content))\n+      }\n+    }\n+  }\n+\n+  def lineCount(file: File): Int = io.Source.fromFile(file).getLines.size\n+\n+  def validateOutput(producedDataFile: File, consumedDataFile: File) {\n+    val producedReader = externalSort(producedDataFile)\n+    val consumedReader = externalSort(consumedDataFile)\n+    val produced = valuesIterator(producedReader)\n+    val consumed = valuesIterator(consumedReader)\n+\n+    val producedDedupedFile = new File(producedDataFile.getAbsolutePath + \".deduped\")\n+    val produceWriter : OutputStreamWriter = new OutputStreamWriter(new FileOutputStream(producedDedupedFile), UTF_8)\n+    val producedDeduped = new BufferedWriter(produceWriter, 1024 * 1024)\n+\n+    val consumedDedupedFile = new File(consumedDataFile.getAbsolutePath + \".deduped\")\n+    val consumeWriter : OutputStreamWriter = new OutputStreamWriter(new FileOutputStream(consumedDedupedFile), UTF_8)\n+    val consumedDeduped = new BufferedWriter(consumeWriter, 1024 * 1024)\n+    var total = 0\n+    var mismatched = 0\n+    while (produced.hasNext && consumed.hasNext) {\n+      val p = produced.next()\n+      producedDeduped.write(p.toString)\n+      producedDeduped.newLine()\n+      val c = consumed.next()\n+      consumedDeduped.write(c.toString)\n+      consumedDeduped.newLine()\n+      if (p != c)\n+        mismatched += 1\n+      total += 1\n+    }\n+    producedDeduped.close()\n+    consumedDeduped.close()\n+    println(\"Validated \" + total + \" values, \" + mismatched + \" mismatches.\")\n+    require(!produced.hasNext, \"Additional values produced not found in consumer log.\")\n+    require(!consumed.hasNext, \"Additional values consumed not found in producer log.\")\n+    require(mismatched == 0, \"Non-zero number of row mismatches.\")\n+    // if all the checks worked out we can delete the deduped files\n+    Utils.delete(producedDedupedFile)\n+    Utils.delete(consumedDedupedFile)\n+  }\n+\n+  def valuesIterator(reader: BufferedReader) = {\n+    new IteratorTemplate[TestRecord] {\n+      def makeNext(): TestRecord = {\n+        var next = readNext(reader)\n+        while (next != null && next.delete)\n+          next = readNext(reader)\n+        if (next == null)\n+          allDone()\n+        else\n+          next\n+      }\n+    }\n+  }\n+\n+  def readNext(reader: BufferedReader): TestRecord = {\n+    var line = reader.readLine()\n+    if (line == null)\n+      return null\n+    var curr = new TestRecord(line)\n+    while (true) {\n+      line = peekLine(reader)\n+      if (line == null)\n+        return curr\n+      val next = new TestRecord(line)\n+      if (next == null || next.topicAndKey != curr.topicAndKey)\n+        return curr\n+      curr = next\n+      reader.readLine()\n+    }\n+    null\n+  }\n+\n+  def peekLine(reader: BufferedReader) = {\n+    reader.mark(4096)\n+    val line = reader.readLine\n+    reader.reset()\n+    line\n+  }\n+\n+  def externalSort(file: File): BufferedReader = {\n+    val builder = new ProcessBuilder(\"sort\", \"--key=1,2\", \"--stable\", \"--buffer-size=20%\", \"--temporary-directory=\" + System.getProperty(\"java.io.tmpdir\"), file.getAbsolutePath)\n+    val process = builder.start()\n+    new Thread() {\n+      override def run() {\n+        val exitCode = process.waitFor()\n+        if (exitCode != 0) {\n+          System.err.println(\"Process exited abnormally.\")\n+          while (process.getErrorStream.available > 0) {\n+            System.err.write(process.getErrorStream().read())\n+          }\n+        }\n+      }\n+    }.start()\n+    new BufferedReader(new InputStreamReader(process.getInputStream(), UTF_8), 10 * 1024 * 1024)\n+  }\n+\n+  def produceMessages(brokerUrl: String,\n+                      topics: Array[String],\n+                      messages: Long,\n+                      compressionType: String,\n+                      dups: Int,\n+                      percentDeletes: Int): File = {\n+    val producerProps = new Properties\n+    producerProps.setProperty(ProducerConfig.MAX_BLOCK_MS_CONFIG, Long.MaxValue.toString)\n+    producerProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl)\n+    producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, \"org.apache.kafka.common.serialization.ByteArraySerializer\")\n+    producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, \"org.apache.kafka.common.serialization.ByteArraySerializer\")\n+    producerProps.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG, compressionType)\n+    val producer = new KafkaProducer[Array[Byte], Array[Byte]](producerProps)\n+    val rand = new Random(1)\n+    val keyCount = (messages / dups).toInt\n+    val producedFile = File.createTempFile(\"kafka-log-cleaner-produced-\", \".txt\")\n+    println(\"Logging produce requests to \" + producedFile.getAbsolutePath)\n+    val produceFileWriter : OutputStreamWriter = new OutputStreamWriter(new FileOutputStream(producedFile), UTF_8)\n+    val producedWriter = new BufferedWriter(produceFileWriter, 1024 * 1024)\n+    for (i <- 0L until (messages * topics.length)) {\n+      val topic = topics((i % topics.length).toInt)\n+      val key = rand.nextInt(keyCount)\n+      val delete = i % 100 < percentDeletes\n+      val msg =\n+        if (delete)\n+          new ProducerRecord[Array[Byte], Array[Byte]](topic, key.toString.getBytes(UTF_8), null)\n+        else\n+          new ProducerRecord[Array[Byte], Array[Byte]](topic, key.toString.getBytes(UTF_8), i.toString.getBytes(UTF_8))\n+      producer.send(msg)","path":"core/src/test/scala/kafka/tools/TestLogCleaning.scala","position":null,"original_position":300,"commit_id":"16ff590264a2aa325ad0cda3fe442007c93250dc","original_commit_id":"b3b7b96358fdffe988a20e62a17e897db6fdb62e","user":{"login":"omkreddy","id":8134545,"node_id":"MDQ6VXNlcjgxMzQ1NDU=","avatar_url":"https://avatars2.githubusercontent.com/u/8134545?v=4","gravatar_id":"","url":"https://api.github.com/users/omkreddy","html_url":"https://github.com/omkreddy","followers_url":"https://api.github.com/users/omkreddy/followers","following_url":"https://api.github.com/users/omkreddy/following{/other_user}","gists_url":"https://api.github.com/users/omkreddy/gists{/gist_id}","starred_url":"https://api.github.com/users/omkreddy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/omkreddy/subscriptions","organizations_url":"https://api.github.com/users/omkreddy/orgs","repos_url":"https://api.github.com/users/omkreddy/repos","events_url":"https://api.github.com/users/omkreddy/events{/privacy}","received_events_url":"https://api.github.com/users/omkreddy/received_events","type":"User","site_admin":false},"body":"test fails at data validation phase if producer failed to send data. I am going leave this as it is.","created_at":"2018-07-26T17:59:16Z","updated_at":"2018-07-26T17:59:16Z","html_url":"https://github.com/apache/kafka/pull/5226#discussion_r205549444","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/5226","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/205549444"},"html":{"href":"https://github.com/apache/kafka/pull/5226#discussion_r205549444"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/5226"}},"in_reply_to_id":205183170},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/5226","id":194898923,"node_id":"MDExOlB1bGxSZXF1ZXN0MTk0ODk4OTIz","html_url":"https://github.com/apache/kafka/pull/5226","diff_url":"https://github.com/apache/kafka/pull/5226.diff","patch_url":"https://github.com/apache/kafka/pull/5226.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/5226","number":5226,"state":"open","locked":false,"title":"KAFKA-7210:  Add a system test to verify the log compaction ","user":{"login":"omkreddy","id":8134545,"node_id":"MDQ6VXNlcjgxMzQ1NDU=","avatar_url":"https://avatars2.githubusercontent.com/u/8134545?v=4","gravatar_id":"","url":"https://api.github.com/users/omkreddy","html_url":"https://github.com/omkreddy","followers_url":"https://api.github.com/users/omkreddy/followers","following_url":"https://api.github.com/users/omkreddy/following{/other_user}","gists_url":"https://api.github.com/users/omkreddy/gists{/gist_id}","starred_url":"https://api.github.com/users/omkreddy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/omkreddy/subscriptions","organizations_url":"https://api.github.com/users/omkreddy/orgs","repos_url":"https://api.github.com/users/omkreddy/repos","events_url":"https://api.github.com/users/omkreddy/events{/privacy}","received_events_url":"https://api.github.com/users/omkreddy/received_events","type":"User","site_admin":false},"body":"- Update TestLogCleaning tool to use Java consumer\r\n- Enable log compaction in System tests\r\n- Remove configs with values same as server defaults from \"kafka.properties\" template file.\r\n- Update the kafka.py logic to handle the duplicates between kafka.properties and server_prop_overides.\r\n\r\n","created_at":"2018-06-14T13:45:35Z","updated_at":"2018-07-26T17:59:16Z","closed_at":null,"merged_at":null,"merge_commit_sha":"4ae66d6493a8c1e5e1260129b1fdd208d252cf9d","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":{"url":"https://api.github.com/repos/apache/kafka/milestones/3","html_url":"https://github.com/apache/kafka/milestone/3","labels_url":"https://api.github.com/repos/apache/kafka/milestones/3/labels","id":3427062,"node_id":"MDk6TWlsZXN0b25lMzQyNzA2Mg==","number":3,"title":"2.1.0","description":null,"creator":{"login":"ijuma","id":24747,"node_id":"MDQ6VXNlcjI0NzQ3","avatar_url":"https://avatars2.githubusercontent.com/u/24747?v=4","gravatar_id":"","url":"https://api.github.com/users/ijuma","html_url":"https://github.com/ijuma","followers_url":"https://api.github.com/users/ijuma/followers","following_url":"https://api.github.com/users/ijuma/following{/other_user}","gists_url":"https://api.github.com/users/ijuma/gists{/gist_id}","starred_url":"https://api.github.com/users/ijuma/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ijuma/subscriptions","organizations_url":"https://api.github.com/users/ijuma/orgs","repos_url":"https://api.github.com/users/ijuma/repos","events_url":"https://api.github.com/users/ijuma/events{/privacy}","received_events_url":"https://api.github.com/users/ijuma/received_events","type":"User","site_admin":false},"open_issues":2,"closed_issues":1,"state":"open","created_at":"2018-06-15T06:37:16Z","updated_at":"2018-07-26T16:13:51Z","due_on":null,"closed_at":null},"commits_url":"https://api.github.com/repos/apache/kafka/pulls/5226/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/5226/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/5226/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/16ff590264a2aa325ad0cda3fe442007c93250dc","head":{"label":"omkreddy:TestLogClean","ref":"TestLogClean","sha":"16ff590264a2aa325ad0cda3fe442007c93250dc","user":{"login":"omkreddy","id":8134545,"node_id":"MDQ6VXNlcjgxMzQ1NDU=","avatar_url":"https://avatars2.githubusercontent.com/u/8134545?v=4","gravatar_id":"","url":"https://api.github.com/users/omkreddy","html_url":"https://github.com/omkreddy","followers_url":"https://api.github.com/users/omkreddy/followers","following_url":"https://api.github.com/users/omkreddy/following{/other_user}","gists_url":"https://api.github.com/users/omkreddy/gists{/gist_id}","starred_url":"https://api.github.com/users/omkreddy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/omkreddy/subscriptions","organizations_url":"https://api.github.com/users/omkreddy/orgs","repos_url":"https://api.github.com/users/omkreddy/repos","events_url":"https://api.github.com/users/omkreddy/events{/privacy}","received_events_url":"https://api.github.com/users/omkreddy/received_events","type":"User","site_admin":false},"repo":{"id":51066257,"node_id":"MDEwOlJlcG9zaXRvcnk1MTA2NjI1Nw==","name":"kafka","full_name":"omkreddy/kafka","owner":{"login":"omkreddy","id":8134545,"node_id":"MDQ6VXNlcjgxMzQ1NDU=","avatar_url":"https://avatars2.githubusercontent.com/u/8134545?v=4","gravatar_id":"","url":"https://api.github.com/users/omkreddy","html_url":"https://github.com/omkreddy","followers_url":"https://api.github.com/users/omkreddy/followers","following_url":"https://api.github.com/users/omkreddy/following{/other_user}","gists_url":"https://api.github.com/users/omkreddy/gists{/gist_id}","starred_url":"https://api.github.com/users/omkreddy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/omkreddy/subscriptions","organizations_url":"https://api.github.com/users/omkreddy/orgs","repos_url":"https://api.github.com/users/omkreddy/repos","events_url":"https://api.github.com/users/omkreddy/events{/privacy}","received_events_url":"https://api.github.com/users/omkreddy/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/omkreddy/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/omkreddy/kafka","forks_url":"https://api.github.com/repos/omkreddy/kafka/forks","keys_url":"https://api.github.com/repos/omkreddy/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/omkreddy/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/omkreddy/kafka/teams","hooks_url":"https://api.github.com/repos/omkreddy/kafka/hooks","issue_events_url":"https://api.github.com/repos/omkreddy/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/omkreddy/kafka/events","assignees_url":"https://api.github.com/repos/omkreddy/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/omkreddy/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/omkreddy/kafka/tags","blobs_url":"https://api.github.com/repos/omkreddy/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/omkreddy/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/omkreddy/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/omkreddy/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/omkreddy/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/omkreddy/kafka/languages","stargazers_url":"https://api.github.com/repos/omkreddy/kafka/stargazers","contributors_url":"https://api.github.com/repos/omkreddy/kafka/contributors","subscribers_url":"https://api.github.com/repos/omkreddy/kafka/subscribers","subscription_url":"https://api.github.com/repos/omkreddy/kafka/subscription","commits_url":"https://api.github.com/repos/omkreddy/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/omkreddy/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/omkreddy/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/omkreddy/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/omkreddy/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/omkreddy/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/omkreddy/kafka/merges","archive_url":"https://api.github.com/repos/omkreddy/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/omkreddy/kafka/downloads","issues_url":"https://api.github.com/repos/omkreddy/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/omkreddy/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/omkreddy/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/omkreddy/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/omkreddy/kafka/labels{/name}","releases_url":"https://api.github.com/repos/omkreddy/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/omkreddy/kafka/deployments","created_at":"2016-02-04T09:42:20Z","updated_at":"2018-07-10T12:11:28Z","pushed_at":"2018-07-26T17:56:38Z","git_url":"git://github.com/omkreddy/kafka.git","ssh_url":"git@github.com:omkreddy/kafka.git","clone_url":"https://github.com/omkreddy/kafka.git","svn_url":"https://github.com/omkreddy/kafka","homepage":null,"size":67359,"stargazers_count":1,"watchers_count":1,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"archived":false,"open_issues_count":0,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":0,"open_issues":0,"watchers":1,"default_branch":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"a932520135d42c7d9731064d96c21ab2fc5de696","user":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","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,"node_id":"MDEwOlJlcG9zaXRvcnkyMjExMjQz","name":"kafka","full_name":"apache/kafka","owner":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","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}","deployments_url":"https://api.github.com/repos/apache/kafka/deployments","created_at":"2011-08-15T18:06:16Z","updated_at":"2018-07-26T17:47:11Z","pushed_at":"2018-07-26T17:56: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":71145,"stargazers_count":9047,"watchers_count":9047,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":5196,"mirror_url":null,"archived":false,"open_issues_count":480,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":5196,"open_issues":480,"watchers":9047,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/5226"},"html":{"href":"https://github.com/apache/kafka/pull/5226"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/5226"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/5226/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/5226/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/5226/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/16ff590264a2aa325ad0cda3fe442007c93250dc"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"omkreddy"}}