text:
package kafka.admin

import java.text.{ParseException, SimpleDateFormat}
import java.util.{Date, Properties}
import javax.xml.datatype.DatatypeFactory
import joptsimple.{OptionParser, OptionSpec}

import kafka.api.{OffsetFetchRequest, OffsetFetchResponse, OffsetRequest, PartitionOffsetRequestInfo}
import kafka.client.ClientUtils
import kafka.common.{OffsetMetadataAndError, TopicAndPartition}
import kafka.consumer.SimpleConsumer
import kafka.utils._
import kafka.utils.Implicits._

import org.I0Itec.zkclient.exception.ZkNoNodeException
import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer, OffsetAndMetadata}
import org.apache.kafka.common.errors.BrokerNotAvailableException
import org.apache.kafka.common.{KafkaException, Node, TopicPartition}
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.security.JaasUtils
import org.apache.kafka.common.security.auth.SecurityProtocol
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.kafka.common.utils.Utils

import scala.collection.JavaConverters._
import scala.collection.{Seq, Set, mutable}

object ConsumerGroupCommand extends Logging {

  def main(args: Array[String]) {
    val opts = new ConsumerGroupCommandOptions(args)

    if (args.length == 0)
      CommandLineUtils.printUsageAndDie(opts.parser, "List all consumer groups, describe a consumer group, delete consumer group info, or reset consumer group offsets.")

    // should have exactly one action
    val actions = Seq(opts.listOpt, opts.describeOpt, opts.deleteOpt, opts.resetOffsetsOpt).count(opts.options.has)
    if (actions != 1)
      CommandLineUtils.printUsageAndDie(opts.parser, "Command must include exactly one action: --list, --describe, --delete, --reset-offsets")

    opts.checkArgs()

    val consumerGroupService = {
      if (opts.useOldConsumer) {
        Console.err.println("Note: This will only show information about consumers that use ZooKeeper (not those using the Java consumer API).")
        new ZkConsumerGroupService(opts)
      } else {
        Console.err.println("Note: This will not show information about old Zookeeper-based consumers.")
        new KafkaConsumerGroupService(opts)
      }
    }

    try {
      if (opts.options.has(opts.listOpt))
        consumerGroupService.listGroups().foreach(println(_))
      else if (opts.options.has(opts.describeOpt))
        consumerGroupService.describeGroup()
      else if (opts.options.has(opts.deleteOpt))
        consumerGroupService.deleteGroups()
      else if (opts.options.has(opts.resetOffsetsOpt)) {
        val offsetsToReset = consumerGroupService.resetOffsets()
        if (opts.options.has(opts.exportOpt)) {
          val exported = consumerGroupService.exportOffsetsToReset(offsetsToReset)
          println(exported)
        } else
          printOffsetsToReset(offsetsToReset)
      }
    } catch {
      case e: Throwable =>
        printError(s"Executing consumer group command failed due to ${e.getMessage}", Some(e))
    } finally {
      consumerGroupService.close()
    }
  }

  val MISSING_COLUMN_VALUE = "-"

  def printError(msg: String, e: Option[Throwable] = None): Unit = {
    println(s"Error: $msg")
    e.foreach(debug("Exception in consumer group command", _))
  }

  def convertTimestamp(timeString: String): java.lang.Long = {
    val datetime: String = timeString match {
      case ts if ts.split("T")(1).contains("+") || ts.split("T")(1).contains("-") || ts.split("T")(1).contains("Z") => ts.toString
      case ts => s"${ts}Z"
    }
    val date = try {
      new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSXXX").parse(datetime)
    } catch {
      case _: ParseException => new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSX").parse(datetime)
    }
    date.getTime
  }

  def printOffsetsToReset(groupAssignmentsToReset: Map[TopicPartition, OffsetAndMetadata]): Unit = {
    println("\n%-30s %-10s %-15s".format("TOPIC", "PARTITION", "NEW-OFFSET"))

    groupAssignmentsToReset.foreach {
      case (consumerAssignment, offsetAndMetadata) =>
        println("%-30s %-10s %-15s".format(
          consumerAssignment.topic,
          consumerAssignment.partition,
          offsetAndMetadata.offset))
    }
  }

  protected case class PartitionAssignmentState(group: String, coordinator: Option[Node], topic: Option[String],
                                                partition: Option[Int], offset: Option[Long], lag: Option[Long],
                                                consumerId: Option[String], host: Option[String],
                                                clientId: Option[String], logEndOffset: Option[Long])

  protected case class MemberAssignmentState(group: String, consumerId: String, host: String, clientId: String,
                                             numPartitions: Int, assignment: List[TopicPartition])

  protected case class GroupState(group: String, coordinator: Node, assignmentStrategy: String, state: String, numMembers: Int)

  sealed trait ConsumerGroupService {

    def listGroups(): List[String]

    private def shouldPrintMemberState(group: String, state: Option[String], numRows: Option[Int]): Boolean = {
      // numRows contains the number of data rows, if any, compiled from the API call in the caller method.
      // if it's undefined or 0, there is no relevant group information to display.
      numRows match {
        case None =>
          // applies to both old and new consumer
          printError(s"The consumer group '$group' does not exist.")
          false
        case Some(num) =>
          opts.useOldConsumer || {
            state match {
              case Some("Dead") =>
                printError(s"Consumer group '$group' does not exist.")
              case Some("Empty") =>
                Console.err.println(s"Consumer group '$group' has no active members.")
              case Some("PreparingRebalance") | Some("CompletingRebalance") =>
                Console.err.println(s"Warning: Consumer group '$group' is rebalancing.")
              case Some("Stable") =>
              case other =>
                // the control should never reach here
                throw new KafkaException(s"Expected a valid consumer group state, but found '${other.getOrElse("NONE")}'.")
            }
            !state.contains("Dead") && num > 0
          }
      }
    }

    private def size(colOpt: Option[Seq[Object]]): Option[Int] = colOpt.map(_.size)

    private def printOffsets(group: String, state: Option[String], assignments: Option[Seq[PartitionAssignmentState]]): Unit = {
      if (shouldPrintMemberState(group, state, size(assignments))) {
        // find proper columns width
        var (maxTopicLen, maxConsumerIdLen, maxHostLen) = (15, 15, 15)
        assignments match {
          case None => // do nothing
          case Some(consumerAssignments) =>
            consumerAssignments.foreach { consumerAssignment =>
              maxTopicLen = Math.max(maxTopicLen, consumerAssignment.topic.getOrElse(MISSING_COLUMN_VALUE).length)
              maxConsumerIdLen = Math.max(maxConsumerIdLen, consumerAssignment.consumerId.getOrElse(MISSING_COLUMN_VALUE).length)
              if (!opts.useOldConsumer)
                maxHostLen = Math.max(maxHostLen, consumerAssignment.host.getOrElse(MISSING_COLUMN_VALUE).length)
            }
        }

        print(s"\n%${-maxTopicLen}s %-10s %-15s %-15s %-15s %${-maxConsumerIdLen}s "
          .format("TOPIC", "PARTITION", "CURRENT-OFFSET", "LOG-END-OFFSET", "LAG", "CONSUMER-ID"))

        if (!opts.useOldConsumer)
          print(s"%${-maxHostLen}s %s".format("HOST", "CLIENT-ID"))
        println()

        assignments match {
          case None => // do nothing
          case Some(consumerAssignments) =>
            consumerAssignments.foreach { consumerAssignment =>
              print(s"%${-maxTopicLen}s %-10s %-15s %-15s %-15s %${-maxConsumerIdLen}s ".format(
                consumerAssignment.topic.getOrElse(MISSING_COLUMN_VALUE), consumerAssignment.partition.getOrElse(MISSING_COLUMN_VALUE),
                consumerAssignment.offset.getOrElse(MISSING_COLUMN_VALUE), consumerAssignment.logEndOffset.getOrElse(MISSING_COLUMN_VALUE),
                consumerAssignment.lag.getOrElse(MISSING_COLUMN_VALUE), consumerAssignment.consumerId.getOrElse(MISSING_COLUMN_VALUE)))
              if (!opts.useOldConsumer)
                print(s"%${-maxHostLen}s %s".format(consumerAssignment.host.getOrElse(MISSING_COLUMN_VALUE),
                  consumerAssignment.clientId.getOrElse(MISSING_COLUMN_VALUE)))
              println()
            }
        }
      }
    }

    private def printMembers(group: String, state: Option[String], assignments: Option[Seq[MemberAssignmentState]], verbose: Boolean): Unit = {
      if (shouldPrintMemberState(group, state, size(assignments))) {
        // find proper columns width
        var (maxConsumerIdLen, maxHostLen, maxClientIdLen) = (15, 15, 15)
        assignments match {
          case None => // do nothing
          case Some(memberAssignments) =>
            memberAssignments.foreach { memberAssignment =>
              maxConsumerIdLen = Math.max(maxConsumerIdLen, memberAssignment.consumerId.length)
              maxHostLen = Math.max(maxHostLen, memberAssignment.host.length)
              maxClientIdLen = Math.max(maxClientIdLen, memberAssignment.clientId.length)
            }
        }

        print(s"\n%${-maxConsumerIdLen}s %${-maxHostLen}s %${-maxClientIdLen}s %-15s "
          .format("CONSUMER-ID", "HOST", "CLIENT-ID", "#PARTITIONS"))
        if (verbose)
          print("%s".format("ASSIGNMENT"))
        println()

        assignments match {
          case None => // do nothing
          case Some(memberAssignments) =>
            memberAssignments.foreach { memberAssignment =>
              print(s"%${-maxConsumerIdLen}s %${-maxHostLen}s %${-maxClientIdLen}s %-15s ".format(
                memberAssignment.consumerId, memberAssignment.host, memberAssignment.clientId, memberAssignment.numPartitions))
              if (verbose) {
                val partitions = memberAssignment.assignment match {
                  case List() => MISSING_COLUMN_VALUE
                  case assignment =>
                    assignment.groupBy(_.topic).map {
                      case (topic, partitionList) => topic + partitionList.map(_.partition).sorted.mkString("(", ",", ")")
                    }.toList.sorted.mkString(", ")
                }
                print("%s".format(partitions))
              }
              println()
            }
        }
      }
    }

    private def printState(group: String, state: GroupState): Unit = {
      // this method is reachable only for the new consumer option (where the given state is always defined)
      if (shouldPrintMemberState(group, Some(state.state), Some(1))) {
        val coordinator = s"${state.coordinator.host}:${state.coordinator.port} (${state.coordinator.idString})"
        val coordinatorColLen = Math.max(25, coordinator.length)
        print(s"\n%${-coordinatorColLen}s %-25s %-20s %s".format("COORDINATOR (ID)", "ASSIGNMENT-STRATEGY", "STATE", "#MEMBERS"))
        print(s"\n%${-coordinatorColLen}s %-25s %-20s %s".format(coordinator, state.assignmentStrategy, state.state, state.numMembers))
        println()
      }
    }

    def describeGroup(): Unit = {
      val group = opts.options.valuesOf(opts.groupOpt).asScala.head
      val membersOptPresent = opts.options.has(opts.membersOpt)
      val stateOptPresent = opts.options.has(opts.stateOpt)
      val offsetsOptPresent = opts.options.has(opts.offsetsOpt)
      val subActions = Seq(membersOptPresent, offsetsOptPresent, stateOptPresent).count(_ == true)

      if (subActions == 0 || offsetsOptPresent) {
        val offsets = collectGroupOffsets()
        printOffsets(group, offsets._1, offsets._2)
      } else if (membersOptPresent) {
        val members = collectGroupMembers(opts.options.has(opts.verboseOpt))
        printMembers(group, members._1, members._2, opts.options.has(opts.verboseOpt))
      } else
        printState(group, collectGroupState())
    }

    def close(): Unit

    protected def opts: ConsumerGroupCommandOptions

    protected def getLogEndOffset(topicPartition: TopicPartition): LogOffsetResult =
      getLogEndOffsets(Seq(topicPartition)).getOrElse(topicPartition, LogOffsetResult.Ignore)

    protected def getLogEndOffsets(topicPartitions: Seq[TopicPartition]): Map[TopicPartition, LogOffsetResult]

    def collectGroupOffsets(): (Option[String], Option[Seq[PartitionAssignmentState]])

    def collectGroupMembers(verbose: Boolean): (Option[String], Option[Seq[MemberAssignmentState]]) = throw new UnsupportedOperationException

    def collectGroupState(): GroupState = throw new UnsupportedOperationException

    protected def collectConsumerAssignment(group: String,
                                            coordinator: Option[Node],
                                            topicPartitions: Seq[TopicPartition],
                                            getPartitionOffset: TopicPartition => Option[Long],
                                            consumerIdOpt: Option[String],
                                            hostOpt: Option[String],
                                            clientIdOpt: Option[String]): Array[PartitionAssignmentState] = {
      if (topicPartitions.isEmpty) {
        Array[PartitionAssignmentState](
          PartitionAssignmentState(group, coordinator, None, None, None, getLag(None, None), consumerIdOpt, hostOpt, clientIdOpt, None)
        )
      }
      else
        describePartitions(group, coordinator, topicPartitions.sortBy(_.partition), getPartitionOffset, consumerIdOpt, hostOpt, clientIdOpt)
    }

    private def getLag(offset: Option[Long], logEndOffset: Option[Long]): Option[Long] =
      offset.filter(_ != -1).flatMap(offset => logEndOffset.map(_ - offset))

    private def describePartitions(group: String,
                                   coordinator: Option[Node],
                                   topicPartitions: Seq[TopicPartition],
                                   getPartitionOffset: TopicPartition => Option[Long],
                                   consumerIdOpt: Option[String],
                                   hostOpt: Option[String],
                                   clientIdOpt: Option[String]): Array[PartitionAssignmentState] = {

      def getDescribePartitionResult(topicPartition: TopicPartition, logEndOffsetOpt: Option[Long]): PartitionAssignmentState = {
        val offset = getPartitionOffset(topicPartition)
        PartitionAssignmentState(group, coordinator, Option(topicPartition.topic), Option(topicPartition.partition), offset,
          getLag(offset, logEndOffsetOpt), consumerIdOpt, hostOpt, clientIdOpt, logEndOffsetOpt)
      }

      getLogEndOffsets(topicPartitions).map {
        logEndOffsetResult =>
          logEndOffsetResult._2 match {
            case LogOffsetResult.LogOffset(logEndOffset) => getDescribePartitionResult(logEndOffsetResult._1, Some(logEndOffset))
            case LogOffsetResult.Unknown => getDescribePartitionResult(logEndOffsetResult._1, None)
            case LogOffsetResult.Ignore => null
          }
      }.toArray
    }

    def resetOffsets(): Map[TopicPartition, OffsetAndMetadata] = throw new UnsupportedOperationException

    def exportOffsetsToReset(assignmentsToReset: Map[TopicPartition, OffsetAndMetadata]): String = throw new UnsupportedOperationException

    def deleteGroups(): Map[String, Errors]
  }

  @deprecated("This class has been deprecated and will be removed in a future release.", "0.11.0.0")
  class ZkConsumerGroupService(val opts: ConsumerGroupCommandOptions) extends ConsumerGroupService {

    private val zkUtils = {
      val zkUrl = opts.options.valueOf(opts.zkConnectOpt)
      ZkUtils(zkUrl, 30000, 30000, JaasUtils.isZkSecurityEnabled)
    }

    def close() {
      zkUtils.close()
    }

    def listGroups(): List[String] = {
      zkUtils.getConsumerGroups().toList
    }

    def deleteGroups(): Map[String, Errors] = {
      if (opts.options.has(opts.groupOpt) && opts.options.has(opts.topicOpt))
        deleteGroupsInfoForTopic()
      else if (opts.options.has(opts.groupOpt))
        deleteGroupsInfo()
      else if (opts.options.has(opts.topicOpt))
        deleteAllGroupsInfoForTopic()

      Map()
    }

    def collectGroupOffsets(): (Option[String], Option[Seq[PartitionAssignmentState]]) = {
      val group = opts.options.valueOf(opts.groupOpt)
      val props = if (opts.options.has(opts.commandConfigOpt)) Utils.loadProps(opts.options.valueOf(opts.commandConfigOpt)) else new Properties()
      val channelSocketTimeoutMs = props.getProperty("channelSocketTimeoutMs", "600").toInt
      val channelRetryBackoffMs = props.getProperty("channelRetryBackoffMsOpt", "300").toInt
      if (!zkUtils.getConsumerGroups().contains(group))
        return (None, None)

      val topics = zkUtils.getTopicsByConsumerGroup(group)
      val topicPartitions = getAllTopicPartitions(topics)
      var groupConsumerIds = zkUtils.getConsumersInGroup(group)

      // mapping of topic partition -> consumer id
      val consumerIdByTopicPartition = topicPartitions.map { topicPartition =>
        val owner = zkUtils.readDataMaybeNull(new ZKGroupTopicDirs(group, topicPartition.topic).consumerOwnerDir + "/" + topicPartition.partition)._1
        topicPartition -> owner.map(o => o.substring(0, o.lastIndexOf('-'))).getOrElse(MISSING_COLUMN_VALUE)
      }.toMap

      // mapping of consumer id -> list of topic partitions
      val consumerTopicPartitions = consumerIdByTopicPartition groupBy{_._2} map {
        case (key, value) => (key, value.unzip._1.toArray) }

      // mapping of consumer id -> list of subscribed topics
      val topicsByConsumerId = zkUtils.getTopicsPerMemberId(group)

      var assignmentRows = topicPartitions.flatMap { topicPartition =>
        val partitionOffsets = getPartitionOffsets(group, List(topicPartition), channelSocketTimeoutMs, channelRetryBackoffMs)
        val consumerId = consumerIdByTopicPartition.get(topicPartition)
        // since consumer id is repeated in client id, leave host and client id empty
        consumerId.foreach(id => groupConsumerIds = groupConsumerIds.filterNot(_ == id))
        collectConsumerAssignment(group, None, List(topicPartition), partitionOffsets.get, consumerId, None, None)
      }

      assignmentRows ++= groupConsumerIds.sortBy(- consumerTopicPartitions.get(_).size).flatMap { consumerId =>
        topicsByConsumerId(consumerId).flatMap { _ =>
          // since consumers with no topic partitions are processed here, we pass empty for topic partitions and offsets
          // since consumer id is repeated in client id, leave host and client id empty
          collectConsumerAssignment(group, None, Array[TopicPartition](), Map[TopicPartition, Option[Long]](), Some(consumerId), None, None)
        }
      }

      (None, Some(assignmentRows))
    }

    private def getAllTopicPartitions(topics: Seq[String]): Seq[TopicPartition] = {
      val topicPartitionMap = zkUtils.getPartitionsForTopics(topics)
      topics.flatMap { topic =>
        val partitions = topicPartitionMap.getOrElse(topic, Seq.empty)
        partitions.map(new TopicPartition(topic, _))
      }
    }

    protected def getLogEndOffsets(topicPartitions: Seq[TopicPartition]): Map[TopicPartition, LogOffsetResult] = {
      topicPartitions.map { topicPartition => (topicPartition,
        zkUtils.getLeaderForPartition(topicPartition.topic, topicPartition.partition) match {
          case Some(-1) => LogOffsetResult.Unknown
          case Some(brokerId) =>
            getZkConsumer(brokerId).map { consumer =>
              val topicAndPartition = new TopicAndPartition(topicPartition)
              val request = OffsetRequest(Map(topicAndPartition -> PartitionOffsetRequestInfo(OffsetRequest.LatestTime, 1)))
              val logEndOffset = consumer.getOffsetsBefore(request).partitionErrorAndOffsets(topicAndPartition).offsets.head
              consumer.close()
              LogOffsetResult.LogOffset(logEndOffset)
            }.getOrElse(LogOffsetResult.Ignore)
          case None =>
            printError(s"No broker for partition '$topicPartition'")
            LogOffsetResult.Ignore
        }
      )}.toMap
    }

    private def getPartitionOffsets(group: String,
                                    topicPartitions: Seq[TopicPartition],
                                    channelSocketTimeoutMs: Int,
                                    channelRetryBackoffMs: Int): Map[TopicPartition, Long] = {
      val offsetMap = mutable.Map[TopicAndPartition, Long]()
      val channel = ClientUtils.channelToOffsetManager(group, zkUtils, channelSocketTimeoutMs, channelRetryBackoffMs)
      channel.send(OffsetFetchRequest(group, topicPartitions.map(new TopicAndPartition(_))))
      val offsetFetchResponse = OffsetFetchResponse.readFrom(channel.receive().payload())

      offsetFetchResponse.requestInfo.foreach { case (topicAndPartition, offsetAndMetadata) =>
        offsetAndMetadata match {
          case OffsetMetadataAndError.NoOffset =>
            val topicDirs = new ZKGroupTopicDirs(group, topicAndPartition.topic)
            // this group may not have migrated off zookeeper for offsets storage (we don't expose the dual-commit option in this tool
            // (meaning the lag may be off until all the consumers in the group have the same setting for offsets storage)
            try {
              val offset = zkUtils.readData(topicDirs.consumerOffsetDir + "/" + topicAndPartition.partition)._1.toLong
              offsetMap.put(topicAndPartition, offset)
            } catch {
              case z: ZkNoNodeException =>
                printError(s"Could not fetch offset from zookeeper for group '$group' partition '$topicAndPartition' due to missing offset data in zookeeper.", Some(z))
            }
          case offsetAndMetaData if offsetAndMetaData.error == Errors.NONE =>
            offsetMap.put(topicAndPartition, offsetAndMetadata.offset)
          case _ =>
            printError(s"Could not fetch offset from kafka for group '$group' partition '$topicAndPartition' due to ${offsetAndMetadata.error.message}.")
        }
      }
      channel.disconnect()
      offsetMap.map { case (topicAndPartition, offset) =>
        (new TopicPartition(topicAndPartition.topic, topicAndPartition.partition), offset)
      }.toMap
    }

    private def deleteGroupsInfo(): Map[String, Errors] = {
      val groups = opts.options.valuesOf(opts.groupOpt)
      groups.asScala.map { group =>
        try {
          if (AdminUtils.deleteConsumerGroupInZK(zkUtils, group)) {
            println(s"Deleted all consumer group information for group '$group' in zookeeper.")
            group -> Errors.NONE
          }
          else {
            printError(s"Delete for group '$group' failed because its consumers are still active.")
            group -> Errors.NON_EMPTY_GROUP
          }
        }
        catch {
          case e: ZkNoNodeException =>
            printError(s"Delete for group '$group' failed because group does not exist.", Some(e))
            group -> Errors.forException(e)
        }
      }.toMap
    }

    private def deleteGroupsInfoForTopic(): Map[String, Errors] = {
      val groups = opts.options.valuesOf(opts.groupOpt)
      val topic = opts.options.valueOf(opts.topicOpt)
      Topic.validate(topic)
      groups.asScala.map { group =>
        try {
          if (AdminUtils.deleteConsumerGroupInfoForTopicInZK(zkUtils, group, topic)) {
            println(s"Deleted consumer group information for group '$group' topic '$topic' in zookeeper.")
            group -> Errors.NONE
          }
          else {
            printError(s"Delete for group '$group' topic '$topic' failed because its consumers are still active.")
            group -> Errors.NON_EMPTY_GROUP
          }
        }
        catch {
          case e: ZkNoNodeException =>
            printError(s"Delete for group '$group' topic '$topic' failed because group does not exist.", Some(e))
            group -> Errors.forException(e)
        }
      }.toMap
    }

    private def deleteAllGroupsInfoForTopic(): Map[String, Errors] = {
      val topic = opts.options.valueOf(opts.topicOpt)
      Topic.validate(topic)
      val deletedGroups = AdminUtils.deleteAllConsumerGroupInfoForTopicInZK(zkUtils, topic)
      println(s"Deleted consumer group information for all inactive consumer groups for topic '$topic' in zookeeper.")
      deletedGroups.map(_ -> Errors.NONE).toMap

    }

    private def getZkConsumer(brokerId: Int): Option[SimpleConsumer] = {
      try {
        zkUtils.getBrokerInfo(brokerId)
          .map(_.brokerEndPoint(ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT)))
          .map(endPoint => new SimpleConsumer(endPoint.host, endPoint.port, 10000, 100000, "ConsumerGroupCommand"))
          .orElse(throw new BrokerNotAvailableException("Broker id %d does not exist".format(brokerId)))
      } catch {
        case t: Throwable =>
          printError(s"Could not parse broker info due to ${t.getMessage}", Some(t))
          None
      }
    }
  }

  class KafkaConsumerGroupService(val opts: ConsumerGroupCommandOptions) extends ConsumerGroupService {

    private val adminClient = createAdminClient()

    // `consumer` is only needed for `describe`, so we instantiate it lazily
    private var consumer: KafkaConsumer[String, String] = _

    override def listGroups(): List[String] = {
      adminClient.listAllConsumerGroupsFlattened().map(_.groupId)
    }

    override def collectGroupOffsets(): (Option[String], Option[Seq[PartitionAssignmentState]]) = {
      val group = opts.options.valueOf(opts.groupOpt)
      val consumerGroupSummary = adminClient.describeConsumerGroup(group, opts.options.valueOf(opts.timeoutMsOpt))
      val assignments = consumerGroupSummary.consumers.map { consumers =>
        var assignedTopicPartitions = Array[TopicPartition]()
        val offsets = adminClient.listGroupOffsets(group)
        val rowsWithConsumer =
          if (offsets.isEmpty)
            List[PartitionAssignmentState]()
          else {
            consumers.filter(_.assignment.nonEmpty).sortWith(_.assignment.size > _.assignment.size).flatMap { consumerSummary =>
              val topicPartitions = consumerSummary.assignment
              assignedTopicPartitions = assignedTopicPartitions ++ consumerSummary.assignment
              val partitionOffsets: Map[TopicPartition, Option[Long]] = consumerSummary.assignment.map { topicPartition =>
                new TopicPartition(topicPartition.topic, topicPartition.partition) -> offsets.get(topicPartition)
              }.toMap
              collectConsumerAssignment(group, Some(consumerGroupSummary.coordinator), topicPartitions,
                partitionOffsets, Some(s"${consumerSummary.consumerId}"), Some(s"${consumerSummary.host}"),
                Some(s"${consumerSummary.clientId}"))
            }
          }

        val rowsWithoutConsumer = offsets.filterKeys(!assignedTopicPartitions.contains(_)).flatMap {
          case (topicPartition, offset) =>
            collectConsumerAssignment(group, Some(consumerGroupSummary.coordinator), Seq(topicPartition),
              Map(topicPartition -> Some(offset)), Some(MISSING_COLUMN_VALUE),
              Some(MISSING_COLUMN_VALUE), Some(MISSING_COLUMN_VALUE))
        }

        rowsWithConsumer ++ rowsWithoutConsumer
      }

      (Some(consumerGroupSummary.state), assignments)
    }

    override def collectGroupMembers(verbose: Boolean): (Option[String], Option[Seq[MemberAssignmentState]]) = {
      val group = opts.options.valueOf(opts.groupOpt)
      val consumerGroupSummary = adminClient.describeConsumerGroup(group, opts.options.valueOf(opts.timeoutMsOpt))
      (Some(consumerGroupSummary.state),
        consumerGroupSummary.consumers.map(_.map {
          consumer => MemberAssignmentState(group, consumer.consumerId, consumer.host, consumer.clientId, consumer.assignment.length,
            if (verbose) consumer.assignment else List())
        })
      )
    }

    override def collectGroupState(): GroupState = {
      val group = opts.options.valueOf(opts.groupOpt)
      val consumerGroupSummary = adminClient.describeConsumerGroup(group, opts.options.valueOf(opts.timeoutMsOpt))
      GroupState(group, consumerGroupSummary.coordinator, consumerGroupSummary.assignmentStrategy,
        consumerGroupSummary.state, consumerGroupSummary.consumers.get.size)
    }

    protected def getLogEndOffsets(topicPartitions: Seq[TopicPartition]): Map[TopicPartition, LogOffsetResult] = {
      val offsets = getConsumer.endOffsets(topicPartitions.asJava)
      topicPartitions.map { topicPartition =>
        val logEndOffset = offsets.get(topicPartition)
        topicPartition -> LogOffsetResult.LogOffset(logEndOffset)
      }.toMap
    }

    protected def getLogStartOffsets(topicPartitions: Seq[TopicPartition]): Map[TopicPartition, LogOffsetResult] = {
      val offsets = getConsumer.beginningOffsets(topicPartitions.asJava)
      topicPartitions.map { topicPartition =>
        val logStartOffset = offsets.get(topicPartition)
        topicPartition -> LogOffsetResult.LogOffset(logStartOffset)
      }.toMap
    }

    protected def getLogTimestampOffsets(topicPartitions: Seq[TopicPartition], timestamp: java.lang.Long): Map[TopicPartition, LogOffsetResult] = {
      val consumer = getConsumer
      consumer.assign(topicPartitions.asJava)

      val (successfulOffsetsForTimes, unsuccessfulOffsetsForTimes) =
        consumer.offsetsForTimes(topicPartitions.map(_ -> timestamp).toMap.asJava).asScala.partition(_._2 != null)

      val successfulLogTimestampOffsets = successfulOffsetsForTimes.map {
        case (topicPartition, offsetAndTimestamp) => topicPartition -> LogOffsetResult.LogOffset(offsetAndTimestamp.offset)
      }.toMap

      successfulLogTimestampOffsets ++ getLogEndOffsets(unsuccessfulOffsetsForTimes.keySet.toSeq)
    }

    def close() {
      adminClient.close()
      if (consumer != null) consumer.close()
    }

    private def createAdminClient(): AdminClient = {
      val props = if (opts.options.has(opts.commandConfigOpt)) Utils.loadProps(opts.options.valueOf(opts.commandConfigOpt)) else new Properties()
      props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, opts.options.valueOf(opts.bootstrapServerOpt))
      AdminClient.create(props)
    }

    private def getConsumer = {
      if (consumer == null)
        consumer = createNewConsumer()
      consumer
    }

    private def createNewConsumer(): KafkaConsumer[String, String] = {
      val properties = new Properties()
      val deserializer = (new StringDeserializer).getClass.getName
      val brokerUrl = opts.options.valueOf(opts.bootstrapServerOpt)
      properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl)
      properties.put(ConsumerConfig.GROUP_ID_CONFIG, opts.options.valueOf(opts.groupOpt))
      properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
      properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000")
      properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, deserializer)
      properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, deserializer)
      if (opts.options.has(opts.commandConfigOpt))
        properties ++= Utils.loadProps(opts.options.valueOf(opts.commandConfigOpt))

      new KafkaConsumer(properties)
    }

    override def resetOffsets(): Map[TopicPartition, OffsetAndMetadata] = {
      val groupId = opts.options.valueOf(opts.groupOpt)
      val consumerGroupSummary = adminClient.describeConsumerGroup(groupId, opts.options.valueOf(opts.timeoutMsOpt))
      consumerGroupSummary.state match {
        case "Empty" | "Dead" =>
          val partitionsToReset = getPartitionsToReset(groupId)
          val preparedOffsets = prepareOffsetsToReset(groupId, partitionsToReset)

          // Dry-run is the default behavior if --execute is not specified
          val dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt)
          if (!dryRun)
            getConsumer.commitSync(preparedOffsets.asJava)
          preparedOffsets
        case currentState =>
          printError(s"Assignments can only be reset if the group '$groupId' is inactive, but the current state is $currentState.")
          Map.empty
      }
    }

    private def parseTopicPartitionsToReset(topicArgs: Seq[String]): Seq[TopicPartition] = topicArgs.flatMap {
      case topicArg if topicArg.contains(":") =>
        val topicPartitions = topicArg.split(":")
        val topic = topicPartitions(0)
        topicPartitions(1).split(",").map(partition => new TopicPartition(topic, partition.toInt))
      case topic => getConsumer.partitionsFor(topic).asScala
        .map(partitionInfo => new TopicPartition(topic, partitionInfo.partition))
    }

    private def getPartitionsToReset(groupId: String): Seq[TopicPartition] = {
      if (opts.options.has(opts.allTopicsOpt)) {
        val allTopicPartitions = adminClient.listGroupOffsets(groupId).keys.toSeq
        allTopicPartitions
      } else if (opts.options.has(opts.topicOpt)) {
        val topics = opts.options.valuesOf(opts.topicOpt).asScala
        parseTopicPartitionsToReset(topics)
      } else {
        if (opts.options.has(opts.resetFromFileOpt))
          Nil
        else
          CommandLineUtils.printUsageAndDie(opts.parser, "One of the reset scopes should be defined: --all-topics, --topic.")
      }
    }

    private def parseResetPlan(resetPlanCsv: String): Map[TopicPartition, OffsetAndMetadata] = {
      resetPlanCsv.split("\n")
        .map { line =>
          val Array(topic, partition, offset) = line.split(",").map(_.trim)
          val topicPartition = new TopicPartition(topic, partition.toInt)
          val offsetAndMetadata = new OffsetAndMetadata(offset.toLong)
          (topicPartition, offsetAndMetadata)
        }.toMap
    }

    private def prepareOffsetsToReset(groupId: String, partitionsToReset: Seq[TopicPartition]): Map[TopicPartition, OffsetAndMetadata] = {
      if (opts.options.has(opts.resetToOffsetOpt)) {
        val offset = opts.options.valueOf(opts.resetToOffsetOpt)
        checkOffsetsRange(partitionsToReset.map((_, offset)).toMap).map {
          case (topicPartition, newOffset) => (topicPartition, new OffsetAndMetadata(newOffset))
        }
      } else if (opts.options.has(opts.resetToEarliestOpt)) {
        val logStartOffsets = getLogStartOffsets(partitionsToReset)
        partitionsToReset.map { topicPartition =>
          logStartOffsets.get(topicPartition) match {
            case Some(LogOffsetResult.LogOffset(offset)) => (topicPartition, new OffsetAndMetadata(offset))
            case _ => CommandLineUtils.printUsageAndDie(opts.parser, s"Error getting starting offset of topic partition: $topicPartition")
          }
        }.toMap
      } else if (opts.options.has(opts.resetToLatestOpt)) {
        val logEndOffsets = getLogEndOffsets(partitionsToReset)
        partitionsToReset.map { topicPartition =>
          logEndOffsets.get(topicPartition) match {
            case Some(LogOffsetResult.LogOffset(offset)) => (topicPartition, new OffsetAndMetadata(offset))
            case _ => CommandLineUtils.printUsageAndDie(opts.parser, s"Error getting ending offset of topic partition: $topicPartition")
          }
        }.toMap
      } else if (opts.options.has(opts.resetShiftByOpt)) {
        val currentCommittedOffsets = adminClient.listGroupOffsets(groupId)
        val requestedOffsets = partitionsToReset.map { topicPartition =>
          val shiftBy = opts.options.valueOf(opts.resetShiftByOpt)
          val currentOffset = currentCommittedOffsets.getOrElse(topicPartition,
            throw new IllegalArgumentException(s"Cannot shift offset for partition $topicPartition since there is no current committed offset"))
          (topicPartition, currentOffset + shiftBy)
        }.toMap
        checkOffsetsRange(requestedOffsets).map {
          case (topicPartition, newOffset) => (topicPartition, new OffsetAndMetadata(newOffset))
        }
      } else if (opts.options.has(opts.resetToDatetimeOpt)) {
        val timestamp = convertTimestamp(opts.options.valueOf(opts.resetToDatetimeOpt))
        val logTimestampOffsets = getLogTimestampOffsets(partitionsToReset, timestamp)
        partitionsToReset.map { topicPartition =>
          val logTimestampOffset = logTimestampOffsets.get(topicPartition)
          logTimestampOffset match {
            case Some(LogOffsetResult.LogOffset(offset)) => (topicPartition, new OffsetAndMetadata(offset))
            case _ => CommandLineUtils.printUsageAndDie(opts.parser, s"Error getting offset by timestamp of topic partition: $topicPartition")
          }
        }.toMap
      } else if (opts.options.has(opts.resetByDurationOpt)) {
        val duration = opts.options.valueOf(opts.resetByDurationOpt)
        val durationParsed = DatatypeFactory.newInstance().newDuration(duration)
        val now = new Date()
        durationParsed.negate().addTo(now)
        val timestamp = now.getTime
        val logTimestampOffsets = getLogTimestampOffsets(partitionsToReset, timestamp)
        partitionsToReset.map { topicPartition =>
          val logTimestampOffset = logTimestampOffsets.get(topicPartition)
          logTimestampOffset match {
            case Some(LogOffsetResult.LogOffset(offset)) => (topicPartition, new OffsetAndMetadata(offset))
            case _ => CommandLineUtils.printUsageAndDie(opts.parser, s"Error getting offset by timestamp of topic partition: $topicPartition")
          }
        }.toMap
      } else if (opts.options.has(opts.resetFromFileOpt)) {
        val resetPlanPath = opts.options.valueOf(opts.resetFromFileOpt)
        val resetPlanCsv = Utils.readFileAsString(resetPlanPath)
        val resetPlan = parseResetPlan(resetPlanCsv)
        val requestedOffsets = resetPlan.keySet.map { topicPartition =>
          (topicPartition, resetPlan(topicPartition).offset())
        }.toMap
        checkOffsetsRange(requestedOffsets).map {
          case (topicPartition, newOffset) => (topicPartition, new OffsetAndMetadata(newOffset))
        }
      } else if (opts.options.has(opts.resetToCurrentOpt)) {
        val currentCommittedOffsets = adminClient.listGroupOffsets(groupId)
        val (partitionsToResetWithCommittedOffset, partitionsToResetWithoutCommittedOffset) =
          partitionsToReset.partition(currentCommittedOffsets.keySet.contains(_))

        val preparedOffsetsForPartitionsWithCommittedOffset = partitionsToResetWithCommittedOffset.map { topicPartition =>
          (topicPartition, new OffsetAndMetadata(currentCommittedOffsets.get(topicPartition) match {
            case Some(offset) => offset
            case _ => throw new IllegalStateException(s"Expected a valid current offset for topic partition: $topicPartition")
          }))
        }.toMap

        val preparedOffsetsForPartitionsWithoutCommittedOffset = getLogEndOffsets(partitionsToResetWithoutCommittedOffset).map {
          case (topicPartition, LogOffsetResult.LogOffset(offset)) => (topicPartition, new OffsetAndMetadata(offset))
          case (topicPartition, _) => CommandLineUtils.printUsageAndDie(opts.parser, s"Error getting ending offset of topic partition: $topicPartition")
        }

        preparedOffsetsForPartitionsWithCommittedOffset ++ preparedOffsetsForPartitionsWithoutCommittedOffset
      } else {
        CommandLineUtils.printUsageAndDie(opts.parser, "Option '%s' requires one of the following scenarios: %s".format(opts.resetOffsetsOpt, opts.allResetOffsetScenarioOpts) )
      }
    }

    private def checkOffsetsRange(requestedOffsets: Map[TopicPartition, Long]) = {
      val logStartOffsets = getLogStartOffsets(requestedOffsets.keySet.toSeq)
      val logEndOffsets = getLogEndOffsets(requestedOffsets.keySet.toSeq)
      requestedOffsets.map { case (topicPartition, offset) => (topicPartition,
        logEndOffsets.get(topicPartition) match {
          case Some(LogOffsetResult.LogOffset(endOffset)) if offset > endOffset =>
            warn(s"New offset ($offset) is higher than latest offset for topic partition $topicPartition. Value will be set to $endOffset")
            endOffset

          case Some(_) => logStartOffsets.get(topicPartition) match {
            case Some(LogOffsetResult.LogOffset(startOffset)) if offset < startOffset =>
              warn(s"New offset ($offset) is lower than earliest offset for topic partition $topicPartition. Value will be set to $startOffset")
              startOffset

            case _ => offset
          }

          case None => // the control should not reach here
            throw new IllegalStateException(s"Unexpected non-existing offset value for topic partition $topicPartition")
        })
      }
    }

    override def exportOffsetsToReset(assignmentsToReset: Map[TopicPartition, OffsetAndMetadata]): String = {
      val rows = assignmentsToReset.map { case (k,v) => s"${k.topic},${k.partition},${v.offset}" }(collection.breakOut): List[String]
      rows.foldRight("")(_ + "\n" + _)
    }

    override def deleteGroups(): Map[String, Errors] = {
      val groupsToDelete = opts.options.valuesOf(opts.groupOpt).asScala.toList
      val result = adminClient.deleteConsumerGroups(groupsToDelete)
      val successfullyDeleted = result.filter {
        case (_, error) => error == Errors.NONE
      }.keySet

      if (successfullyDeleted.size == result.size)
        println(s"Deletion of requested consumer groups (${successfullyDeleted.mkString("'", "', '", "'")}) was successful.")
      else {
        printError("Deletion of some consumer groups failed:")
        result.foreach {
          case (group, error) if error != Errors.NONE => println(s"* Group '$group' could not be deleted due to: ${error.toString}")
          case _ => // no need to print successful deletions individually
        }
        if (successfullyDeleted.nonEmpty)
          println(s"\nThese consumer groups were deleted successfully: ${successfullyDeleted.mkString("'", "', '", "'")}")
      }

      result
    }
  }

  sealed trait LogOffsetResult

  object LogOffsetResult {
    case class LogOffset(value: Long) extends LogOffsetResult
    case object Unknown extends LogOffsetResult
    case object Ignore extends LogOffsetResult
  }

  class ConsumerGroupCommandOptions(args: Array[String]) {
    val ZkConnectDoc = "REQUIRED (for consumer groups based on the old consumer): The connection string for the zookeeper connection in the form host:port. " +
      "Multiple URLS can be given to allow fail-over."
    val BootstrapServerDoc = "REQUIRED (for consumer groups based on the new consumer): The server to connect to."
    val GroupDoc = "The consumer group we wish to act on."
    val TopicDoc = "The topic whose consumer group information should be deleted or topic whose should be included in the reset offset process. " +
      "In `reset-offsets` case, partitions can be specified using this format: `topic1:0,1,2`, where 0,1,2 are the partition to be included in the process. " +
      "Reset-offsets also supports multiple topic inputs."
    val AllTopicsDoc = "Consider all topics assigned to a group in the `reset-offsets` process."
    val ListDoc = "List all consumer groups."
    val DescribeDoc = "Describe consumer group and list offset lag (number of messages not yet processed) related to given group."
    val nl = System.getProperty("line.separator")
    val DeleteDoc = "Pass in groups to delete topic partition offsets and ownership information " +
      "over the entire consumer group. For instance --group g1 --group g2" + nl +
      "Pass in groups with a single topic to just delete the given topic's partition offsets and ownership " +
      "information for the given consumer groups. For instance --group g1 --group g2 --topic t1" + nl +
      "Pass in just a topic to delete the given topic's partition offsets and ownership information " +
      "for every consumer group. For instance --topic t1" + nl +
      "WARNING: Group deletion only works for old ZK-based consumer groups, and one has to use it carefully to only delete groups that are not active."
    val TimeoutMsDoc = "The timeout that can be set for some use cases. For example, it can be used when describing the group " +
      "to specify the maximum amount of time in milliseconds to wait before the group stabilizes (when the group is just created, " +
      "or is going through some changes)."
    val CommandConfigDoc = "Property file containing configs to be passed to Admin Client and Consumer."
    val ResetOffsetsDoc = "Reset offsets of consumer group. Supports one consumer group at the time, and instances should be inactive" + nl +
      "Has 2 execution options: --dry-run (the default) to plan which offsets to reset, and --execute to update the offsets. " +
      "Additionally, the --export option is used to export the results to a CSV format." + nl +
      "You must choose one of the following reset specifications: --to-datetime, --by-period, --to-earliest, " +
      "--to-latest, --shift-by, --from-file, --to-current." + nl +
      "To define the scope use --all-topics or --topic. One scope must be specified unless you use '--from-file'."
    val DryRunDoc = "Only show results without executing changes on Consumer Groups. Supported operations: reset-offsets."
    val ExecuteDoc = "Execute operation. Supported operations: reset-offsets."
    val ExportDoc = "Export operation execution to a CSV file. Supported operations: reset-offsets."
    val ResetToOffsetDoc = "Reset offsets to a specific offset."
    val ResetFromFileDoc = "Reset offsets to values defined in CSV file."
    val ResetToDatetimeDoc = "Reset offsets to offset from datetime. Format: 'YYYY-MM-DDTHH:mm:SS.sss'"
    val ResetByDurationDoc = "Reset offsets to offset by duration from current timestamp. Format: 'PnDTnHnMnS'"
    val ResetToEarliestDoc = "Reset offsets to earliest offset."
    val ResetToLatestDoc = "Reset offsets to latest offset."
    val ResetToCurrentDoc = "Reset offsets to current offset."
    val ResetShiftByDoc = "Reset offsets shifting current offset by 'n', where 'n' can be positive or negative."
    val MembersDoc = "Describe members of the group. This option may be used with '--describe' and '--bootstrap-server' options only." + nl +
      "Example: --bootstrap-server localhost:9092 --describe --group group1 --members"
    val VerboseDoc = "Provide additional information, if any, when describing the group. This option may be used " +
      "with '--offsets'/'--members'/'--state' and '--bootstrap-server' options only." + nl + "Example: --bootstrap-server localhost:9092 --describe --group group1 --members --verbose"
    val OffsetsDoc = "Describe the group and list all topic partitions in the group along with their offset lag. " +
      "This is the default sub-action of and may be used with '--describe' and '--bootstrap-server' options only." + nl +
      "Example: --bootstrap-server localhost:9092 --describe --group group1 --offsets"
    val StateDoc = "Describe the group state. This option may be used with '--describe' and '--bootstrap-server' options only." + nl +
      "Example: --bootstrap-server localhost:9092 --describe --group group1 --state"

    val parser = new OptionParser(false)
    val zkConnectOpt = parser.accepts("zookeeper", ZkConnectDoc)
                             .withRequiredArg
                             .describedAs("urls")
                             .ofType(classOf[String])
    val bootstrapServerOpt = parser.accepts("bootstrap-server", BootstrapServerDoc)
                                   .withRequiredArg
                                   .describedAs("server to connect to")
                                   .ofType(classOf[String])
    val groupOpt = parser.accepts("group", GroupDoc)
                         .withRequiredArg
                         .describedAs("consumer group")
                         .ofType(classOf[String])
    val topicOpt = parser.accepts("topic", TopicDoc)
                         .withRequiredArg
                         .describedAs("topic")
                         .ofType(classOf[String])
    val allTopicsOpt = parser.accepts("all-topics", AllTopicsDoc)
    val listOpt = parser.accepts("list", ListDoc)
    val describeOpt = parser.accepts("describe", DescribeDoc)
    val deleteOpt = parser.accepts("delete", DeleteDoc)
    val timeoutMsOpt = parser.accepts("timeout", TimeoutMsDoc)
                             .withRequiredArg
                             .describedAs("timeout (ms)")
                             .ofType(classOf[Long])
                             .defaultsTo(5000)
    val commandConfigOpt = parser.accepts("command-config", CommandConfigDoc)
                                  .withRequiredArg
                                  .describedAs("command config property file")
                                  .ofType(classOf[String])
    val resetOffsetsOpt = parser.accepts("reset-offsets", ResetOffsetsDoc)
    val dryRunOpt = parser.accepts("dry-run", DryRunDoc)
    val executeOpt = parser.accepts("execute", ExecuteDoc)
    val exportOpt = parser.accepts("export", ExportDoc)
    val resetToOffsetOpt = parser.accepts("to-offset", ResetToOffsetDoc)
                           .withRequiredArg()
                           .describedAs("offset")
                           .ofType(classOf[Long])
    val resetFromFileOpt = parser.accepts("from-file", ResetFromFileDoc)
                                 .withRequiredArg()
                                 .describedAs("path to CSV file")
                                 .ofType(classOf[String])
    val resetToDatetimeOpt = parser.accepts("to-datetime", ResetToDatetimeDoc)
                                   .withRequiredArg()
                                   .describedAs("datetime")
                                   .ofType(classOf[String])
    val resetByDurationOpt = parser.accepts("by-duration", ResetByDurationDoc)
                                   .withRequiredArg()
                                   .describedAs("duration")
                                   .ofType(classOf[String])
    val resetToEarliestOpt = parser.accepts("to-earliest", ResetToEarliestDoc)
    val resetToLatestOpt = parser.accepts("to-latest", ResetToLatestDoc)
    val resetToCurrentOpt = parser.accepts("to-current", ResetToCurrentDoc)
    val resetShiftByOpt = parser.accepts("shift-by", ResetShiftByDoc)
                             .withRequiredArg()
                             .describedAs("number-of-offsets")
                             .ofType(classOf[Long])
    val membersOpt = parser.accepts("members", MembersDoc)
                           .availableIf(describeOpt)
                           .availableUnless(zkConnectOpt)
    val verboseOpt = parser.accepts("verbose", VerboseDoc)
                           .availableIf(describeOpt)
                           .availableUnless(zkConnectOpt)
    val offsetsOpt = parser.accepts("offsets", OffsetsDoc)
                           .availableIf(describeOpt)
                           .availableUnless(zkConnectOpt)
    val stateOpt = parser.accepts("state", StateDoc)
                         .availableIf(describeOpt)
                         .availableUnless(zkConnectOpt)
    parser.mutuallyExclusive(membersOpt, offsetsOpt, stateOpt)

    val options = parser.parse(args : _*)

    val useOldConsumer = options.has(zkConnectOpt)
    val describeOptPresent = options.has(describeOpt)

    val allConsumerGroupLevelOpts: Set[OptionSpec[_]] = Set(listOpt, describeOpt, deleteOpt, resetOffsetsOpt)
    val allResetOffsetScenarioOpts: Set[OptionSpec[_]] = Set(resetToOffsetOpt, resetShiftByOpt,
      resetToDatetimeOpt, resetByDurationOpt, resetToEarliestOpt, resetToLatestOpt, resetToCurrentOpt, resetFromFileOpt)

    def checkArgs() {
      // check required args
      if (options.has(timeoutMsOpt) && (!describeOptPresent || useOldConsumer))
        debug(s"Option $timeoutMsOpt is applicable only when both $bootstrapServerOpt and $describeOpt are used.")

      if (useOldConsumer) {
        if (options.has(bootstrapServerOpt))
          CommandLineUtils.printUsageAndDie(parser, s"Option $bootstrapServerOpt is not valid with $zkConnectOpt.")
      } else {
        CommandLineUtils.checkRequiredArgs(parser, options, bootstrapServerOpt)

        if (options.has(deleteOpt) && options.has(topicOpt))
          CommandLineUtils.printUsageAndDie(parser, s"When deleting a consumer group the option $topicOpt is only " +
            s"valid with $zkConnectOpt. The new consumer does not support topic-specific offset deletion from a consumer group.")
      }

      if (describeOptPresent)
        CommandLineUtils.checkRequiredArgs(parser, options, groupOpt)

      if (options.has(deleteOpt) && !options.has(groupOpt) && !options.has(topicOpt))
        CommandLineUtils.printUsageAndDie(parser, s"Option $deleteOpt either takes $groupOpt, $topicOpt, or both")

      if (options.has(resetOffsetsOpt)) {
        if (options.has(dryRunOpt) && options.has(executeOpt))
          CommandLineUtils.printUsageAndDie(parser, s"Option $resetOffsetsOpt only accepts one of $executeOpt and $dryRunOpt")

        if (!options.has(dryRunOpt) && !options.has(executeOpt)) {
          Console.err.println("WARN: No action will be performed as the --execute option is missing." +
            "In a future major release, the default behavior of this command will be to prompt the user before " +
            "executing the reset rather than doing a dry run. You should add the --dry-run option explicitly " +
            "if you are scripting this command and want to keep the current default behavior without prompting.")
        }

        CommandLineUtils.checkRequiredArgs(parser, options, groupOpt)
        CommandLineUtils.checkInvalidArgs(parser, options, resetToOffsetOpt, allResetOffsetScenarioOpts - resetToOffsetOpt)
        CommandLineUtils.checkInvalidArgs(parser, options, resetToDatetimeOpt, allResetOffsetScenarioOpts - resetToDatetimeOpt)
        CommandLineUtils.checkInvalidArgs(parser, options, resetByDurationOpt, allResetOffsetScenarioOpts - resetByDurationOpt)
        CommandLineUtils.checkInvalidArgs(parser, options, resetToEarliestOpt, allResetOffsetScenarioOpts - resetToEarliestOpt)
        CommandLineUtils.checkInvalidArgs(parser, options, resetToLatestOpt, allResetOffsetScenarioOpts - resetToLatestOpt)
        CommandLineUtils.checkInvalidArgs(parser, options, resetToCurrentOpt, allResetOffsetScenarioOpts - resetToCurrentOpt)
        CommandLineUtils.checkInvalidArgs(parser, options, resetShiftByOpt, allResetOffsetScenarioOpts - resetShiftByOpt)
        CommandLineUtils.checkInvalidArgs(parser, options, resetFromFileOpt, allResetOffsetScenarioOpts - resetFromFileOpt)
      }

      // check invalid args
      CommandLineUtils.checkInvalidArgs(parser, options, groupOpt, allConsumerGroupLevelOpts - describeOpt - deleteOpt - resetOffsetsOpt)
      CommandLineUtils.checkInvalidArgs(parser, options, topicOpt, allConsumerGroupLevelOpts - deleteOpt - resetOffsetsOpt)
    }
  }
}


