{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/43668275","id":43668275,"diff_hunk":"@@ -0,0 +1,919 @@\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.coordinator\n+\n+import java.util.concurrent.locks.ReentrantReadWriteLock\n+\n+import kafka.utils.CoreUtils._\n+import org.apache.kafka.common.protocol.Errors\n+import org.apache.kafka.common.protocol.types.{ArrayOf, Struct, Schema, Field}\n+import org.apache.kafka.common.protocol.types.Type.STRING\n+import org.apache.kafka.common.protocol.types.Type.INT32\n+import org.apache.kafka.common.protocol.types.Type.INT64\n+import org.apache.kafka.common.protocol.types.Type.BYTES\n+import org.apache.kafka.common.utils.Utils\n+\n+import kafka.utils._\n+import kafka.common._\n+import kafka.message._\n+import kafka.log.FileMessageSet\n+import kafka.metrics.KafkaMetricsGroup\n+import kafka.common.TopicAndPartition\n+import kafka.tools.MessageFormatter\n+import kafka.api.ProducerResponseStatus\n+import kafka.server.ReplicaManager\n+\n+import scala.collection._\n+import java.io.PrintStream\n+import java.nio.ByteBuffer\n+import java.util.concurrent.atomic.AtomicBoolean\n+import java.util.concurrent.{CountDownLatch, TimeUnit}\n+\n+import com.yammer.metrics.core.Gauge\n+\n+class GroupMetadataManager(val brokerId: Int,\n+                           val config: OffsetConfig,\n+                           replicaManager: ReplicaManager,\n+                           zkUtils: ZkUtils,\n+                           scheduler: Scheduler) extends Logging with KafkaMetricsGroup {\n+\n+  /* offsets cache */\n+  private val offsetsCache = new Pool[GroupTopicPartition, OffsetAndMetadata]\n+\n+  /* group metadata cache */\n+  private val groupsCache = new Pool[String, GroupMetadata]\n+\n+  /* partitions of consumer groups that are being loaded */\n+  private val loadingPartitions: mutable.Set[Int] = mutable.Set()\n+\n+  /* partitions of consumer groups that are assigned, using the same loading partition lock */\n+  private val ownedPartitions: mutable.Set[Int] = mutable.Set()\n+\n+  /* lock for expiring stale offsets */\n+  private val offsetExpireLock = new ReentrantReadWriteLock()\n+\n+  /* lock for removing offsets of a range partition */\n+  private val offsetRemoveLock = new ReentrantReadWriteLock()\n+\n+  /* shutting down flag */\n+  private val shuttingDown = new AtomicBoolean(false)\n+\n+  /* number of partitions for the consumer metadata topic */\n+  private val groupMetadataTopicPartitionCount = getOffsetsTopicPartitionCount\n+\n+  this.logIdent = \"[Group Metadata Manager on Broker \" + brokerId + \"]: \"\n+\n+  scheduler.schedule(name = \"delete-expired-consumer-offsets\",\n+    fun = deleteExpiredOffsets,\n+    period = config.offsetsRetentionCheckIntervalMs,\n+    unit = TimeUnit.MILLISECONDS)\n+\n+  newGauge(\"NumOffsets\",\n+    new Gauge[Int] {\n+      def value = offsetsCache.size\n+    }\n+  )\n+\n+  newGauge(\"NumGroups\",\n+    new Gauge[Int] {\n+      def value = offsetsCache.keys.map(_.group).toSet.size\n+    }\n+  )\n+\n+  def partitionFor(group: String): Int = Utils.abs(group.hashCode) % groupMetadataTopicPartitionCount\n+\n+  def partitionLeaderIsLocal(partition: Int): Boolean = loadingPartitions synchronized ownedPartitions.contains(partition)\n+\n+  def partitionLoadingInProgress(partition: Int): Boolean = loadingPartitions synchronized loadingPartitions.contains(partition)\n+\n+  /**\n+   * Get the group associated with the given groupId, or null if not found\n+   */\n+  def getGroup(groupId: String): GroupMetadata = {\n+      groupsCache.get(groupId)\n+  }\n+\n+  /**\n+   * Add a group or get the group associated with the given groupId if it already exists\n+   */\n+  def addGroup(groupId: String, protocolType: String): GroupMetadata = {\n+    addGroup(groupId, new GroupMetadata(groupId, protocolType))\n+  }\n+\n+  private def addGroup(groupId: String, group: GroupMetadata): GroupMetadata = {\n+    groupsCache.putIfNotExists(groupId, group)\n+    group\n+  }\n+\n+  /**\n+   * Remove all metadata associated with the group, note this function needs to be\n+   * called inside the group lock\n+   * @param groupId the groupId of the group we are removing\n+   */\n+  def removeGroup(groupId: String) {\n+    if (groupsCache.remove(groupId) == null)\n+      throw new IllegalArgumentException(\"Cannot remove non-existing group\")\n+\n+    // Append the tombstone messages to the partition. It is okay if the replicas don't receive these (say,\n+    // if we crash or leaders move) since the new leaders will still expire the consumers with heartbeat and\n+    // retry removing this group.\n+    val groupPartition = partitionFor(groupId)\n+    val tomstone = new Message(bytes = null, key = GroupMetadataManager.groupMetadataKey(groupId))\n+\n+    val partitionOpt = replicaManager.getPartition(GroupCoordinator.GroupMetadataTopicName, groupPartition)\n+    partitionOpt.foreach { partition =>\n+      val appendPartition = TopicAndPartition(GroupCoordinator.GroupMetadataTopicName, groupPartition)\n+\n+      trace(\"Marking group %s as deleted.\".format(groupId))\n+\n+      try {\n+        // do not need to require acks since even if the tombsone is lost,\n+        // it will be appended again by the new leader\n+        // TODO: have periodic purging instead of immediate removal of groups\n+        partition.appendMessagesToLeader(new ByteBufferMessageSet(config.offsetsTopicCompressionCodec, tomstone))\n+      } catch {\n+        case t: Throwable =>\n+          error(\"Failed to mark group %s as deleted in %s.\".format(groupId, appendPartition), t)\n+          // ignore and continue\n+      }\n+    }\n+  }\n+\n+  def storeGroup(group: GroupMetadata,\n+                 groupAssignment: Map[String, Array[Byte]]) {\n+    // construct the message to append\n+    val message = new Message(\n+      key = GroupMetadataManager.groupMetadataKey(group.groupId),\n+      bytes = GroupMetadataManager.groupMetadataValue(group, groupAssignment)\n+    )\n+\n+    val groupMetadataPartition = TopicAndPartition(GroupCoordinator.GroupMetadataTopicName, partitionFor(group.groupId))\n+\n+    val groupMetadataMessageSet = Map(groupMetadataPartition ->\n+      new ByteBufferMessageSet(config.offsetsTopicCompressionCodec, message))\n+\n+    // set the callback function to insert offsets into cache after log append completed\n+    def putCacheCallback(responseStatus: Map[TopicAndPartition, ProducerResponseStatus]) {\n+      // the append response should only contain the topics partition\n+      if (responseStatus.size != 1 || ! responseStatus.contains(groupMetadataPartition))\n+        throw new IllegalStateException(\"Append status %s should only have one partition %s\"\n+          .format(responseStatus, groupMetadataPartition))\n+\n+      // construct the commit response status and insert\n+      // the offset and metadata to cache if the append status has no error\n+      val status = responseStatus(groupMetadataPartition)\n+\n+      var responseCode = Errors.NONE.code\n+      if (status.error != ErrorMapping.NoError) {\n+        debug(\"Metadata from group %s with generation %d failed when appending to log due to %s\"\n+          .format(group.groupId, group.generationId, ErrorMapping.exceptionNameFor(status.error)))\n+\n+        // transform the log append error code to the corresponding the commit status error code\n+        responseCode = if (status.error == ErrorMapping.UnknownTopicOrPartitionCode) {\n+          Errors.GROUP_COORDINATOR_NOT_AVAILABLE.code\n+        } else if (status.error == ErrorMapping.NotLeaderForPartitionCode) {\n+          Errors.NOT_COORDINATOR_FOR_GROUP.code\n+        } else if (status.error == ErrorMapping.MessageSizeTooLargeCode\n+          || status.error == ErrorMapping.MessageSetSizeTooLargeCode\n+          || status.error == ErrorMapping.InvalidFetchSizeCode) {\n+\n+          error(\"Appending metadata message for group %s generation %d failed due to %s, returning UNKNOWN error code to the client\"\n+            .format(group.groupId, group.generationId, ErrorMapping.exceptionNameFor(status.error)))\n+\n+          Errors.UNKNOWN.code\n+        } else {\n+\n+          error(\"Appending metadata message for group %s generation %d failed due to unexpected error: %s\"\n+            .format(group.groupId, group.generationId, status.error))\n+\n+          status.error\n+        }\n+      }\n+\n+      for (member <- group.allMembers) {\n+        member.assignment = groupAssignment.getOrElse(member.memberId, Array.empty[Byte])\n+      }\n+\n+      // propagate the assignments\n+      propagateAssignment(group, Errors.NONE.code)\n+    }\n+\n+    // call replica manager to append the offset messages\n+    replicaManager.appendMessages(\n+      config.offsetCommitTimeoutMs.toLong,\n+      config.offsetCommitRequiredAcks,\n+      true, // allow appending to internal offset topic\n+      groupMetadataMessageSet,\n+      putCacheCallback)\n+  }\n+\n+  def propagateAssignment(group: GroupMetadata,\n+                          errorCode: Short) {\n+    val hasError = errorCode != Errors.NONE.code\n+    for (member <- group.allMembers) {\n+      if (member.awaitingSyncCallback != null) {\n+        member.awaitingSyncCallback(if (hasError) Array.empty else member.assignment, errorCode)\n+        member.awaitingSyncCallback = null\n+      }\n+    }\n+  }\n+\n+  /**\n+   * Store offsets by appending it to the replicated log and then inserting to cache\n+   */\n+  def storeOffsets(groupId: String,\n+                   consumerId: String,\n+                   generationId: Int,\n+                   offsetMetadata: immutable.Map[TopicAndPartition, OffsetAndMetadata],\n+                   responseCallback: immutable.Map[TopicAndPartition, Short] => Unit) {\n+    // first filter out partitions with offset metadata size exceeding limit\n+    val filteredOffsetMetadata = offsetMetadata.filter { case (topicAndPartition, offsetAndMetadata) =>\n+      validateOffsetMetadataLength(offsetAndMetadata.metadata)\n+    }\n+\n+    // construct the message set to append\n+    val messages = filteredOffsetMetadata.map { case (topicAndPartition, offsetAndMetadata) =>\n+      new Message(\n+        key = GroupMetadataManager.offsetCommitKey(groupId, topicAndPartition.topic, topicAndPartition.partition),\n+        bytes = GroupMetadataManager.offsetCommitValue(offsetAndMetadata)\n+      )\n+    }.toSeq\n+\n+    val offsetTopicPartition = TopicAndPartition(GroupCoordinator.GroupMetadataTopicName, partitionFor(groupId))\n+\n+    val offsetsAndMetadataMessageSet = Map(offsetTopicPartition ->\n+      new ByteBufferMessageSet(config.offsetsTopicCompressionCodec, messages:_*))\n+\n+    // set the callback function to insert offsets into cache after log append completed\n+    def putCacheCallback(responseStatus: Map[TopicAndPartition, ProducerResponseStatus]) {\n+      // the append response should only contain the topics partition\n+      if (responseStatus.size != 1 || ! responseStatus.contains(offsetTopicPartition))\n+        throw new IllegalStateException(\"Append status %s should only have one partition %s\"\n+          .format(responseStatus, offsetTopicPartition))\n+\n+      // construct the commit response status and insert\n+      // the offset and metadata to cache if the append status has no error\n+      val status = responseStatus(offsetTopicPartition)\n+\n+      val responseCode =\n+        if (status.error == ErrorMapping.NoError) {\n+          filteredOffsetMetadata.foreach { case (topicAndPartition, offsetAndMetadata) =>\n+            putOffset(GroupTopicPartition(groupId, topicAndPartition), offsetAndMetadata)\n+          }\n+          ErrorMapping.NoError\n+        } else {\n+          debug(\"Offset commit %s from group %s consumer %s with generation %d failed when appending to log due to %s\"\n+            .format(filteredOffsetMetadata, groupId, consumerId, generationId, ErrorMapping.exceptionNameFor(status.error)))\n+\n+          // transform the log append error code to the corresponding the commit status error code\n+          if (status.error == ErrorMapping.UnknownTopicOrPartitionCode)\n+            ErrorMapping.ConsumerCoordinatorNotAvailableCode\n+          else if (status.error == ErrorMapping.NotLeaderForPartitionCode)\n+            ErrorMapping.NotCoordinatorForConsumerCode\n+          else if (status.error == ErrorMapping.MessageSizeTooLargeCode\n+            || status.error == ErrorMapping.MessageSetSizeTooLargeCode\n+            || status.error == ErrorMapping.InvalidFetchSizeCode)\n+            Errors.INVALID_COMMIT_OFFSET_SIZE.code\n+          else\n+            status.error\n+        }\n+\n+\n+      // compute the final error codes for the commit response\n+      val commitStatus = offsetMetadata.map { case (topicAndPartition, offsetAndMetadata) =>\n+        if (validateOffsetMetadataLength(offsetAndMetadata.metadata))\n+          (topicAndPartition, responseCode)\n+        else\n+          (topicAndPartition, ErrorMapping.OffsetMetadataTooLargeCode)\n+      }\n+\n+      // finally trigger the callback logic passed from the API layer\n+      responseCallback(commitStatus)\n+    }\n+\n+    // call replica manager to append the offset messages\n+    replicaManager.appendMessages(\n+      config.offsetCommitTimeoutMs.toLong,\n+      config.offsetCommitRequiredAcks,\n+      true, // allow appending to internal offset topic\n+      offsetsAndMetadataMessageSet,\n+      putCacheCallback)\n+  }\n+\n+  /**\n+   * The most important guarantee that this API provides is that it should never return a stale offset. i.e., it either\n+   * returns the current offset or it begins to sync the cache from the log (and returns an error code).\n+   */\n+  def getOffsets(group: String, topicPartitions: Seq[TopicAndPartition]): Map[TopicAndPartition, OffsetMetadataAndError] = {\n+    trace(\"Getting offsets %s for group %s.\".format(topicPartitions, group))\n+\n+    val offsetsPartition = partitionFor(group)\n+\n+    /**\n+     * we need to put the leader-is-local check inside the offsetLock to protects against fetching from an empty/cleared\n+     * offset cache (i.e., cleared due to a leader->follower transition right after the check and clear the cache).\n+     */\n+    inReadLock(offsetRemoveLock) {\n+      if (partitionLeaderIsLocal(offsetsPartition)) {\n+        if (topicPartitions.isEmpty) {\n+          // Return offsets for all partitions owned by this consumer group. (this only applies to consumers that commit offsets to Kafka.)\n+          offsetsCache.filter(_._1.group == group).map { case(groupTopicPartition, offsetAndMetadata) =>\n+            (groupTopicPartition.topicPartition, OffsetMetadataAndError(offsetAndMetadata.offset, offsetAndMetadata.metadata, ErrorMapping.NoError))\n+          }.toMap\n+        } else {\n+          topicPartitions.map { topicAndPartition =>\n+            val groupTopicPartition = GroupTopicPartition(group, topicAndPartition)\n+            (groupTopicPartition.topicPartition, getOffset(groupTopicPartition))\n+          }.toMap\n+        }\n+      } else {\n+        debug(\"Could not fetch offsets for group %s (not offset coordinator).\".format(group))\n+        topicPartitions.map { topicAndPartition =>\n+          val groupTopicPartition = GroupTopicPartition(group, topicAndPartition)\n+          (groupTopicPartition.topicPartition, OffsetMetadataAndError.NotCoordinatorForGroup)\n+        }.toMap\n+      }\n+    }\n+  }\n+\n+  /**\n+   * Asynchronously read the partition from the offsets topic and populate the cache\n+   */\n+  def loadGroupsForPartition(offsetsPartition: Int) {\n+\n+    val topicPartition = TopicAndPartition(GroupCoordinator.GroupMetadataTopicName, offsetsPartition)\n+\n+    loadingPartitions synchronized {\n+      ownedPartitions.add(offsetsPartition)\n+\n+      if (loadingPartitions.contains(offsetsPartition)) {\n+        info(\"Offset load from %s already in progress.\".format(topicPartition))\n+      } else {\n+        loadingPartitions.add(offsetsPartition)\n+        scheduler.schedule(topicPartition.toString, loadGroupsAndOffsets)\n+      }\n+    }\n+\n+    def loadGroupsAndOffsets() {\n+      info(\"Loading offsets from \" + topicPartition)\n+\n+      val startMs = SystemTime.milliseconds\n+      try {\n+        replicaManager.logManager.getLog(topicPartition) match {\n+          case Some(log) =>\n+            var currOffset = log.logSegments.head.baseOffset\n+            val buffer = ByteBuffer.allocate(config.loadBufferSize)\n+            // loop breaks if leader changes at any time during the load, since getHighWatermark is -1\n+            inWriteLock(offsetExpireLock) {\n+              while (currOffset < getHighWatermark(offsetsPartition) && !shuttingDown.get()) {\n+                buffer.clear()\n+                val messages = log.read(currOffset, config.loadBufferSize).messageSet.asInstanceOf[FileMessageSet]\n+                messages.readInto(buffer, 0)\n+                val messageSet = new ByteBufferMessageSet(buffer)\n+                messageSet.foreach { msgAndOffset =>\n+                  require(msgAndOffset.message.key != null, \"Offset entry key should not be null\")\n+                  val baseKey = GroupMetadataManager.readMessageKey(msgAndOffset.message.key)\n+\n+                  if (baseKey.isInstanceOf[OffsetKey]) {\n+                    // load offset\n+                    val key = baseKey.key.asInstanceOf[GroupTopicPartition]\n+                    if (msgAndOffset.message.payload == null) {\n+                      if (offsetsCache.remove(key) != null)\n+                        trace(\"Removed offset for %s due to tombstone entry.\".format(key))\n+                      else\n+                        trace(\"Ignoring redundant tombstone for %s.\".format(key))\n+                    } else {\n+                      // special handling for version 0:\n+                      // set the expiration time stamp as commit time stamp + server default retention time\n+                      val value = GroupMetadataManager.readOffsetMessageValue(msgAndOffset.message.payload)\n+                      putOffset(key, value.copy (\n+                        expireTimestamp = {\n+                          if (value.expireTimestamp == org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_TIMESTAMP)\n+                            value.commitTimestamp + config.offsetsRetentionMs\n+                          else\n+                            value.expireTimestamp\n+                        }\n+                      ))\n+                      trace(\"Loaded offset %s for %s.\".format(value, key))\n+                    }\n+                  } else {\n+                    // load group metadata\n+                    val groupId = baseKey.key.asInstanceOf[String]\n+                    val groupMetadata = GroupMetadataManager.readGroupMessageValue(groupId, msgAndOffset.message.payload)\n+\n+                    addGroup(groupId, groupMetadata)\n+                  }\n+\n+                  currOffset = msgAndOffset.nextOffset\n+                }\n+              }\n+            }\n+\n+            if (!shuttingDown.get())\n+              info(\"Finished loading offsets from %s in %d milliseconds.\"\n+                .format(topicPartition, SystemTime.milliseconds - startMs))\n+          case None =>\n+            warn(\"No log found for \" + topicPartition)\n+        }\n+      }\n+      catch {\n+        case t: Throwable =>\n+          error(\"Error in loading offsets from \" + topicPartition, t)\n+      }\n+      finally {\n+        loadingPartitions synchronized loadingPartitions.remove(offsetsPartition)\n+      }\n+    }\n+  }\n+\n+  /**\n+   * When this broker becomes a follower for an offsets topic partition clear out the cache for groups that belong to\n+   * that partition.\n+   * @param offsetsPartition Groups belonging to this partition of the offsets topic will be deleted from the cache.\n+   */\n+  def removeGroupsForPartition(offsetsPartition: Int) {\n+    var numRemoved = 0\n+    inWriteLock(offsetRemoveLock) {\n+      offsetsCache.keys.foreach { key =>\n+        if (partitionFor(key.group) == offsetsPartition) {\n+          offsetsCache.remove(key)\n+          numRemoved += 1\n+        }\n+      }\n+\n+      // TODO: we may also remove the group metadata cache here\n+    }\n+\n+    if (numRemoved > 0) info(\"Removed %d cached offsets for %s on follower transition.\"\n+      .format(numRemoved, TopicAndPartition(GroupCoordinator.GroupMetadataTopicName, offsetsPartition)))\n+\n+    loadingPartitions synchronized ownedPartitions.remove(offsetsPartition)\n+  }\n+\n+  /**\n+   * Fetch the current offset for the given group/topic/partition from the underlying offsets storage.\n+   *\n+   * @param key The requested group-topic-partition\n+   * @return If the key is present, return the offset and metadata; otherwise return None\n+   */\n+  private def getOffset(key: GroupTopicPartition) = {\n+    val offsetAndMetadata = offsetsCache.get(key)\n+    if (offsetAndMetadata == null)\n+      OffsetMetadataAndError.NoOffset\n+    else\n+      OffsetMetadataAndError(offsetAndMetadata.offset, offsetAndMetadata.metadata, ErrorMapping.NoError)\n+  }\n+\n+  /**\n+   * Put the (already committed) offset for the given group/topic/partition into the cache.\n+   *\n+   * @param key The group-topic-partition\n+   * @param offsetAndMetadata The offset/metadata to be stored\n+   */\n+  private def putOffset(key: GroupTopicPartition, offsetAndMetadata: OffsetAndMetadata) {\n+    offsetsCache.put(key, offsetAndMetadata)\n+  }\n+\n+  private def deleteExpiredOffsets() {\n+    debug(\"Collecting expired offsets.\")\n+    val startMs = SystemTime.milliseconds\n+\n+    val numExpiredOffsetsRemoved = inWriteLock(offsetExpireLock) {\n+      val expiredOffsets = offsetsCache.filter { case (groupTopicPartition, offsetAndMetadata) =>\n+        offsetAndMetadata.expireTimestamp < startMs\n+      }\n+\n+      debug(\"Found %d expired offsets.\".format(expiredOffsets.size))\n+\n+      // delete the expired offsets from the table and generate tombstone messages to remove them from the log\n+      val tombstonesForPartition = expiredOffsets.map { case (groupTopicAndPartition, offsetAndMetadata) =>\n+        val offsetsPartition = partitionFor(groupTopicAndPartition.group)\n+        trace(\"Removing expired offset and metadata for %s: %s\".format(groupTopicAndPartition, offsetAndMetadata))\n+\n+        offsetsCache.remove(groupTopicAndPartition)\n+\n+        val commitKey = GroupMetadataManager.offsetCommitKey(groupTopicAndPartition.group,\n+          groupTopicAndPartition.topicPartition.topic, groupTopicAndPartition.topicPartition.partition)\n+\n+        (offsetsPartition, new Message(bytes = null, key = commitKey))\n+      }.groupBy { case (partition, tombstone) => partition }\n+\n+      // Append the tombstone messages to the offset partitions. It is okay if the replicas don't receive these (say,\n+      // if we crash or leaders move) since the new leaders will get rid of expired offsets during their own purge cycles.\n+      tombstonesForPartition.flatMap { case (offsetsPartition, tombstones) =>\n+        val partitionOpt = replicaManager.getPartition(GroupCoordinator.GroupMetadataTopicName, offsetsPartition)\n+        partitionOpt.map { partition =>\n+          val appendPartition = TopicAndPartition(GroupCoordinator.GroupMetadataTopicName, offsetsPartition)\n+          val messages = tombstones.map(_._2).toSeq\n+\n+          trace(\"Marked %d offsets in %s for deletion.\".format(messages.size, appendPartition))\n+\n+          try {\n+            // do not need to require acks since even if the tombsone is lost,\n+            // it will be appended again in the next purge cycle\n+            partition.appendMessagesToLeader(new ByteBufferMessageSet(config.offsetsTopicCompressionCodec, messages: _*))\n+            tombstones.size\n+          }\n+          catch {\n+            case t: Throwable =>\n+              error(\"Failed to mark %d expired offsets for deletion in %s.\".format(messages.size, appendPartition), t)\n+              // ignore and continue\n+              0\n+          }\n+        }\n+      }.sum\n+    }\n+\n+    info(\"Removed %d expired offsets in %d milliseconds.\".format(numExpiredOffsetsRemoved, SystemTime.milliseconds - startMs))\n+  }\n+\n+  private def getHighWatermark(partitionId: Int): Long = {\n+    val partitionOpt = replicaManager.getPartition(GroupCoordinator.GroupMetadataTopicName, partitionId)\n+\n+    val hw = partitionOpt.map { partition =>\n+      partition.leaderReplicaIfLocal().map(_.highWatermark.messageOffset).getOrElse(-1L)\n+    }.getOrElse(-1L)\n+\n+    hw\n+  }\n+\n+  /*\n+   * Check if the offset metadata length is valid\n+   */\n+  private def validateOffsetMetadataLength(metadata: String) : Boolean = {\n+    metadata == null || metadata.length() <= config.maxMetadataSize\n+  }\n+\n+  def shutdown() {\n+    shuttingDown.set(true)\n+\n+    // TODO: clear the caches\n+  }\n+\n+  /**\n+   * Gets the partition count of the offsets topic from ZooKeeper.\n+   * If the topic does not exist, the configured partition count is returned.\n+   */\n+  private def getOffsetsTopicPartitionCount = {\n+    val topic = GroupCoordinator.GroupMetadataTopicName\n+    val topicData = zkUtils.getPartitionAssignmentForTopics(Seq(topic))\n+    if (topicData(topic).nonEmpty)\n+      topicData(topic).size\n+    else\n+      config.offsetsTopicNumPartitions\n+  }\n+\n+  /**\n+   * Add the partition into the owned list\n+   *\n+   * NOTE: this is for test only\n+   */\n+  def addPartitionOwnership(partition: Int) {\n+    loadingPartitions synchronized {\n+      ownedPartitions.add(partition)\n+    }\n+  }\n+}\n+\n+\n+object GroupMetadataManager {\n+\n+  private val CURRENT_OFFSET_KEY_SCHEMA_VERSION = 1.toShort\n+  private val CURRENT_GROUP_KEY_SCHEMA_VERSION = 2.toShort\n+\n+  private val OFFSET_COMMIT_KEY_SCHEMA = new Schema(new Field(\"group\", STRING),\n+    new Field(\"topic\", STRING),\n+    new Field(\"partition\", INT32))\n+  private val OFFSET_KEY_GROUP_FIELD = OFFSET_COMMIT_KEY_SCHEMA.get(\"group\")\n+  private val OFFSET_KEY_TOPIC_FIELD = OFFSET_COMMIT_KEY_SCHEMA.get(\"topic\")\n+  private val OFFSET_KEY_PARTITION_FIELD = OFFSET_COMMIT_KEY_SCHEMA.get(\"partition\")\n+\n+  private val OFFSET_COMMIT_VALUE_SCHEMA_V0 = new Schema(new Field(\"offset\", INT64),\n+    new Field(\"metadata\", STRING, \"Associated metadata.\", \"\"),\n+    new Field(\"timestamp\", INT64))\n+  private val OFFSET_VALUE_OFFSET_FIELD_V0 = OFFSET_COMMIT_VALUE_SCHEMA_V0.get(\"offset\")\n+  private val OFFSET_VALUE_METADATA_FIELD_V0 = OFFSET_COMMIT_VALUE_SCHEMA_V0.get(\"metadata\")\n+  private val OFFSET_VALUE_TIMESTAMP_FIELD_V0 = OFFSET_COMMIT_VALUE_SCHEMA_V0.get(\"timestamp\")\n+\n+\n+  private val OFFSET_COMMIT_VALUE_SCHEMA_V1 = new Schema(new Field(\"offset\", INT64),\n+    new Field(\"metadata\", STRING, \"Associated metadata.\", \"\"),\n+    new Field(\"commit_timestamp\", INT64),\n+    new Field(\"expire_timestamp\", INT64))\n+  private val OFFSET_VALUE_OFFSET_FIELD_V1 = OFFSET_COMMIT_VALUE_SCHEMA_V1.get(\"offset\")\n+  private val OFFSET_VALUE_METADATA_FIELD_V1 = OFFSET_COMMIT_VALUE_SCHEMA_V1.get(\"metadata\")\n+  private val OFFSET_VALUE_COMMIT_TIMESTAMP_FIELD_V1 = OFFSET_COMMIT_VALUE_SCHEMA_V1.get(\"commit_timestamp\")\n+  private val OFFSET_VALUE_EXPIRE_TIMESTAMP_FIELD_V1 = OFFSET_COMMIT_VALUE_SCHEMA_V1.get(\"expire_timestamp\")\n+\n+  private val GROUP_METADATA_KEY_SCHEMA = new Schema(new Field(\"group\", STRING))\n+  private val GROUP_KEY_GROUP_FIELD = GROUP_METADATA_KEY_SCHEMA.get(\"group\")\n+\n+  private val MEMBER_METADATA_V0 = new Schema(new Field(\"member_id\", STRING),\n+    new Field(\"session_timeout\", INT32),\n+    new Field(\"subscription\", BYTES),\n+    new Field(\"assignment\", BYTES))\n+  private val MEMBER_METADATA_MEMBER_ID_V0 = MEMBER_METADATA_V0.get(\"member_id\")\n+  private val MEMBER_METADATA_SESSION_TIMEOUT_V0 = MEMBER_METADATA_V0.get(\"session_timeout\")\n+  private val MEMBER_METADATA_SUBSCRIPTION_V0 = MEMBER_METADATA_V0.get(\"subscription\")\n+  private val MEMBER_METADATA_ASSIGNMENT_V0 = MEMBER_METADATA_V0.get(\"assignment\")\n+\n+\n+  private val GROUP_METADATA_VALUE_SCHEMA_V0 = new Schema(new Field(\"protocol_type\", STRING),\n+    new Field(\"generation\", INT32),\n+    new Field(\"protocol\", STRING),\n+    new Field(\"leader\", STRING),\n+    new Field(\"members\", new ArrayOf(MEMBER_METADATA_V0)))\n+  private val GROUP_METADATA_PROTOCOL_TYPE_V0 = GROUP_METADATA_VALUE_SCHEMA_V0.get(\"protocol_type\")\n+  private val GROUP_METADATA_GENERATION_V0 = GROUP_METADATA_VALUE_SCHEMA_V0.get(\"generation\")\n+  private val GROUP_METADATA_PROTOCOL_V0 = GROUP_METADATA_VALUE_SCHEMA_V0.get(\"protocol\")\n+  private val GROUP_METADATA_LEADER_V0 = GROUP_METADATA_VALUE_SCHEMA_V0.get(\"leader\")\n+  private val GROUP_METADATA_MEMBERS_V0 = GROUP_METADATA_VALUE_SCHEMA_V0.get(\"members\")\n+\n+  // map of versions to key schemas as data types\n+  private val MESSAGE_TYPE_SCHEMAS = Map(0 -> OFFSET_COMMIT_KEY_SCHEMA,\n+    1 -> OFFSET_COMMIT_KEY_SCHEMA,\n+    2 -> GROUP_METADATA_KEY_SCHEMA)\n+\n+  // map of version of offset value schemas\n+  private val OFFSET_VALUE_SCHEMAS = Map(0 -> OFFSET_COMMIT_VALUE_SCHEMA_V0,\n+    1 -> OFFSET_COMMIT_VALUE_SCHEMA_V1)\n+  private val CURRENT_OFFSET_VALUE_SCHEMA_VERSION = 1.toShort\n+\n+  // map of version of group metadata value schemas\n+  private val GROUP_VALUE_SCHEMAS = Map(0 -> GROUP_METADATA_VALUE_SCHEMA_V0)\n+  private val CURRENT_GROUP_VALUE_SCHEMA_VERSION = 0.toShort\n+\n+  private val CURRENT_OFFSET_KEY_SCHEMA = schemaForKey(CURRENT_OFFSET_KEY_SCHEMA_VERSION)\n+  private val CURRENT_GROUP_KEY_SCHEMA = schemaForKey(CURRENT_GROUP_KEY_SCHEMA_VERSION)\n+\n+  private val CURRENT_OFFSET_VALUE_SCHEMA = schemaForOffset(CURRENT_OFFSET_VALUE_SCHEMA_VERSION)\n+  private val CURRENT_GROUP_VALUE_SCHEMA = schemaForGroup(CURRENT_GROUP_VALUE_SCHEMA_VERSION)\n+\n+  private def schemaForKey(version: Int) = {\n+    val schemaOpt = MESSAGE_TYPE_SCHEMAS.get(version)\n+    schemaOpt match {\n+      case Some(schema) => schema\n+      case _ => throw new KafkaException(\"Unknown offset schema version \" + version)\n+    }\n+  }\n+\n+  private def schemaForOffset(version: Int) = {\n+    val schemaOpt = OFFSET_VALUE_SCHEMAS.get(version)\n+    schemaOpt match {\n+      case Some(schema) => schema\n+      case _ => throw new KafkaException(\"Unknown offset schema version \" + version)\n+    }\n+  }\n+\n+  private def schemaForGroup(version: Int) = {\n+    val schemaOpt = GROUP_VALUE_SCHEMAS.get(version)\n+    schemaOpt match {\n+      case Some(schema) => schema\n+      case _ => throw new KafkaException(\"Unknown offset schema version \" + version)\n+    }\n+  }\n+\n+  /**\n+   * Generates the key for offset commit message for given (group, topic, partition)\n+   *\n+   * @return key for offset commit message\n+   */\n+  private def offsetCommitKey(group: String, topic: String, partition: Int, versionId: Short = 0): Array[Byte] = {\n+    val key = new Struct(CURRENT_OFFSET_KEY_SCHEMA)\n+    key.set(OFFSET_KEY_GROUP_FIELD, group)\n+    key.set(OFFSET_KEY_TOPIC_FIELD, topic)\n+    key.set(OFFSET_KEY_PARTITION_FIELD, partition)\n+\n+    val byteBuffer = ByteBuffer.allocate(2 /* version */ + key.sizeOf)\n+    byteBuffer.putShort(CURRENT_OFFSET_KEY_SCHEMA_VERSION)\n+    key.writeTo(byteBuffer)\n+    byteBuffer.array()\n+  }\n+\n+  /**\n+   * Generates the key for group metadata message for given group\n+   *\n+   * @return key bytes for group metadata message\n+   */\n+  private def groupMetadataKey(group: String): Array[Byte] = {\n+    val key = new Struct(CURRENT_GROUP_KEY_SCHEMA)\n+    key.set(GROUP_KEY_GROUP_FIELD, group)\n+\n+    val byteBuffer = ByteBuffer.allocate(2 /* version */ + key.sizeOf)\n+    byteBuffer.putShort(CURRENT_GROUP_KEY_SCHEMA_VERSION)\n+    key.writeTo(byteBuffer)\n+    byteBuffer.array()\n+  }\n+\n+  /**\n+   * Generates the payload for offset commit message from given offset and metadata\n+   *\n+   * @param offsetAndMetadata consumer's current offset and metadata\n+   * @return payload for offset commit message\n+   */\n+  private def offsetCommitValue(offsetAndMetadata: OffsetAndMetadata): Array[Byte] = {\n+    // generate commit value with schema version 1\n+    val value = new Struct(CURRENT_OFFSET_VALUE_SCHEMA)\n+    value.set(OFFSET_VALUE_OFFSET_FIELD_V1, offsetAndMetadata.offset)\n+    value.set(OFFSET_VALUE_METADATA_FIELD_V1, offsetAndMetadata.metadata)\n+    value.set(OFFSET_VALUE_COMMIT_TIMESTAMP_FIELD_V1, offsetAndMetadata.commitTimestamp)\n+    value.set(OFFSET_VALUE_EXPIRE_TIMESTAMP_FIELD_V1, offsetAndMetadata.expireTimestamp)\n+    val byteBuffer = ByteBuffer.allocate(2 /* version */ + value.sizeOf)\n+    byteBuffer.putShort(CURRENT_OFFSET_KEY_SCHEMA_VERSION)\n+    value.writeTo(byteBuffer)\n+    byteBuffer.array()\n+  }\n+\n+  /**\n+   * Generates the payload for group metadata message from given offset and metadata\n+   * assuming the generation id, selected protocol, leader and member assignment are all available\n+   *\n+   * @param groupMetadata\n+   * @return payload for offset commit message\n+   */\n+  private def groupMetadataValue(groupMetadata: GroupMetadata, assignment: Map[String, Array[Byte]]): Array[Byte] = {\n+    // generate commit value with schema version 1\n+    val value = new Struct(CURRENT_GROUP_VALUE_SCHEMA)\n+    value.set(GROUP_METADATA_PROTOCOL_TYPE_V0, groupMetadata.protocolType)\n+    value.set(GROUP_METADATA_GENERATION_V0, groupMetadata.generationId)\n+    value.set(GROUP_METADATA_PROTOCOL_V0, groupMetadata.protocol)\n+    value.set(GROUP_METADATA_LEADER_V0, groupMetadata.leaderId)\n+\n+    val memberArray = groupMetadata.allMembers.map {\n+      case memberMetadata =>\n+        val memberStruct = value.instance(GROUP_METADATA_MEMBERS_V0)\n+        memberStruct.set(MEMBER_METADATA_MEMBER_ID_V0, memberMetadata.memberId)\n+        memberStruct.set(MEMBER_METADATA_SESSION_TIMEOUT_V0, memberMetadata.sessionTimeoutMs)\n+\n+        val metadata = memberMetadata.metadata(groupMetadata.protocol)\n+        memberStruct.set(MEMBER_METADATA_SUBSCRIPTION_V0, ByteBuffer.wrap(metadata))\n+\n+        val memberAssignment = assignment(memberMetadata.memberId)\n+        assert(memberAssignment != null)\n+\n+        memberStruct.set(MEMBER_METADATA_ASSIGNMENT_V0, ByteBuffer.wrap(memberAssignment))\n+\n+        memberStruct\n+    }\n+\n+    value.set(GROUP_METADATA_MEMBERS_V0, memberArray.toArray)\n+\n+    val byteBuffer = ByteBuffer.allocate(2 /* version */ + value.sizeOf)\n+    byteBuffer.putShort(CURRENT_OFFSET_KEY_SCHEMA_VERSION)\n+    value.writeTo(byteBuffer)\n+    byteBuffer.array()\n+  }\n+\n+  /**\n+   * Decodes the offset messages' key\n+   *\n+   * @param buffer input byte-buffer\n+   * @return an GroupTopicPartition object\n+   */\n+  private def readMessageKey(buffer: ByteBuffer): BaseKey = {\n+    val version = buffer.getShort\n+    val keySchema = schemaForKey(version)\n+    val key = keySchema.read(buffer).asInstanceOf[Struct]\n+\n+    if (version <= CURRENT_OFFSET_KEY_SCHEMA_VERSION) {\n+      // version 0 and 1 refer to offset\n+      val group = key.get(OFFSET_KEY_GROUP_FIELD).asInstanceOf[String]\n+      val topic = key.get(OFFSET_KEY_TOPIC_FIELD).asInstanceOf[String]\n+      val partition = key.get(OFFSET_KEY_PARTITION_FIELD).asInstanceOf[Int]\n+\n+      OffsetKey(version, GroupTopicPartition(group, TopicAndPartition(topic, partition)))\n+\n+    } else if (version == CURRENT_GROUP_KEY_SCHEMA_VERSION) {\n+      // version 2 refers to offset\n+      val group = key.get(GROUP_KEY_GROUP_FIELD).asInstanceOf[String]\n+\n+      GroupKey(version, group)\n+    } else {\n+      throw new IllegalStateException(\"Unknown version \" + version + \" for group metadata message\")\n+    }\n+  }\n+\n+  /**\n+   * Decodes the offset messages' payload and retrieves offset and metadata from it\n+   *\n+   * @param buffer input byte-buffer\n+   * @return an offset-metadata object from the message\n+   */\n+  private def readOffsetMessageValue(buffer: ByteBuffer): OffsetAndMetadata = {\n+    if(buffer == null) { // tombstone\n+      null\n+    } else {\n+      val version = buffer.getShort\n+      val valueSchema = schemaForOffset(version)\n+      val value = valueSchema.read(buffer).asInstanceOf[Struct]\n+\n+      if (version == 0) {\n+        val offset = value.get(OFFSET_VALUE_OFFSET_FIELD_V0).asInstanceOf[Long]\n+        val metadata = value.get(OFFSET_VALUE_METADATA_FIELD_V0).asInstanceOf[String]\n+        val timestamp = value.get(OFFSET_VALUE_TIMESTAMP_FIELD_V0).asInstanceOf[Long]\n+\n+        OffsetAndMetadata(offset, metadata, timestamp)\n+      } else if (version == 1) {\n+        val offset = value.get(OFFSET_VALUE_OFFSET_FIELD_V1).asInstanceOf[Long]\n+        val metadata = value.get(OFFSET_VALUE_METADATA_FIELD_V1).asInstanceOf[String]\n+        val commitTimestamp = value.get(OFFSET_VALUE_COMMIT_TIMESTAMP_FIELD_V1).asInstanceOf[Long]\n+        val expireTimestamp = value.get(OFFSET_VALUE_EXPIRE_TIMESTAMP_FIELD_V1).asInstanceOf[Long]\n+\n+        OffsetAndMetadata(offset, metadata, commitTimestamp, expireTimestamp)\n+      } else {\n+        throw new IllegalStateException(\"Unknown offset message version\")\n+      }\n+    }\n+  }\n+\n+  /**\n+   * Decodes the group metadata messages' payload and retrieves its member metadatafrom it\n+   *\n+   * @param buffer input byte-buffer\n+   * @return a group metadata object from the message\n+   */\n+  private def readGroupMessageValue(groupId: String, buffer: ByteBuffer): GroupMetadata = {\n+    if(buffer == null) { // tombstone\n+      null\n+    } else {\n+      val version = buffer.getShort\n+      val valueSchema = schemaForGroup(version)\n+      val value = valueSchema.read(buffer).asInstanceOf[Struct]\n+\n+      if (version == 0) {\n+        val protocolType = value.get(GROUP_METADATA_PROTOCOL_TYPE_V0).asInstanceOf[String]\n+\n+        val group = new GroupMetadata(groupId, protocolType)\n+\n+        group.generationId = value.get(GROUP_METADATA_GENERATION_V0).asInstanceOf[Int]\n+        group.leaderId = value.get(GROUP_METADATA_LEADER_V0).asInstanceOf[String]\n+        group.protocol = value.get(GROUP_METADATA_PROTOCOL_V0).asInstanceOf[String]\n+\n+        value.getArray(GROUP_METADATA_MEMBERS_V0).foreach {\n+          case memberMetadataObj =>\n+            val memberMetadata = memberMetadataObj.asInstanceOf[Struct]\n+            val memberId = memberMetadata.get(MEMBER_METADATA_MEMBER_ID_V0).asInstanceOf[String]\n+            val sessionTimeout = memberMetadata.get(MEMBER_METADATA_SESSION_TIMEOUT_V0).asInstanceOf[Int]\n+            val subscription = Utils.toArray(memberMetadata.get(MEMBER_METADATA_SUBSCRIPTION_V0).asInstanceOf[ByteBuffer])\n+\n+            val member = new MemberMetadata(memberId, groupId, sessionTimeout, List((group.protocol, subscription)))\n+\n+            member.assignment = Utils.toArray(memberMetadata.get(MEMBER_METADATA_ASSIGNMENT_V0).asInstanceOf[ByteBuffer])\n+\n+            group.add(memberId, member)\n+        }\n+\n+        group\n+      } else {\n+        throw new IllegalStateException(\"Unknown group metadata message version\")\n+      }\n+    }\n+  }\n+\n+  // Formatter for use with tools such as console consumer: Consumer should also set exclude.internal.topics to false.\n+  // (specify --formatter \"kafka.coordinator.GroupMetadataManager\\$OffsetsMessageFormatter\" when consuming __consumer_offsets)\n+  class OffsetsMessageFormatter extends MessageFormatter {\n+    def writeTo(key: Array[Byte], value: Array[Byte], output: PrintStream) {\n+      val formattedKey = if (key == null) \"NULL\" else GroupMetadataManager.readMessageKey(ByteBuffer.wrap(key)).toString\n+      val formattedValue = if (value == null) \"NULL\" else GroupMetadataManager.readOffsetMessageValue(ByteBuffer.wrap(value)).toString","path":"core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala","position":915,"original_position":893,"commit_id":"0e0d67fa9f1d6676af77d322f9dc1fed272493e5","original_commit_id":"2db31f23d695b90e527e904dce869591514f05e9","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":"I was thinking it will be added in the patch of adding the describe-group request; I will coordinate with @hachikuji about this.","created_at":"2015-11-02T19:17:19Z","updated_at":"2015-11-02T19:17:19Z","html_url":"https://github.com/apache/kafka/pull/386#discussion_r43668275","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/386","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/43668275"},"html":{"href":"https://github.com/apache/kafka/pull/386#discussion_r43668275"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/386"}}},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/386","id":49213914,"html_url":"https://github.com/apache/kafka/pull/386","diff_url":"https://github.com/apache/kafka/pull/386.diff","patch_url":"https://github.com/apache/kafka/pull/386.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/386","number":386,"state":"open","locked":false,"title":"KAFKA-2017: Persist Group Metadata and Assignment before Responding","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":"","created_at":"2015-10-29T23:24:33Z","updated_at":"2015-11-02T19:17:19Z","closed_at":null,"merged_at":null,"merge_commit_sha":"4fee73f64f612e18ab0cac6250413a9bf0f8cf74","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/386/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/386/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/386/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/0e0d67fa9f1d6676af77d322f9dc1fed272493e5","head":{"label":"guozhangwang:K2017","ref":"K2017","sha":"0e0d67fa9f1d6676af77d322f9dc1fed272493e5","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},"repo":{"id":39164613,"name":"kafka","full_name":"guozhangwang/kafka","owner":{"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},"private":false,"html_url":"https://github.com/guozhangwang/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/guozhangwang/kafka","forks_url":"https://api.github.com/repos/guozhangwang/kafka/forks","keys_url":"https://api.github.com/repos/guozhangwang/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/guozhangwang/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/guozhangwang/kafka/teams","hooks_url":"https://api.github.com/repos/guozhangwang/kafka/hooks","issue_events_url":"https://api.github.com/repos/guozhangwang/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/guozhangwang/kafka/events","assignees_url":"https://api.github.com/repos/guozhangwang/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/guozhangwang/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/guozhangwang/kafka/tags","blobs_url":"https://api.github.com/repos/guozhangwang/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/guozhangwang/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/guozhangwang/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/guozhangwang/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/guozhangwang/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/guozhangwang/kafka/languages","stargazers_url":"https://api.github.com/repos/guozhangwang/kafka/stargazers","contributors_url":"https://api.github.com/repos/guozhangwang/kafka/contributors","subscribers_url":"https://api.github.com/repos/guozhangwang/kafka/subscribers","subscription_url":"https://api.github.com/repos/guozhangwang/kafka/subscription","commits_url":"https://api.github.com/repos/guozhangwang/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/guozhangwang/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/guozhangwang/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/guozhangwang/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/guozhangwang/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/guozhangwang/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/guozhangwang/kafka/merges","archive_url":"https://api.github.com/repos/guozhangwang/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/guozhangwang/kafka/downloads","issues_url":"https://api.github.com/repos/guozhangwang/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/guozhangwang/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/guozhangwang/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/guozhangwang/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/guozhangwang/kafka/labels{/name}","releases_url":"https://api.github.com/repos/guozhangwang/kafka/releases{/id}","created_at":"2015-07-15T22:40:38Z","updated_at":"2015-10-14T21:53:20Z","pushed_at":"2015-11-02T03:45:18Z","git_url":"git://github.com/guozhangwang/kafka.git","ssh_url":"git@github.com:guozhangwang/kafka.git","clone_url":"https://github.com/guozhangwang/kafka.git","svn_url":"https://github.com/guozhangwang/kafka","homepage":null,"size":26001,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"e0ca8fdd7cf3fcc51a5a1152eacdf585434052d8","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-11-02T16:35:31Z","pushed_at":"2015-11-02T19:08:41Z","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":42287,"stargazers_count":2151,"watchers_count":2151,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1267,"mirror_url":"git://git.apache.org/kafka.git","open_issues_count":100,"forks":1267,"open_issues":100,"watchers":2151,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/386"},"html":{"href":"https://github.com/apache/kafka/pull/386"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/386"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/386/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/386/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/386/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/0e0d67fa9f1d6676af77d322f9dc1fed272493e5"}}}}