/* Ċ Ġ* ĠĠ ĠCopyright Ġ 2 0 2 0 ĠAmazon . com , ĠInc . Ġor Ġits Ġaffiliates . Ġ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 . Ċ Ġ* ĠĠ ĠA Ġcopy Ġof Ġthe ĠLicense Ġis Ġlocated Ġat Ċ Ġ* Ċ Ġ* ĠĠĠĠĠĠ Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠĠ Ġor Ġin Ġthe Ġ" license " Ġfile Ġaccompanying Ġthis Ġfile . ĠThis Ġfile Ġis Ġdistributed Ċ Ġ* ĠĠ Ġon Ġan Ġ" AS ĠIS " ĠBASIS , ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ċ Ġ* ĠĠ Ġexpress Ġor Ġimplied . ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ċ Ġ* ĠĠ Ġpermissions Ġand Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ġ*/ Ċ Ċ package Ġcom . amazon . elasticsearch . replication . inte g . rest Ċ Ċ import Ġcom . amazon . elasticsearch . replication . Multi Cluster Annotations Ċ import Ġcom . amazon . elasticsearch . replication . Multi Cluster Rest TestCase Ċ import Ġcom . amazon . elasticsearch . replication . Start Replication Request Ċ import Ġcom . amazon . elasticsearch . replication . delete Auto Follow Pattern Ċ import Ġcom . amazon . elasticsearch . replication . start Replication Ċ import Ġcom . amazon . elasticsearch . replication . stop Replication Ċ import Ġcom . amazon . elasticsearch . replication . task . aut of ollow . Auto Follow Executor Ċ import Ġcom . amazon . elasticsearch . replication . task . index . Index Replication Executor Ċ import Ġcom . amazon . elasticsearch . replication . update Auto Follow Pattern Ċ import Ġorg . apache . http . HttpStatus Ċ import Ġorg . apache . http . entity . ContentType Ċ import Ġorg . apache . http . nio . entity . N String Entity Ċ import Ġorg . assertj . core . api . Assertions Ċ import Ġorg . elasticsearch . client . Request Ċ import Ġorg . elasticsearch . client . RequestOptions Ċ import Ġorg . elasticsearch . client . Response Exception Ċ import Ġorg . elasticsearch . client . Rest High Level Client Ċ import Ġorg . elasticsearch . client . indices . Create Index Request Ċ import Ġorg . elasticsearch . client . indices . Get Index Request Ċ import Ġorg . elasticsearch . common . unit . Time Value Ċ import Ġorg . elasticsearch . tasks . Task Info Ċ import Ġjava . util . Locale Ċ Ċ import Ġjava . util . concurrent . TimeUnit ĊĊ Ċ @ Multi Cluster Annotations . Cluster Configurations ( ĊĠĠĠĠĠĠĠ ĠMulti Cluster Annotations . Cluster Configuration ( cluster Name Ġ= ĠLE ADER ), ĊĠĠĠĠĠĠĠ ĠMulti Cluster Annotations . Cluster Configuration ( cluster Name Ġ= ĠFOL LOWER ) Ċ ) Ċ class ĠUpdate Auto Follow Pattern IT : ĠMulti Cluster Rest TestCase () Ġ{ ĊĠĠĠ Ġprivate Ġval Ġindex Prefix Ġ= Ġ" leader _ index _" ĊĠĠĠ Ġprivate Ġval Ġindex Pattern Ġ= Ġ" leader _ index *" ĊĠĠĠ Ġprivate Ġval Ġindex Pattern Name Ġ= Ġ" test _ pattern " ĊĠĠĠ Ġprivate Ġval Ġconnection Alias Ġ= Ġ" test _ conn " ĊĊĠĠĠ Ġfun Ġ` test Ġauto Ġfollow Ġpattern `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġfollow er Client Ġ= ĠgetClient For Cluster ( FOL LOWER ) ĊĠĠĠĠĠĠĠ Ġval Ġleader Client Ġ= ĠgetClient For Cluster ( LE ADER ) ĊĠĠĠĠĠĠĠ Ġval Ġleader Index Name Ġ= Ġcreate Random Index ( leader Client ) ĊĠĠĠĠĠĠĠ Ġcreate Connection Between Clusters ( FOL LOWER , ĠLE ADER , Ġconnection Alias ) ĊĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfollow er Client . update Auto Follow Pattern ( connection Alias , Ġindex Pattern Name , Ġindex Pattern ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠVerify Ġthat Ġexisting Ġindex Ġmatching Ġthe Ġpattern Ġare Ġreplic ated . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert Busy Ġ({ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠAssertions . assertThat ( follow er Client . indices () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. exists ( Get Index Request ( leader Index Name ), ĠRequest Options . DEFAULT )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}, Ġ 3 0 , ĠTimeUnit . SECONDS ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠAssertions . assertThat ( get Auto Follow Tasks ( FOL LOWER ). size ). isEqualTo ( 1 ) ĊĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠVerify Ġthat Ġnewly Ġcreated Ġindex Ġon Ġleader Ġwhich Ġmatch Ġthe Ġpattern Ġare Ġalso Ġreplic ated . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġleader Index Name New Ġ= Ġcreate Random Index ( leader Client ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert Busy Ġ({ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠAssertions . assertThat ( follow er Client . indices () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. exists ( Get Index Request ( leader Index Name New ), ĠRequest Options . DEFAULT )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}, Ġ 3 0 , ĠTimeUnit . SECONDS ) ĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfollow er Client . delete Auto Follow Pattern ( connection Alias , Ġindex Pattern Name ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ` test Ġauto Ġfollow Ġshouldn 't Ġadd Ġalready Ġtriggered Ġindex `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġfollow er Client Ġ= ĠgetClient For Cluster ( FOL LOWER ) ĊĠĠĠĠĠĠĠ Ġval Ġleader Client Ġ= ĠgetClient For Cluster ( LE ADER ) ĊĠĠĠĠĠĠĠ Ġval Ġleader Index Name Ġ= Ġcreate Random Index ( leader Client ) ĊĠĠĠĠĠĠĠ Ġcreate Connection Between Clusters ( FOL LOWER , ĠLE ADER , Ġconnection Alias ) ĊĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfollow er Client . start Replication ( Start Replication Request ( connection Alias , Ġleader Index Name , Ġleader Index Name ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTime Value . time Value Seconds ( 1 0 ), true ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert Busy ({ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠAssertions . assertThat ( follow er Client . indices () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. exists ( Get Index Request ( leader Index Name ), ĠRequest Options . DEFAULT )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}, Ġ 3 0 , ĠTimeUnit . SECONDS ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠAssert Ġthat Ġthere Ġis Ġno Ġauto Ġfollow Ġtask Ġ& Ġone Ġindex Ġreplication Ġtask ĊĠĠĠĠĠĠĠĠĠĠĠ ĠAssertions . assertThat ( get Auto Follow Tasks ( FOL LOWER ). size ). isEqualTo ( 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠAssertions . assertThat ( getIndex Replication Tasks ( FOL LOWER ). size ). isEqualTo ( 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfollow er Client . update Auto Follow Pattern ( connection Alias , Ġindex Pattern Name , Ġindex Pattern ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠAssert Ġthat Ġthere Ġis Ġstill Ġonly Ġone Ġindex Ġreplication Ġtask ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠAssertions . assertThat ( get Auto Follow Tasks ( FOL LOWER ). size ). isEqualTo ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠAssertions . assertThat ( getIndex Replication Tasks ( FOL LOWER ). size ). isEqualTo ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfollow er Client . delete Auto Follow Pattern ( connection Alias , Ġindex Pattern Name ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfollow er Client . stop Replication ( leader Index Name ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ` test Ġauto Ġfollow Ġshould Ġfail Ġif Ġremote Ġconnection Ġdoesn 't Ġexist `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġfollow er Client Ġ= ĠgetClient For Cluster ( FOL LOWER ) ĊĠĠĠĠĠĠĠ Ġ// ĠCall Ġautof ollow Ġpattern Ġwithout Ġsetting Ġup Ġremote Ġconnection . ĊĠĠĠĠĠĠĠ ĠAssertions . assertThat Thrown By Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfollow er Client . update Auto Follow Pattern ( connection Alias , Ġindex Pattern Name , Ġindex Pattern ) ĊĠĠĠĠĠĠĠ Ġ}. isInstanceOf ( Response Exception :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. has Message Containing (" no Ġsuch Ġremote Ġcluster ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġ` test Ġremoving Ġautof ollow Ġpattern Ġstop Ġautof ollow Ġtask `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġfollow er Client Ġ= ĠgetClient For Cluster ( FOL LOWER ) ĊĠĠĠĠĠĠĠ Ġval Ġleader Client Ġ= ĠgetClient For Cluster ( LE ADER ) ĊĠĠĠĠĠĠĠ Ġcreate Connection Between Clusters ( FOL LOWER , ĠLE ADER , Ġconnection Alias ) ĊĊĠĠĠĠĠĠĠ Ġval Ġleader Index Name Ġ= Ġcreate Random Index ( leader Client ) ĊĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfollow er Client . update Auto Follow Pattern ( connection Alias , Ġindex Pattern Name , Ġindex Pattern ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠVerify Ġthat Ġexisting Ġindex Ġmatching Ġthe Ġpattern Ġare Ġreplic ated . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert Busy Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠAssertions . assertThat ( follow er Client . indices () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. exists ( Get Index Request ( leader Index Name ), ĠRequest Options . DEFAULT )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠAssertions . assertThat ( get Auto Follow Tasks ( FOL LOWER ). size ). isEqualTo ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠAssertions . assertThat ( getIndex Replication Tasks ( FOL LOWER ). size ). isEqualTo ( 1 ) ĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfollow er Client . delete Auto Follow Pattern ( connection Alias , Ġindex Pattern Name ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠVerify Ġthat Ġauto Ġfollow Ġtasks Ġis Ġstopped Ġbut Ġthe Ġshard Ġreplication Ġtask Ġremains . ĊĠĠĠĠĠĠĠ Ġassert Busy Ġ({ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠAssertions . assertThat ( get Auto Follow Tasks ( FOL LOWER ). size ). isEqualTo ( 0 ) ĊĠĠĠĠĠĠĠ Ġ}, Ġ 3 0 , ĠTimeUnit . SECONDS ) ĊĊĠĠĠĠĠĠĠ ĠAssertions . assertThat ( getIndex Replication Tasks ( FOL LOWER ). size ). isEqualTo ( 1 ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġcreate Random Index ( client : ĠRest High Level Client ): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġindex Name Ġ= Ġindex Prefix Ġ+ Ġrandom Alpha Of Length ( 6 ). toLowerCase ( Locale . ROOT ) ĊĠĠĠĠĠĠĠ Ġval Ġcreate Index Response Ġ= Ġclient . indices (). create ( Create Index Request ( index Name ), ĠRequest Options . DEFAULT ) ĊĠĠĠĠĠĠĠ ĠAssertions . assertThat ( create Index Response . is Ack nowled ged ). isTrue () ĊĠĠĠĠĠĠĠ Ġassert Busy Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠAssertions . assertThat ( client . indices () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. exists ( Get Index Request ( index Name ), ĠRequest Options . DEFAULT )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. isEqualTo ( true ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġindex Name ĊĠĠĠ Ġ} ĊĠĠĠ Ġfun Ġget Auto Follow Tasks ( cluster Name : ĠString ): ĠList < Task Info > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġget Replication TaskList ( cluster Name , ĠAuto Follow Executor . TASK _ NAME Ġ+ Ġ"* ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġget Index Replication Tasks ( cluster Name : ĠString ): ĠList < Task Info > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġget Replication TaskList ( cluster Name , ĠIndex Replication Executor . TASK _ NAME Ġ+ Ġ"* ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġcreate Dummy Connection ( from Cluster Name : ĠString , Ġconnection Name : ĠString =" source ") Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġfrom Cluster Ġ= Ġget Named Cluster ( from Cluster Name ) ĊĠĠĠĠĠĠĠ Ġval Ġpersistent Connection Request Ġ= ĠRequest (" PUT ", Ġ"_ cluster / settings ") ĊĠĠĠĠĠĠĠ Ġval Ġto Cluster Dummy Host Seed Ġ= Ġ" localhost : 6 5 5 3 6 " ĊĠĠĠĠĠĠĠ Ġval Ġentity AsString Ġ= Ġ""" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" persistent ": Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" cluster ": Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" remote ": Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"$ connection Name ": Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" seeds ": Ġ[ Ġ"$ to Cluster Dummy Host Seed " Ġ] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} """. trim Margin () ĊĊĠĠĠĠĠĠĠ Ġpersistent Connection Request . entity Ġ= ĠN String Entity ( entity AsString , ĠContentType . APPLICATION _ JSON ) ĊĠĠĠĠĠĠĠ Ġval Ġpersistent Connection Response Ġ= Ġfrom Cluster . low Level Client . perform Request ( persistent Connection Request ) ĊĠĠĠĠĠĠĠ ĠassertEquals ( HttpStatus . SC _ OK . to Long (), Ġpersistent Connection Response . status Line . statusCode . to Long ()) ĊĠĠĠ Ġ} Ċ Ċ }