text:
#include "Test.h"

#include "GrBackendSemaphore.h"
#include "GrClip.h"
#include "GrContextPriv.h"
#include "GrDefaultGeoProcFactory.h"
#include "GrOnFlushResourceProvider.h"
#include "GrProxyProvider.h"
#include "GrQuad.h"
#include "GrRenderTargetContextPriv.h"
#include "GrResourceProvider.h"
#include "GrTexture.h"

#include "SkBitmap.h"
#include "SkPointPriv.h"
#include "effects/generated/GrSimpleTextureEffect.h"
#include "ops/GrSimpleMeshDrawOpHelper.h"

namespace {
// This is a simplified mesh drawing op that can be used in the atlas generation test.
// Please see AtlasedRectOp below.
class NonAARectOp : public GrMeshDrawOp {
protected:
    using Helper = GrSimpleMeshDrawOpHelper;

public:
    DEFINE_OP_CLASS_ID

    // This creates an instance of a simple non-AA solid color rect-drawing Op
    static std::unique_ptr<GrDrawOp> Make(GrRecordingContext* context,
                                          GrPaint&& paint,
                                          const SkRect& r) {
        return Helper::FactoryHelper<NonAARectOp>(context, std::move(paint), r, nullptr, ClassID());
    }

    // This creates an instance of a simple non-AA textured rect-drawing Op
    static std::unique_ptr<GrDrawOp> Make(GrRecordingContext* context,
                                          GrPaint&& paint,
                                          const SkRect& r,
                                          const SkRect& local) {
        return Helper::FactoryHelper<NonAARectOp>(context, std::move(paint), r, &local, ClassID());
    }

