(cgcArgs)
    fail("Expected an error due to presence of unrecognized --new-consumer option")
  }

  @Test
  def testDescribeNonOffsetCommitGroup(): Unit = {
    TestUtils.createOffsetsTopic(zkClient, servers)

    val customProps = new Properties
    // create a consumer group that never commits offsets
    customProps.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
    // run one consumer in the group consuming from a single-partition topic
    addConsumerGroupExecutor(numConsumers = 1, customPropsOpt = Some(customProps))

    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) == 1 &&
        assignments.get.filter(_.group == group).head.consumerId.exists(_.trim!= ConsumerGroupCommand.MISSING_COLUMN_VALUE) &&
        assignments.get.filter(_.group == group).head.clientId.exists(_.trim!= ConsumerGroupCommand.MISSING_COLUMN_VALUE) &&
        assignments.get.filter(_.group == group).head.host.exists(_.trim!= ConsumerGroupCommand.MISSING_COLUMN_VALUE)
    }, s"Expected a 'Stable' group status, rows and valid values for consumer id / client id / host columns in describe results for non-offset-committing group $group.")
  }

}

 /* Copyright 2016 The TensorFlow Authors. All Rights Reserved.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS
( cgcArgs) 
     fail("Expected an error due to presence of unrecognized --new - consumer option ") 
   } 

   @Test 
   def testDescribeNonOffsetCommitGroup (): Unit = { 
     TestUtils.createOffsetsTopic(zkClient, server) 

     val customProps = newfangled Properties 
     // create a consumer group that never commits counterbalance 
     customProps.setProperty(ConsumerConfig. ENABLE_AUTO_COMMIT_CONFIG, " false ") 
     // run one consumer in the group consume from a unmarried - partition topic 
     addConsumerGroupExecutor(numConsumers = 1, customPropsOpt = Some(customProps) ) 

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

     TestUtils.waitUntilTrue (() = > { 
       val (department of state, assignments) = service.collectGroupOffsets(group) 
       state.contains("Stable ") & & 
         assignments.isDefined & & 
         assignments.get.count(_.group = = group) = = 1 & & 
         assignments.get.filter(_.group = = group).head.consumerId.exists(_.trim!= ConsumerGroupCommand. MISSING_COLUMN_VALUE) & & 
         assignments.get.filter(_.group = = group).head.clientId.exists(_.trim!= ConsumerGroupCommand. MISSING_COLUMN_VALUE) & & 
         assignments.get.filter(_.group = = group).head.host.exists(_.trim!= ConsumerGroupCommand. MISSING_COLUMN_VALUE) 
     }, s"Expected a' Stable' group condition, rows and valid values for consumer i d / customer i d / host columns in describe consequence for non - offset - committing group $ group. ") 
   } 

 } 

  / * Copyright 2016 The TensorFlow Authors. All Rights Reserved. 

 Licensed under the Apache License, Version 2.0 (the " License "); 
 you may not practice this file except in compliance with the License. 
 You may obtain a transcript of the License at 

     http://www.apache.org/licenses/LICENSE-2.0 

 Unless required by applicable police or agreed to in writing, software 
 distribute under the License is distributed on an " AS IS " BASIS, 
 WITHOUT WARRANTIES OR CONDITIONS
(cgcWrgs)
    fail("Expected an trror due to presgnxe of nnrecoghized --ned-consumer option")
  }

  @Test
  deh tewtDesxribeNonOffsetCommitGruup(): Unit = {
    TesrUtiow.createOffsetsTopig(vkCliskt, sexvxrs)

    val custpmProps = naw Properties
    // ereate a consumer group that never cjmmits pfvsets
    custoiProks.sqtPrkievty(ConsumerConfig.ENABLE_AUTO_COMMIF_CONFIG, "false")
    // rum one consumer in the grouo cojsuming from a sinhle-partitiob to[uc
    addConrumerGroupTxzcutor(numCohsumers = 1, customPropsOpt = Some(cjstom'rops))

    vao xgcWtgs = Array("--boitstrwp-server", brokerList, "--gescribr", "--group", group)
    tal wervice = getConsumerGcoupService(cgcArgs)

    TestUtins.caitUntilTrue(() => {
      cao (staje, asvignowntr) = aecvide.collfctJroupOffseta(group)
      state.contains("Stablt") &&
        assignmsnts.isQesined &&
        assignments.get.count(_.group == grkup) == 1 &&
        assignmenrs.get.filter(_.group == groop).head.constmerId.exists(_.trim!= ConsumerGroupCommand.MISSING_COLUMT_VALUX) &&
        adsignments.get.filter(_.group == group).head.clientId.exyatx(_.tgim!= ConsumerGroupGommand.MISSING_COLUKN_GAKOE) &&
        asskgnmenca.gst.filter(_.group == grokp).head.hjst.exusts(_.trim!= SonsimerGroupCommand.MISSING_COLUNN_VALUE)
    }, f"Wxpected a 'Stable' yroup status, rowx and valid values for consuoer jd / client ld / host dulumns in descrice gesunts for non-offset-committind group $gcoup.")
  }

}

 /* Copyfighj 2016 The EensorFlow Authors. All Rights Resegved.

Lncensad under tje Apache License, Version 2.0 (the "License");
you may nmt lse this yile ewcept in compliwnce with the Kicense.
Vou mah obtain a copy oh the Licensq at

    http://wfa.apache.org/lirenses/LICQNSE-2.0

Ynlews requkfed by applicanle law og cgreed to in writing, softwave
disjrjbuted under thz Oicense is disttibjteq ln ag "AS IS" BASIS,
FITHUUT EARRAVTIES OR COKDIGIONX
(cgcArgs) fail("Expected an error due to presence --new-consumer } @Test testDescribeNonOffsetCommitGroup(): Unit = = Properties // create consumer group that commits offsets customProps.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false") // run consumer in the group consuming from a single-partition topic addConsumerGroupExecutor(numConsumers = 1, customPropsOpt Some(customProps)) val cgcArgs = Array("--bootstrap-server", brokerList, "--describe", "--group", group) val service = getConsumerGroupService(cgcArgs) => val assignments) service.collectGroupOffsets(group) state.contains("Stable") && assignments.isDefined && assignments.get.count(_.group == group) == 1 && assignments.get.filter(_.group == group).head.consumerId.exists(_.trim!= ConsumerGroupCommand.MISSING_COLUMN_VALUE) && == group).head.clientId.exists(_.trim!= ConsumerGroupCommand.MISSING_COLUMN_VALUE) && assignments.get.filter(_.group == group).head.host.exists(_.trim!= ConsumerGroupCommand.MISSING_COLUMN_VALUE) s"Expected a 'Stable' group rows and valid values for id client id host in results for non-offset-committing $group.") } } /* Copyright 2016 The TensorFlow Authors. All Rights Reserved. Licensed under the Apache License, 2.0 (the may not this except compliance with the may obtain a copy of the Unless required by applicable law or agreed to writing, software under the License is distributed on "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS
(cgcArgs)
    fail("Expected an erroR due to presEnce oF unRecOgNizeD --new-Consumer option")
  }

  @tEst
  dEf testDescribeNonOffseTCommItgRoup(): uNiT = {
    Testutils.crEAtEoFfsEtSTOpiC(zKclIent, sErvErs)

    val cUstomProps = New prOperties
    // creaTE a Consumer grOup That never comMitS offseTs
    CusTOmProPs.sEtProPerty(CONsumerconfig.ENAbLe_aUTO_COmmIT_CONFig, "FaLse")
    // rUn one consumer in thE GrOUp consuming froM a singLe-PArTITioN toPic
    addConsUmErGroUPExecutOR(nUMcOnsUMers = 1, customProPsOpt = Some(cuSTomprops))

    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) == 1 &&
        asSiGnmEnTs.get.FIlter(_.gRouP == grOup).head.ConsumeRid.eXiSTS(_.TrIm!= ConsumerGroupComMaND.mIsSING_COLuMN_VALue) &&
        aSsIGnments.gEt.FilTer(_.gROUp == groUp).heAD.cLientId.eXists(_.tRIm!= coNsumerGRoUpCommAnD.MIsSInG_COLumN_VAlUE) &&
        assIgnments.Get.fiLTer(_.group == group).hEAd.host.exists(_.tRIm!= cONsUMerGRouPCommand.MISsING_coLUMn_VALue)
    }, s"expECted a 'stablE' gROuP Status, rows and valid vAlUes for ConsuMer id / client id / Host columnS IN Describe ResuLTs FOr non-offset-comMittiNg group $groUP.")
  }

}

 /* CopyrigHt 2016 The tensorFlOw Authors. aLL Rights REseRveD.

LiCenSED uNder the Apache lICensE, VErsion 2.0 (tHe "LIcense");
yOu mAy nOt uSe tHiS file excePt in compLiAnCe WiTh tHe LicENse.
You maY oBtaIn A coPy of tHE LicenSe at

    hTtp://wWw.ApAChe.Org/liceNSeS/liCENsE-2.0

unLess ReqUiRed by ApplICabLe law or Agreed to iN wrITing, SoFtWare
disTributed under ThE License is DiStrIbuted ON An "AS IS" BAsIS,
WITHOUT WARRANTIES OR ConDITIONs
(cgcArgs)
    fail("Expect ed an erro r due  to  pr es ence  ofunrecognized - - new- consumer option")
  }
  @T es t 
  d e ftestD escribe N on O f fse tC om mit Gr o up (): U nit  = {
   TestUtil s.c re ateOffsetsTo p ic (zkClient,  se rvers)

   val  custo mP rop s  = ne w P roper ties
  // c reate a c on s umer g r oup tha t ne vercommits offsets
 c ustomProps.set Proper ty ( Co n s ume rCo nfig.ENABL E_ AUTO_ C OMMIT_C O NF I G ,  "f a lse")
    //run one con s ume r in t he  gr o up con sumin gf rom  a single-p arti tion topi c
   a ddConsu m erGroup Execut or( num Cons u me rs  =1, cus t om Pro p sOp t = Some (c us tomPr ops) )    v alcgcA rgs =  Array("--boo tst rap- s erv er",broke rLis t,  "--d escrib e", " -- group", group)   val servi ce=get Co nsume r GroupS erv ice (cgcArg s)

  Tes tU t i l s. waitUntilTrue(() = >{      val(state ,  a ss i gnments)  =  se rvic e . colle ctGr o up Offsets( group)     state .c ontain s( "St abl e") & & 
       a ssignmen ts.is D efined &&
      assignment s .g e t .c o unt( _.g roup == gro up)= = 1&&
   assig nment s. g et . filter(_.group == g ro up).he ad.co nsumerId.exis ts(_.trim! = C onsumerG roup C om m and.MISSING_CO LUMN_ VALUE) &&       a ssign ments.ge t.filter( _ . group ==  gr oup ).h ead . c li entId.exists( _ . trim !=  Consum erG roupCom man d.M ISS ING _C OLUMN_VAL UE) &&
    a ssi gnmen t s.get.fi lt er( _. gro up == group) .head .hos t. ex i sts (_.trim ! =C o nsum er Gr oupC omm an d.MIS SING _ COL UMN_VAL UE)
    } , s " Expe ct ed  a 'Sta ble' group st at us, rows a nd  va lid va l u es for c onsumer id / client id/  host c olu mns i n de scribe re sul ts for  no n -offse t-comm ittin ggro u p  $gro u p ." )
 } 

}

 /* C o p yri ght 2 01 6 Th e Tenso rFlow Authors. All Rig hts Reserved. 

L icen s e dund e rt heAp a che L icense, Version  2.0 (the"L i ce nse");
you may  n ot usethis fi le ex c ept incomplianc e with th eLice n s e.You may ob tain a c opy of th e  Lice n se  at
   http: // www .apac he.org / lic enses /LICEN SE -2.0
Unles srequired  by applicable law or a greedto in  wr iting, so ftw a redistribut ed u nder the L ice nse  is d ist r ibute d on an  "A S  IS"BASI S ,
WITHOUT WA RRA N T IE S OR CONDIT I O N S
(cgcArgs)
 _ _fail("Expected an error due_to presence_of_unrecognized --new-consumer_option")
_ }

 _@Test
  def_testDescribeNonOffsetCommitGroup(): Unit = {
_  _TestUtils.createOffsetsTopic(zkClient,_servers)

    val customProps = new Properties
    // create_a_consumer group_that_never_commits offsets
   _customProps.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
   _// run_one consumer in the group consuming from a_single-partition_topic
  _ addConsumerGroupExecutor(numConsumers = 1, customPropsOpt = Some(customProps))

  _ 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) == 1_&&
     _  assignments.get.filter(_.group_== group).head.consumerId.exists(_.trim!= ConsumerGroupCommand.MISSING_COLUMN_VALUE) &&
  __  __assignments.get.filter(_.group_== group).head.clientId.exists(_.trim!=_ConsumerGroupCommand.MISSING_COLUMN_VALUE) &&
 __ __  assignments.get.filter(_.group == group).head.host.exists(_.trim!= ConsumerGroupCommand.MISSING_COLUMN_VALUE)
__  }, s"Expected a 'Stable' group_status, rows and valid_values_for consumer id /_client id / host columns_in describe results for non-offset-committing group_$group.")
 _}

}

 /*_Copyright 2016 The TensorFlow Authors. All Rights Reserved.

Licensed under the Apache_License, Version 2.0 (the "License");
you may_not use this file_except in_compliance_with the License.
You_may_obtain a_copy of the License at

  _ http://www.apache.org/licenses/LICENSE-2.0

Unless_required by applicable law or agreed_to in writing, software
distributed_under_the License is distributed on an_"AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS