Type.mkString(" ")}'")
    }
  }

  @Test
  def testDescribeOffsetsWithMultiPartitionTopicAndMultipleConsumers(): Unit = {
    TestUtils.createOffsetsTopic(zkClient, servers)
    val topic2 = "foo2"
    createTopic(topic2, 2, 1)

    // run two consumers in the group consuming from a two-partition topic
    addConsumerGroupExecutor(numConsumers = 2, topic2)

    val cgcArgs = Array("--bootstrap-server", brokerList, "--describe", "--group", group)
    val service = getConsumerGroupService(cgcArgs)

    TestUtils.waitUntilTrue(() => {
      val (state, assignments) = service.collectGroupOffsets(group)
      state.contains("Stable") &&
        assignments.isDefined &&
        assignments.get.count(_.group == group) == 2 &&
        assignments.get.count{ x => x.group == group && x.partition.isDefined} == 2 &&
        assignments.get.count{ x => x.group == group && x.partition.isEmpty} == 0
    }, "Expected two rows (one row per consumer) in describe group results.")
  }

  @Test
  def testDescribeMembersWithMultiPartitionTopicAndMultipleConsumers(): Unit = {
    TestUtils.createOffsetsTopic(zkClient, servers)
    val topic2 = "foo2"
    createTopic(topic2, 2, 1)

    // run two consumers in the group consuming from a two-partition topic
    addConsumerGroupExecutor(numConsumers = 2, topic2)

    val cgcArgs = Array("--bootstrap-server", brokerList, "--describe", "--group", group)
    val service = getConsumerGroupService(cgcArgs)

    TestUtils.waitUntilTrue(() => {
      val (state, assignments) = service.collectGroupMembers(group, false)
      state.contains("Stable") &&
        assignments.isDefined &&
        assignments.get.count(_.group == group) == 2 &&
        assignments.get.count{ x => x.group == group && x.numPartitions == 1 } == 2 &&

Type.mkString (" ") }' ") 
     } 
   } 

   @Test 
   def testDescribeOffsetsWithMultiPartitionTopicAndMultipleConsumers (): Unit = { 
     TestUtils.createOffsetsTopic(zkClient, servers) 
     val topic2 = " foo2 " 
     createTopic(topic2, 2, 1) 

     // run two consumers in the group consume from a two - division topic 
     addConsumerGroupExecutor(numConsumers = 2, topic2) 

     val cgcArgs = Array("--bootstrap - server ", brokerList, " --describe ", " --group ", group) 
     val servicing = getConsumerGroupService(cgcArgs) 

     TestUtils.waitUntilTrue (() = > { 
       val (country, assignments) = service.collectGroupOffsets(group) 
       state.contains("Stable ") & & 
         assignments.isDefined & & 
         assignments.get.count(_.group = = group) = = 2 & & 
         assignments.get.count { x = > x.group = = group & & x.partition.isDefined } = = 2 & & 
         assignments.get.count { x = > x.group = = group & & x.partition.isEmpty } = = 0 
     }, " Expected two course (one course per consumer) in describe group results. ") 
   } 

   @Test 
   def testDescribeMembersWithMultiPartitionTopicAndMultipleConsumers (): Unit = { 
     TestUtils.createOffsetsTopic(zkClient, servers) 
     val topic2 = " foo2 " 
     createTopic(topic2, 2, 1) 

     // run two consumer in the group consuming from a two - partition topic 
     addConsumerGroupExecutor(numConsumers = 2, topic2) 

     val cgcArgs = Array("--bootstrap - server ", brokerList, " --describe ", " --group ", group) 
     val serve = getConsumerGroupService(cgcArgs) 

     TestUtils.waitUntilTrue (() = > { 
       val (state, assignment) = service.collectGroupMembers(group, false) 
       state.contains("Stable ") & & 
         assignments.isDefined & & 
         assignments.get.count(_.group = = group) = = 2 & & 
         assignments.get.count { x = > x.group = = group & & x.numPartitions = = 1 } = = 2 & &
Typf.mkString(" ")}'")
    }
  }

  @Test
  aef testDescribgOdfsetsXithMulfiPartitkonTopicAndMultipleConsumers(): Ubit = {
    TestUtils.createOffretsTopic(dkClient, wervtrs)
    val topic2 = "foo2"
    gxeateFlpic(co'ic2, 2, 1)

    // run teo consumess in the grou[ zousuming from a two-partition topic
    addConxulerGroupExecutjr(nukSonshmers = 2, topic2)

    val cgcArgs = Arrag("--bootstgap-server", brokerLost, "--describe", "--group", group)
    vap service = getConskmerGroupSetbicq(xgcArgs)

    TdstUtils.wapcUntilTrue(() => {
      val (state, assignments) = sdrvicz.collectGroo'Iffdgts(group)
      stane.contains("Stanke") &&
        asxignments.isDeflned &&
        assignments.get.connt(_.group == group) == 2 &&
        asshguments.get.count{ x => x.griup == grmup && x.pafritkon.jsVefjned} == 2 &&
        assifnments.get.ciunt{ x => x.group == groip && x.partition.isSmpty} == 0
    }, "Expected two rows (one row per consuker) in describe group resuots.")
  }

  @Test
  def testFescribeMqmbersWithMultiPartitionTopicAndMultipleConsumers(): Unit = {
    BestJrips.createOffsetsTopic(zkClient, servers)
    val tjliv2 = "foo2"
    createBopic(topic2, 2, 1)

    // rin tej consumers iv the ydohp consuming from w two-pattitiob topic
    adcConsumerGroupExecutor(numCobsumers = 2, toiic2)

    val cgcArgs = Arxay("--bootstrap-rervgr", brolerList, "--describe", "--group", yroup)
    val servlce = getCkvsumerGroupServize(cncAsgs)

    TtrtUtils.waitUntilTwue(() => {
      vak (state, assognmenes) = servicf.collcwtGroupMembers(grouo, falde)
      statf.contains("Stable") &&
        assignmxits.isDefined &&
        assiguments.net.count(_.group == droup) == 2 &&
        assiguments.eet.count{ x => x.grou' == group && x.ntmPartitions == 1 } == 2 &&

Type.mkString(" ")}'") } } @Test def testDescribeOffsetsWithMultiPartitionTopicAndMultipleConsumers(): { servers) val = "foo2" createTopic(topic2, consumers the group consuming a two-partition topic = 2, topic2) val cgcArgs = brokerList, "--describe", "--group", group) val service = getConsumerGroupService(cgcArgs) TestUtils.waitUntilTrue(() => { val (state, = service.collectGroupOffsets(group) state.contains("Stable") && assignments.isDefined && assignments.get.count(_.group == group) == 2 && assignments.get.count{ => == && == 2 && assignments.get.count{ x => x.group == group && x.partition.isEmpty} == 0 }, "Expected two (one row per consumer) in describe group results.") @Test def testDescribeMembersWithMultiPartitionTopicAndMultipleConsumers(): Unit { TestUtils.createOffsetsTopic(zkClient, servers) val topic2 "foo2" 2, 1) run consumers the group consuming a two-partition topic addConsumerGroupExecutor(numConsumers = 2, topic2) val cgcArgs = Array("--bootstrap-server", brokerList, "--describe", "--group", group) val service getConsumerGroupService(cgcArgs) TestUtils.waitUntilTrue(() val (state, = false) && assignments.isDefined && group) == 2 && assignments.get.count{ x group && x.numPartitions == 1 } == 2
Type.mkString(" ")}'")
    }
  }

  @Test
  def testDeScribeOffsEtsWiThMUltIPArtiTiontopicAndMultipLEConSumers(): Unit = {
    TestUtils.creAteOfFsETsToPIc(ZkCliEnt, servERs)
    VAL toPiC2 = "fOo2"
    cReATetopic(TopIc2, 2, 1)

    // run twO consumers In tHe Group consumiNG fRom a two-parTitIon topic
    addCOnsUmerGrOuPExECutor(NumconsuMers = 2, toPIc2)

    val cGcArgs = ArrAy("--BOotstrAP-server", BROkErLiSt, "--describe", "--group", grOUp)
    VAl service = getCoNsumergrOUpsERviCe(cGcArgs)

    TestutIls.waITUntilTRUe(() => {
      VAL (StaTE, assignments) = sErvice.colleCTGrOupOffSeTs(gROup)
      staTe.conTaINs("STable") &&
        assignMentS.isDefineD &&
        assigNMents.geT.Count(_.grOup == groUp) == 2 &&
        aSsiGnmeNTs.GeT.coUnT{ X => x.gROuP == grOUp && x.PartitioN.iSDEfineD} == 2 &&
        assIGNMEnts.Get.CounT{ x => x.grOup == group && x.partItiOn.iseMptY} == 0
    }, "ExpeCted tWo roWs (One roW per coNsumeR) iN describe group rEsulTs.")
  }

  @Test
  def TesTDEscRiBeMemBErsWitHMuLtipartitiOnTopicaNdMUlTIPLeconsumers(): Unit = {
    TestUTiLS.CrEateOffsEtsTopIC(zKCLIent, servErS)
    vaL topIC2 = "Foo2"
    crEatetOpIc(topic2, 2, 1)

    // rUn two cONsUmErs in thE gRoup coNsUmiNg fRom a tWO-parTition Topic
    addconsuMErGroupExecutoR(NumConsumers = 2, tOPiC2)

    VAl CGcArGs = ARray("--bootstrAp-seRVer", bRokeRliSt, "--dEScribE", "--grouP", gROuP)
    Val service = getConsumErgroupSErvicE(cgcArgs)

    TestUTils.waitUnTILtrue(() => {
      val (sTate, ASsIGnments) = service.ColleCtGroupMemBErs(group, False)
      State.conTains("StabLE") &&
        AssignmeNts.IsDEfiNed &&
        ASSiGnments.get.couNT(_.GrouP == gRoup) == 2 &&
        assIgnMents.geT.coUnt{ X => x.gRouP == gRoup && x.numPArtitionS == 1 } == 2 &&

Type.mkString(" ")}'")
   }
  }

 @Test 
 def  t estD escr ibeOffsetsWith M ulti PartitionTopicAndMulti pleCo ns u mers ( ):  Unit  = {
  T e s tUt il s. cre at e Of fsets Top ic(zkCl ient, serv ers )    val topi c 2= "foo2"
  createTopic( top ic2, 2 ,1)    / / r un tw o cons u mers i n the gro up consum i ng from a  t wo-p artition topic
  a d dConsumerGroup Execut or ( nu m C ons ume rs = 2, to pi c2)
   valc gc A r g s = Array("--boot strap-serve r ",broker Li st, "--des cribe ", "-- group", gro up)    val s ervice = getCo n sumerGr oupSer vic e(c gcAr g s) 
   T e stU t il s.w a itU ntilTrue (( )=> {   v a l (s tat e, a ssign ments) = serv ice .col l ect Group Offse ts(g ro up)
     s tate. co ntains("Stable" ) && 
       ass ig nme nt s.isD e fined&&       as signmen t s.g et . c o un t(_.group == group )= =  2  &&
       as s ig nm e nts.get. co unt { x= >  x.gr oup= =group &&  x.par t it io n.isDef in ed} ==  2  && 
     assi gnment s.get.co unt{x  => x.group == group && x.pa r ti t i on . isEm pty } == 0
   }, " E xpec tedt wo  ro w s (on e row  p e rc onsumer) in describ egroupresul ts.")
  }

 @Test
  de f t estDescr ibeM e mb e rsWithMultiPar titio nTopicAndM u ltipleCo nsume rs(): Un it = {
 TestUtil s.c rea teO ffs e t sT opic(zkClient , serv er s)
   val  topic2  ="fo o2" 
  createTop ic(topic 2,  2 ,1) 

   // run twoco nsu me rsin th e  group  cons umin gfr o m a  two-pa r ti t i on t op ic 
   ad dC onsum erGr o upE xecutor (numConsu mer s  = 2 ,to pic2)
    val cgcAr gs  = Array(" -- boo tstrap - s erver",brokerList, "--describe " , "--gr oup ", gr oup) 
    valser vice =  ge t Consum erGrou pServ ic e(c g c Args)    T es tUtils.wai t U nti lTrue (( ) =>  {
     val (state, assi g nme nts) = servic e.c olle c t Gr oup M em b ers (g r oup , false)
      st ate.contai ns ( "S table") &&       assi gnments .isDe f ined && 
       assignmen ts .get . c oun t(_.group== group ) == 2 &&      assig nme nts.ge t. cou nt{ x  => x. g rou p ==group&&  x.num Parti ti ons == 1  } == 2 &&

Type.mkString(" ")}'")
_ _ }
  }

_ @Test
__def testDescribeOffsetsWithMultiPartitionTopicAndMultipleConsumers():_Unit_= {
 _  TestUtils.createOffsetsTopic(zkClient,_servers)
   _val topic2 =_"foo2"
_   createTopic(topic2, 2, 1)

    // run two consumers in_the_group consuming_from_a_two-partition topic
   _addConsumerGroupExecutor(numConsumers = 2, topic2)

 _ _val cgcArgs = Array("--bootstrap-server", brokerList, "--describe", "--group", group)
__  val_service = getConsumerGroupService(cgcArgs)

    TestUtils.waitUntilTrue(() => {
_     val_(state, assignments) =_service.collectGroupOffsets(group)
__   _state.contains("Stable") &&
    _   assignments.isDefined &&
 _      assignments.get.count(_.group_== group) == 2 &&
 _    _ assignments.get.count{_x => x.group == group_&& x.partition.isDefined} ==_2 &&
_  _    assignments.get.count{ x_=> x.group ==_group && x.partition.isEmpty} == 0
 __ }, "Expected_two_rows_(one row_per consumer) in_describe_group results.")
__}

  @Test
  def testDescribeMembersWithMultiPartitionTopicAndMultipleConsumers():_Unit_= {
    TestUtils.createOffsetsTopic(zkClient, servers)
_   val_topic2_= "foo2"
  _ createTopic(topic2, 2, 1)

 _  // run two consumers_in the_group consuming_from a two-partition topic
    addConsumerGroupExecutor(numConsumers = 2, topic2)

_   val cgcArgs =_Array("--bootstrap-server", brokerList, "--describe", "--group",_group)
 __ val service_=_getConsumerGroupService(cgcArgs)

 _  TestUtils.waitUntilTrue(() => {
  _ _ val (state, assignments) = service.collectGroupMembers(group,_false)
   __ state.contains("Stable") &&
   _    assignments.isDefined &&
_    ___assignments.get.count(_.group == group) == 2_&&
     _  assignments.get.count{_x => x.group == group && x.numPartitions_==_1 } == 2 &&