    const SkPMColor4f& color() const { return fColor; }

    NonAARectOp(const Helper::MakeArgs& helperArgs, const SkPMColor4f& color, const SkRect& r,
                const SkRect* localRect, int32_t classID)
            : INHERITED(classID)
            , fColor(color)
            , fHasLocalRect(SkToBool(localRect))
            , fRect(r)
            , fHelper(helperArgs, GrAAType::kNone) {
        if (fHasLocalRect) {
            fLocalQuad = GrQuad(*localRect);
        }
        // Choose some conservative values for aa bloat and zero area.
        this->setBounds(r, HasAABloat::kYes, IsZeroArea::kYes);
    }

    const char* name() const override { return "NonAARectOp"; }

    void visitProxies(const VisitProxyFunc& func, VisitorType) const override {
        fHelper.visitProxies(func);
    }

    FixedFunctionFlags fixedFunctionFlags() const override { return FixedFunctionFlags::kNone; }

    GrProcessorSet::Analysis finalize(const GrCaps& caps, const GrAppliedClip*,
                                      GrFSAAType fsaaType, GrClampType clampType) override {
        // Set the color to unknown because the subclass may change the color later.
        GrProcessorAnalysisColor gpColor;
        gpColor.setToUnknown();
        // We ignore the clip so pass this rather than the GrAppliedClip param.
        static GrAppliedClip kNoClip;
        return fHelper.finalizeProcessors(
                caps, &kNoClip, fsaaType, clampType, GrProcessorAnalysisCoverage::kNone, &gpColor);
    }

protected:
    SkPMColor4f fColor;
    bool        fHasLocalRect;
    GrQuad      fLocalQuad;
    SkRect      fRect;

private:
    void onPrepareDraws(Target* target) override {
        using namespace GrDefaultGeoProcFactory;

        // The vertex attrib order is always pos, color, local coords.
        static const int kColorOffset = sizeof(SkPoint);
        static const int kLocalOffset = sizeof(SkPoint) + sizeof(GrColor);

        sk_sp<GrGeometryProcessor> gp =
                GrDefaultGeoProcFactory::Make(target->caps().shaderCaps(),
                                              Color::kPremulGrColorAttribute_Type,
                                              Coverage::kSolid_Type,
                                              fHasLocalRect ? LocalCoords::kHasExplicit_Type
                                                            : LocalCoords::kUnused_Type,
                                              SkMatrix::I());
        if (!gp) {
            SkDebugf("Couldn't create GrGeometryProcessor for GrAtlasedOp\n");
            return;
        }

        size_t vertexStride = gp->vertexStride();

        sk_sp<const GrBuffer> indexBuffer;
        int firstIndex;
        uint16_t* indices = target->makeIndexSpace(6, &indexBuffer, &firstIndex);
        if (!indices) {
            SkDebugf("Indices could not be allocated for GrAtlasedOp.\n");
            return;
        }

        sk_sp<const GrBuffer> vertexBuffer;
        int firstVertex;
        void* vertices = target->makeVertexSpace(vertexStride, 4, &vertexBuffer, &firstVertex);
        if (!vertices) {
            SkDebugf("Vertices could not be allocated for GrAtlasedOp.\n");
            return;
        }

        // Setup indices
        indices[0] = 0;
        indices[1] = 1;
        indices[2] = 2;
        indices[3] = 2;
        indices[4] = 1;
        indices[5] = 3;

        // Setup positions
        SkPoint* position = (SkPoint*) vertices;
        SkPointPriv::SetRectTriStrip(position, fRect, vertexStride);

        // Setup vertex colors
        GrColor* color = (GrColor*)((intptr_t)vertices + kColorOffset);
        for (int i = 0; i < 4; ++i) {
            *color = fColor.toBytes_RGBA();
            color = (GrColor*)((intptr_t)color + vertexStride);
        }

        // Setup local coords
        if (fHasLocalRect) {
            SkPoint* coords = (SkPoint*)((intptr_t) vertices + kLocalOffset);
            for (int i = 0; i < 4; i++) {
                *coords = fLocalQuad.point(i);
                coords = (SkPoint*)((intptr_t) coords + vertexStride);
            }
        }

        GrMesh* mesh = target->allocMesh(GrPrimitiveType::kTriangles);
        mesh->setIndexed(indexBuffer, 6, firstIndex, 0, 3, GrPrimitiveRestart::kNo);
        mesh->setVertexData(vertexBuffer, firstVertex);

        target->recordDraw(std::move(gp), mesh);
    }

    void onExecute(GrOpFlushState* flushState, const SkRect& chainBounds) override {
        fHelper.executeDrawsAndUploads(this, flushState, chainBounds);
    }

    Helper fHelper;

    typedef GrMeshDrawOp INHERITED;
};

}  // anonymous namespace

static constexpr SkRect kEmptyRect = SkRect::MakeEmpty();

namespace {

/*
 * Atlased ops just draw themselves as textured rects with the texture pixels being
 * pulled out of the atlas. Their color is based on their ID.
 */
class AtlasedRectOp final : public NonAARectOp {
public:
    DEFINE_OP_CLASS_ID

    ~AtlasedRectOp() override {
        fID = -1;
    }

    const char* name() const override { return "AtlasedRectOp"; }

    int id() const { return fID; }

    static std::unique_ptr<AtlasedRectOp> Make(GrContext* context,
                                               GrPaint&& paint,
                                               const SkRect& r,
                                               int id) {
        GrDrawOp* op = Helper::FactoryHelper<AtlasedRectOp>(context, std::move(paint),
                                                            r, id).release();
        return std::unique_ptr<AtlasedRectOp>(static_cast<AtlasedRectOp*>(op));
    }

    // We set the initial color of the NonAARectOp based on the ID.
    // Note that we force creation of a NonAARectOp that has local coords in anticipation of
    // pulling from the atlas.
    AtlasedRectOp(const Helper::MakeArgs& helperArgs, const SkPMColor4f& color, const SkRect& r,
                  int id)
            : INHERITED(helperArgs, SkPMColor4f::FromBytes_RGBA(kColors[id]), r, &kEmptyRect,
                        ClassID())
            , fID(id)
            , fNext(nullptr) {
        SkASSERT(fID < kMaxIDs);
    }

    void setColor(const SkPMColor4f& color) { fColor = color; }
    void setLocalRect(const SkRect& localRect) {
        SkASSERT(fHasLocalRect);    // This should've been created to anticipate this
        fLocalQuad = GrQuad(localRect);
    }

    AtlasedRectOp* next() const { return fNext; }
    void setNext(AtlasedRectOp* next) {
        fNext = next;
    }

private:

    static const int kMaxIDs = 9;
    static const GrColor kColors[kMaxIDs];

    int            fID;
    // The Atlased ops have an internal singly-linked list of ops that land in the same opList
    AtlasedRectOp* fNext;

    typedef NonAARectOp INHERITED;
};

}  // anonymous namespace

const GrColor AtlasedRectOp::kColors[kMaxIDs] = {
    GrColorPackRGBA(255, 0, 0, 255),
    GrColorPackRGBA(0, 255, 0, 255),
    GrColorPackRGBA(0, 0, 255, 255),
    GrColorPackRGBA(0, 255, 255, 255),
    GrColorPackRGBA(255, 0, 255, 255),
    GrColorPackRGBA(255, 255, 0, 255),
    GrColorPackRGBA(0, 0, 0, 255),
    GrColorPackRGBA(128, 128, 128, 255),
    GrColorPackRGBA(255, 255, 255, 255)
};

static const int kDrawnTileSize = 16;

/*
 * Rather than performing any rect packing, this atlaser just lays out constant-sized
 * tiles in an Nx1 row
 */
static const int kAtlasTileSize = 2;

/*
 * This class aggregates the op information required for atlasing
 */
class AtlasObject final : public GrOnFlushCallbackObject {
public:
    AtlasObject() : fDone(false) { }

    ~AtlasObject() override {
        SkASSERT(fDone);
    }

    void markAsDone() {
        fDone = true;
    }

    // Insert the new op in an internal singly-linked list for 'opListID'
    void addOp(uint32_t opListID, AtlasedRectOp* op) {
        LinkedListHeader* header = nullptr;
        for (int i = 0; i < fOps.count(); ++i) {
            if (opListID == fOps[i].fID) {
                header = &(fOps[i]);
            }
        }

        if (!header) {
            fOps.push_back({opListID, nullptr});
            header = &(fOps[fOps.count()-1]);
        }

        op->setNext(header->fHead);
        header->fHead = op;
    }

    int numOps() const { return fOps.count(); }

    // Get the fully lazy proxy that is backing the atlas. Its actual width isn't
    // known until flush time.
    sk_sp<GrTextureProxy> getAtlasProxy(GrProxyProvider* proxyProvider, const GrCaps* caps) {
        if (fAtlasProxy) {
            return fAtlasProxy;
        }

        const GrBackendFormat format = caps->getBackendFormatFromColorType(kRGBA_8888_SkColorType);

        fAtlasProxy = GrProxyProvider::MakeFullyLazyProxy(
                [](GrResourceProvider* resourceProvider)
                        -> GrSurfaceProxy::LazyInstantiationResult {
                    GrSurfaceDesc desc;
                    desc.fFlags = kRenderTarget_GrSurfaceFlag;
                    // TODO: until partial flushes in MDB lands we're stuck having
                    // all 9 atlas draws occur
                    desc.fWidth = 9 /*this->numOps()*/ * kAtlasTileSize;
                    desc.fHeight = kAtlasTileSize;
                    desc.fConfig = kRGBA_8888_GrPixelConfig;

                    auto texture = resourceProvider->createTexture(
                            desc, SkBudgeted::kYes, GrResourceProvider::Flags::kNoPendingIO);
                    return std::move(texture);
                },
                format,
                GrProxyProvider::Renderable::kYes,
                kBottomLeft_GrSurfaceOrigin,
                kRGBA_8888_GrPixelConfig,
                *proxyProvider->caps());
        return fAtlasProxy;
    }

    /*
     * This callback creates the atlas and updates the AtlasedRectOps to read from it
     */
    void preFlush(GrOnFlushResourceProvider* resourceProvider,
                  const uint32_t* opListIDs, int numOpListIDs,
                  SkTArray<sk_sp<GrRenderTargetContext>>* results) override {
        SkASSERT(!results->count());

        // Until MDB is landed we will most-likely only have one opList.
        SkTDArray<LinkedListHeader*> lists;
        for (int i = 0; i < numOpListIDs; ++i) {
            if (LinkedListHeader* list = this->getList(opListIDs[i])) {
                lists.push_back(list);
            }
        }

        if (!lists.count()) {
            return; // nothing to atlas
        }

        if (!resourceProvider->instatiateProxy(fAtlasProxy.get())) {
            return;
        }

        // At this point all the GrAtlasedOp's should have lined up to read from 'atlasDest' and
        // there should either be two writes to clear it or no writes.
        SkASSERT(9 == fAtlasProxy->getPendingReadCnt_TestOnly());
        SkASSERT(2 == fAtlasProxy->getPendingWriteCnt_TestOnly() ||
                 0 == fAtlasProxy->getPendingWriteCnt_TestOnly());
        sk_sp<GrRenderTargetContext> rtc = resourceProvider->makeRenderTargetContext(
                                                                           fAtlasProxy,
                                                                           nullptr, nullptr);

        // clear the atlas
        rtc->clear(nullptr, SK_PMColor4fTRANSPARENT,
                   GrRenderTargetContext::CanClearFullscreen::kYes);

        int blocksInAtlas = 0;
        for (int i = 0; i < lists.count(); ++i) {
            for (AtlasedRectOp* op = lists[i]->fHead; op; op = op->next()) {
                SkIRect r = SkIRect::MakeXYWH(blocksInAtlas*kAtlasTileSize, 0,
                                              kAtlasTileSize, kAtlasTileSize);

                // For now, we avoid the resource buffer issues and just use clears
#if 1
                rtc->clear(&r, op->color(), GrRenderTargetContext::CanClearFullscreen::kNo);
#else
                GrPaint paint;
                paint.setColor4f(op->color());
                std::unique_ptr<GrDrawOp> drawOp(NonAARectOp::Make(std::move(paint),
                                                                   SkRect::Make(r)));
                rtc->priv().testingOnly_addDrawOp(std::move(drawOp));
#endif
                blocksInAtlas++;

                // Set the atlased Op's color to white (so we know we're not using it for
                // the final draw).
                op->setColor(SK_PMColor4fWHITE);

                // Set the atlased Op's localRect to point to where it landed in the atlas
                op->setLocalRect(SkRect::Make(r));
            }

            // We've updated all these ops and we certainly don't want to process them again
            this->clearOpsFor(lists[i]);
        }

        results->push_back(std::move(rtc));
    }

private:
    typedef struct {
        uint32_t       fID;
        AtlasedRectOp* fHead;
    } LinkedListHeader;

    LinkedListHeader* getList(uint32_t opListID) {
        for (int i = 0; i < fOps.count(); ++i) {
            if (opListID == fOps[i].fID) {
                return &(fOps[i]);
            }
        }
        return nullptr;
    }

    void clearOpsFor(LinkedListHeader* header) {
        // The AtlasedRectOps have yet to execute (and this class doesn't own them) so just
        // forget about them in the laziest way possible.
        header->fHead = nullptr;
        header->fID = 0;            // invalid opList ID
    }

    // Each opList containing AtlasedRectOps gets its own internal singly-linked list
    SkTDArray<LinkedListHeader>  fOps;

    // The fully lazy proxy for the atlas
    sk_sp<GrTextureProxy>        fAtlasProxy;

    // Set to true when the testing harness expects this object to be no longer used
    bool                         fDone;
};

// This creates an off-screen rendertarget whose ops which eventually pull from the atlas.
static sk_sp<GrTextureProxy> make_upstream_image(GrContext* context, AtlasObject* object, int start,
                                                 sk_sp<GrTextureProxy> atlasProxy) {
    const GrBackendFormat format =
            context->priv().caps()->getBackendFormatFromColorType(kRGBA_8888_SkColorType);

    sk_sp<GrRenderTargetContext> rtc(context->priv().makeDeferredRenderTargetContext(
                                                                      format,
                                                                      SkBackingFit::kApprox,
                                                                      3*kDrawnTileSize,
                                                                      kDrawnTileSize,
                                                                      kRGBA_8888_GrPixelConfig,
                                                                      nullptr));

    rtc->clear(nullptr, { 1, 0, 0, 1 }, GrRenderTargetContext::CanClearFullscreen::kYes);

    for (int i = 0; i < 3; ++i) {
        SkRect r = SkRect::MakeXYWH(i*kDrawnTileSize, 0, kDrawnTileSize, kDrawnTileSize);

        auto fp = GrSimpleTextureEffect::Make(atlasProxy, SkMatrix::I());
        GrPaint paint;
        paint.addColorFragmentProcessor(std::move(fp));
        paint.setPorterDuffXPFactory(SkBlendMode::kSrc);
        std::unique_ptr<AtlasedRectOp> op(AtlasedRectOp::Make(context,
                                                              std::move(paint), r, start + i));

        AtlasedRectOp* sparePtr = op.get();

        uint32_t opListID;
        rtc->priv().testingOnly_addDrawOp(GrNoClip(), std::move(op),
                                          [&opListID](GrOp* op, uint32_t id) { opListID = id; });
        SkASSERT(SK_InvalidUniqueID != opListID);

        object->addOp(opListID, sparePtr);
    }

    return rtc->asTextureProxyRef();
}

// Enable this if you want to debug the final draws w/o having the atlasCallback create the
// atlas
#if 0
#include "SkGrPriv.h"
#include "SkImageEncoder.h"
#include "ToolUtils.h"

static void save_bm(const SkBitmap& bm, const char name[]) {
    bool result = ToolUtils::EncodeImageToFile(name, bm, SkEncodedImageFormat::kPNG, 100);
    SkASSERT(result);
}

sk_sp<GrTextureProxy> pre_create_atlas(GrContext* context) {
    SkBitmap bm;
    bm.allocN32Pixels(18, 2, true);
    bm.erase(SK_ColorRED,     SkIRect::MakeXYWH(0, 0, 2, 2));
    bm.erase(SK_ColorGREEN,   SkIRect::MakeXYWH(2, 0, 2, 2));
    bm.erase(SK_ColorBLUE,    SkIRect::MakeXYWH(4, 0, 2, 2));
    bm.erase(SK_ColorCYAN,    SkIRect::MakeXYWH(6, 0, 2, 2));
    bm.erase(SK_ColorMAGENTA, SkIRect::MakeXYWH(8, 0, 2, 2));
    bm.erase(SK_ColorYELLOW,  SkIRect::MakeXYWH(10, 0, 2, 2));
    bm.erase(SK_ColorBLACK,   SkIRect::MakeXYWH(12, 0, 2, 2));
    bm.erase(SK_ColorGRAY,    SkIRect::MakeXYWH(14, 0, 2, 2));
    bm.erase(SK_ColorWHITE,   SkIRect::MakeXYWH(16, 0, 2, 2));

#if 1
    save_bm(bm, "atlas-fake.png");
#endif

    GrSurfaceDesc desc = GrImageInfoToSurfaceDesc(bm.info());
    desc.fFlags |= kRenderTarget_GrSurfaceFlag;

    sk_sp<GrSurfaceProxy> tmp = GrSurfaceProxy::MakeDeferred(*context->caps(),
                                                             context->textureProvider(),
                                                             desc, SkBudgeted::kYes,
                                                             bm.getPixels(), bm.rowBytes());

    return sk_ref_sp(tmp->asTextureProxy());
}
#endif

static void test_color(skiatest::Reporter* reporter, const SkBitmap& bm, int x, SkColor expected) {
    SkColor readback = bm.getColor(x, kDrawnTileSize/2);
    REPORTER_ASSERT(reporter, expected == readback);
    if (expected != readback) {
        SkDebugf("Color mismatch: %x %x\n", expected, readback);
    }
}

/*
 * For the atlasing test we make a DAG that looks like:
 *
 *    RT1 with ops: 0,1,2       RT2 with ops: 3,4,5       RT3 with ops: 6,7,8
 *                     \         /
 *                      \       /
 *                         RT4
 * We then flush RT4 and expect only ops 0-5 to be atlased together.
 * Each op is just a solid colored rect so both the atlas and the final image should appear as:
 *           R G B C M Y
 * with the atlas having width = 6*kAtlasTileSize and height = kAtlasTileSize.
 *
 * Note: until partial flushes in MDB lands, the atlas will actually have width= 9*kAtlasTileSize
 * and look like:
 *           R G B C M Y K Grey White
 */
DEF_GPUTEST_FOR_GL_RENDERING_CONTEXTS(OnFlushCallbackTest, reporter, ctxInfo) {
    static const int kNumProxies = 3;

    GrContext* context = ctxInfo.grContext();
    auto proxyProvider = context->priv().proxyProvider();

    AtlasObject object;

    context->priv().addOnFlushCallbackObject(&object);

    sk_sp<GrTextureProxy> proxies[kNumProxies];
    for (int i = 0; i < kNumProxies; ++i) {
        proxies[i] = make_upstream_image(context, &object, i*3,
                                         object.getAtlasProxy(proxyProvider,
                                                              context->priv().caps()));
    }

    static const int kFinalWidth = 6*kDrawnTileSize;
    static const int kFinalHeight = kDrawnTileSize;

    const GrBackendFormat format =
            context->priv().caps()->getBackendFormatFromColorType(kRGBA_8888_SkColorType);

    sk_sp<GrRenderTargetContext> rtc(context->priv().makeDeferredRenderTargetContext(
                                                                      format,
                                                                      SkBackingFit::kApprox,
                                                                      kFinalWidth,
                                                                      kFinalHeight,
                                                                      kRGBA_8888_GrPixelConfig,
                                                                      nullptr));

    rtc->clear(nullptr, SK_PMColor4fWHITE, GrRenderTargetContext::CanClearFullscreen::kYes);

    // Note that this doesn't include the third texture proxy
    for (int i = 0; i < kNumProxies-1; ++i) {
        SkRect r = SkRect::MakeXYWH(i*3*kDrawnTileSize, 0, 3*kDrawnTileSize, kDrawnTileSize);

        SkMatrix t = SkMatrix::MakeTrans(-i*3*kDrawnTileSize, 0);

        GrPaint paint;
        auto fp = GrSimpleTextureEffect::Make(std::move(proxies[i]), t);
        paint.setPorterDuffXPFactory(SkBlendMode::kSrc);
        paint.addColorFragmentProcessor(std::move(fp));

        rtc->drawRect(GrNoClip(), std::move(paint), GrAA::kNo, SkMatrix::I(), r);
    }

    rtc->flush(SkSurface::BackendSurfaceAccess::kNoAccess, GrFlushInfo());

    SkBitmap readBack;
    readBack.allocN32Pixels(kFinalWidth, kFinalHeight);

    SkDEBUGCODE(bool result =) rtc->readPixels(readBack.info(), readBack.getPixels(),
                                               readBack.rowBytes(), 0, 0);
    SkASSERT(result);

    context->priv().testingOnly_flushAndRemoveOnFlushCallbackObject(&object);

    object.markAsDone();

    int x = kDrawnTileSize/2;
    test_color(reporter, readBack, x, SK_ColorRED);
    x += kDrawnTileSize;
    test_color(reporter, readBack, x, SK_ColorGREEN);
    x += kDrawnTileSize;
    test_color(reporter, readBack, x, SK_ColorBLUE);
    x += kDrawnTileSize;
    test_color(reporter, readBack, x, SK_ColorCYAN);
    x += kDrawnTileSize;
    test_color(reporter, readBack, x, SK_ColorMAGENTA);
    x += kDrawnTileSize;
    test_color(reporter, readBack, x, SK_ColorYELLOW);
}

redpajama_set_name: RedPajamaGithub