5
5
*/
6
6
package org .elasticsearch .xpack .ccr .action ;
7
7
8
+ import com .carrotsearch .hppc .cursors .ObjectObjectCursor ;
8
9
import org .elasticsearch .Version ;
9
10
import org .elasticsearch .action .admin .cluster .state .ClusterStateResponse ;
11
+ import org .elasticsearch .action .support .replication .ClusterStateCreationUtils ;
10
12
import org .elasticsearch .client .Client ;
11
13
import org .elasticsearch .cluster .ClusterName ;
12
14
import org .elasticsearch .cluster .ClusterState ;
18
20
import org .elasticsearch .cluster .routing .ShardRoutingState ;
19
21
import org .elasticsearch .cluster .routing .TestShardRouting ;
20
22
import org .elasticsearch .cluster .service .ClusterService ;
23
+ import org .elasticsearch .common .UUIDs ;
21
24
import org .elasticsearch .common .collect .Tuple ;
22
25
import org .elasticsearch .common .settings .ClusterSettings ;
23
26
import org .elasticsearch .common .settings .Settings ;
24
27
import org .elasticsearch .common .unit .ByteSizeValue ;
25
28
import org .elasticsearch .common .unit .TimeValue ;
29
+ import org .elasticsearch .common .util .concurrent .ConcurrentCollections ;
26
30
import org .elasticsearch .common .util .concurrent .EsRejectedExecutionException ;
27
31
import org .elasticsearch .index .Index ;
28
32
import org .elasticsearch .index .IndexSettings ;
44
48
import java .util .LinkedList ;
45
49
import java .util .List ;
46
50
import java .util .Map ;
51
+ import java .util .Set ;
47
52
import java .util .concurrent .CountDownLatch ;
48
53
import java .util .concurrent .ExecutorService ;
49
54
import java .util .concurrent .Executors ;
50
55
import java .util .concurrent .atomic .AtomicInteger ;
56
+ import java .util .concurrent .atomic .AtomicReference ;
51
57
import java .util .function .BiConsumer ;
52
58
import java .util .function .Consumer ;
53
59
import java .util .function .Function ;
57
63
import static org .elasticsearch .xpack .ccr .action .AutoFollowCoordinator .AutoFollower .recordLeaderIndexAsFollowFunction ;
58
64
import static org .hamcrest .Matchers .equalTo ;
59
65
import static org .hamcrest .Matchers .greaterThan ;
66
+ import static org .hamcrest .Matchers .hasItem ;
60
67
import static org .hamcrest .Matchers .is ;
61
68
import static org .hamcrest .Matchers .notNullValue ;
62
69
import static org .hamcrest .Matchers .nullValue ;
@@ -80,7 +87,7 @@ public void testAutoFollower() {
80
87
Map <String , List <String >> followedLeaderIndexUUIDS = new HashMap <>();
81
88
followedLeaderIndexUUIDS .put ("remote" , new ArrayList <>());
82
89
Map <String , Map <String , String >> autoFollowHeaders = new HashMap <>();
83
- autoFollowHeaders .put ("remote" , Collections . singletonMap ("key" , "val" ));
90
+ autoFollowHeaders .put ("remote" , Map . of ("key" , "val" ));
84
91
AutoFollowMetadata autoFollowMetadata = new AutoFollowMetadata (patterns , followedLeaderIndexUUIDS , autoFollowHeaders );
85
92
86
93
ClusterState currentState = ClusterState .builder (new ClusterName ("name" ))
@@ -315,7 +322,7 @@ public void testGetLeaderIndicesToFollow() {
315
322
Map <String , Map <String , String >> headers = new HashMap <>();
316
323
ClusterState clusterState = ClusterState .builder (new ClusterName ("remote" ))
317
324
.metaData (MetaData .builder ().putCustom (AutoFollowMetadata .TYPE ,
318
- new AutoFollowMetadata (Collections . singletonMap ("remote" , autoFollowPattern ), Collections .emptyMap (), headers )))
325
+ new AutoFollowMetadata (Map . of ("remote" , autoFollowPattern ), Collections .emptyMap (), headers )))
319
326
.build ();
320
327
321
328
RoutingTable .Builder routingTableBuilder = RoutingTable .builder ();
@@ -377,7 +384,7 @@ public void testGetLeaderIndicesToFollow_shardsNotStarted() {
377
384
Map <String , Map <String , String >> headers = new HashMap <>();
378
385
ClusterState clusterState = ClusterState .builder (new ClusterName ("remote" ))
379
386
.metaData (MetaData .builder ().putCustom (AutoFollowMetadata .TYPE ,
380
- new AutoFollowMetadata (Collections . singletonMap ("remote" , autoFollowPattern ), Collections .emptyMap (), headers )))
387
+ new AutoFollowMetadata (Map . of ("remote" , autoFollowPattern ), Collections .emptyMap (), headers )))
381
388
.build ();
382
389
383
390
// 1 shard started and another not started:
@@ -416,9 +423,29 @@ public void testGetLeaderIndicesToFollow_shardsNotStarted() {
416
423
assertThat (result .get (1 ).getName (), equalTo ("index2" ));
417
424
}
418
425
426
+ public void testGetLeaderIndicesToFollowWithClosedIndices () {
427
+ final AutoFollowPattern autoFollowPattern = new AutoFollowPattern ("remote" , Collections .singletonList ("*" ),
428
+ null , null , null , null , null , null , null , null , null , null , null );
429
+
430
+ // index is opened
431
+ ClusterState remoteState = ClusterStateCreationUtils .stateWithActivePrimary ("test-index" , true , randomIntBetween (1 , 3 ), 0 );
432
+ List <Index > result = AutoFollower .getLeaderIndicesToFollow (autoFollowPattern , remoteState , Collections .emptyList ());
433
+ assertThat (result .size (), equalTo (1 ));
434
+ assertThat (result , hasItem (remoteState .metaData ().index ("test-index" ).getIndex ()));
435
+
436
+ // index is closed
437
+ remoteState = ClusterState .builder (remoteState )
438
+ .metaData (MetaData .builder (remoteState .metaData ())
439
+ .put (IndexMetaData .builder (remoteState .metaData ().index ("test-index" )).state (IndexMetaData .State .CLOSE ).build (), true )
440
+ .build ())
441
+ .build ();
442
+ result = AutoFollower .getLeaderIndicesToFollow (autoFollowPattern , remoteState , Collections .emptyList ());
443
+ assertThat (result .size (), equalTo (0 ));
444
+ }
445
+
419
446
public void testRecordLeaderIndexAsFollowFunction () {
420
447
AutoFollowMetadata autoFollowMetadata = new AutoFollowMetadata (Collections .emptyMap (),
421
- Collections . singletonMap ("pattern1" , Collections .emptyList ()), Collections .emptyMap ());
448
+ Map . of ("pattern1" , Collections .emptyList ()), Collections .emptyMap ());
422
449
ClusterState clusterState = new ClusterState .Builder (new ClusterName ("name" ))
423
450
.metaData (new MetaData .Builder ().putCustom (AutoFollowMetadata .TYPE , autoFollowMetadata ))
424
451
.build ();
@@ -445,7 +472,7 @@ public void testRecordLeaderIndexAsFollowFunctionNoEntry() {
445
472
446
473
public void testCleanFollowedLeaderIndices () {
447
474
AutoFollowMetadata autoFollowMetadata = new AutoFollowMetadata (Collections .emptyMap (),
448
- Collections . singletonMap ("pattern1" , Arrays .asList ("index1" , "index2" , "index3" )), Collections .emptyMap ());
475
+ Map . of ("pattern1" , Arrays .asList ("index1" , "index2" , "index3" )), Collections .emptyMap ());
449
476
ClusterState clusterState = new ClusterState .Builder (new ClusterName ("name" ))
450
477
.metaData (new MetaData .Builder ().putCustom (AutoFollowMetadata .TYPE , autoFollowMetadata ))
451
478
.build ();
@@ -474,7 +501,7 @@ public void testCleanFollowedLeaderIndices() {
474
501
475
502
public void testCleanFollowedLeaderIndicesNoChanges () {
476
503
AutoFollowMetadata autoFollowMetadata = new AutoFollowMetadata (Collections .emptyMap (),
477
- Collections . singletonMap ("pattern1" , Arrays .asList ("index1" , "index2" , "index3" )), Collections .emptyMap ());
504
+ Map . of ("pattern1" , Arrays .asList ("index1" , "index2" , "index3" )), Collections .emptyMap ());
478
505
ClusterState clusterState = new ClusterState .Builder (new ClusterName ("name" ))
479
506
.metaData (new MetaData .Builder ().putCustom (AutoFollowMetadata .TYPE , autoFollowMetadata ))
480
507
.build ();
@@ -507,7 +534,7 @@ public void testCleanFollowedLeaderIndicesNoChanges() {
507
534
508
535
public void testCleanFollowedLeaderIndicesNoEntry () {
509
536
AutoFollowMetadata autoFollowMetadata = new AutoFollowMetadata (Collections .emptyMap (),
510
- Collections . singletonMap ("pattern2" , Arrays .asList ("index1" , "index2" , "index3" )), Collections .emptyMap ());
537
+ Map . of ("pattern2" , Arrays .asList ("index1" , "index2" , "index3" )), Collections .emptyMap ());
511
538
ClusterState clusterState = new ClusterState .Builder (new ClusterName ("name" ))
512
539
.metaData (new MetaData .Builder ().putCustom (AutoFollowMetadata .TYPE , autoFollowMetadata ))
513
540
.build ();
@@ -717,7 +744,7 @@ public void testWaitForMetadataVersion() {
717
744
Map <String , List <String >> followedLeaderIndexUUIDS = new HashMap <>();
718
745
followedLeaderIndexUUIDS .put ("remote" , new ArrayList <>());
719
746
Map <String , Map <String , String >> autoFollowHeaders = new HashMap <>();
720
- autoFollowHeaders .put ("remote" , Collections . singletonMap ("key" , "val" ));
747
+ autoFollowHeaders .put ("remote" , Map . of ("key" , "val" ));
721
748
AutoFollowMetadata autoFollowMetadata = new AutoFollowMetadata (patterns , followedLeaderIndexUUIDS , autoFollowHeaders );
722
749
723
750
final LinkedList <ClusterState > leaderStates = new LinkedList <>();
@@ -763,7 +790,9 @@ void updateAutoFollowMetadata(Function<ClusterState, ClusterState> updateFunctio
763
790
autoFollower .start ();
764
791
assertThat (allResults .size (), equalTo (states .length ));
765
792
for (int i = 0 ; i < states .length ; i ++) {
766
- assertThat (allResults .get (i ).autoFollowExecutionResults .containsKey (new Index ("logs-" + i , "_na_" )), is (true ));
793
+ final String indexName = "logs-" + i ;
794
+ assertThat (allResults .get (i ).autoFollowExecutionResults .keySet ().stream ()
795
+ .anyMatch (index -> index .getName ().equals (indexName )), is (true ));
767
796
}
768
797
}
769
798
@@ -778,7 +807,7 @@ public void testWaitForTimeOut() {
778
807
Map <String , List <String >> followedLeaderIndexUUIDS = new HashMap <>();
779
808
followedLeaderIndexUUIDS .put ("remote" , new ArrayList <>());
780
809
Map <String , Map <String , String >> autoFollowHeaders = new HashMap <>();
781
- autoFollowHeaders .put ("remote" , Collections . singletonMap ("key" , "val" ));
810
+ autoFollowHeaders .put ("remote" , Map . of ("key" , "val" ));
782
811
AutoFollowMetadata autoFollowMetadata = new AutoFollowMetadata (patterns , followedLeaderIndexUUIDS , autoFollowHeaders );
783
812
784
813
ClusterState [] states = new ClusterState [16 ];
@@ -836,7 +865,7 @@ public void testAutoFollowerSoftDeletesDisabled() {
836
865
Map <String , List <String >> followedLeaderIndexUUIDS = new HashMap <>();
837
866
followedLeaderIndexUUIDS .put ("remote" , new ArrayList <>());
838
867
Map <String , Map <String , String >> autoFollowHeaders = new HashMap <>();
839
- autoFollowHeaders .put ("remote" , Collections . singletonMap ("key" , "val" ));
868
+ autoFollowHeaders .put ("remote" , Map . of ("key" , "val" ));
840
869
AutoFollowMetadata autoFollowMetadata = new AutoFollowMetadata (patterns , followedLeaderIndexUUIDS , autoFollowHeaders );
841
870
842
871
ClusterState currentState = ClusterState .builder (new ClusterName ("name" ))
@@ -902,14 +931,14 @@ public void testAutoFollowerFollowerIndexAlreadyExists() {
902
931
Map <String , List <String >> followedLeaderIndexUUIDS = new HashMap <>();
903
932
followedLeaderIndexUUIDS .put ("remote" , new ArrayList <>());
904
933
Map <String , Map <String , String >> autoFollowHeaders = new HashMap <>();
905
- autoFollowHeaders .put ("remote" , Collections . singletonMap ("key" , "val" ));
934
+ autoFollowHeaders .put ("remote" , Map . of ("key" , "val" ));
906
935
AutoFollowMetadata autoFollowMetadata = new AutoFollowMetadata (patterns , followedLeaderIndexUUIDS , autoFollowHeaders );
907
936
908
937
ClusterState currentState = ClusterState .builder (new ClusterName ("name" ))
909
938
.metaData (MetaData .builder ()
910
939
.put (IndexMetaData .builder ("logs-20190101" )
911
940
.settings (settings (Version .CURRENT ).put (IndexSettings .INDEX_SOFT_DELETES_SETTING .getKey (), true ))
912
- .putCustom (Ccr .CCR_CUSTOM_METADATA_KEY , Collections . singletonMap (Ccr .CCR_CUSTOM_METADATA_LEADER_INDEX_UUID_KEY ,
941
+ .putCustom (Ccr .CCR_CUSTOM_METADATA_KEY , Map . of (Ccr .CCR_CUSTOM_METADATA_LEADER_INDEX_UUID_KEY ,
913
942
remoteState .metaData ().index ("logs-20190101" ).getIndexUUID ()))
914
943
.numberOfShards (1 )
915
944
.numberOfReplicas (0 ))
@@ -1045,6 +1074,85 @@ void updateAutoFollowMetadata(
1045
1074
}
1046
1075
}
1047
1076
1077
+ public void testClosedIndicesAreNotAutoFollowed () {
1078
+ final Client client = mock (Client .class );
1079
+ when (client .getRemoteClusterClient (anyString ())).thenReturn (client );
1080
+
1081
+ final String pattern = "pattern1" ;
1082
+ final ClusterState localState = ClusterState .builder (new ClusterName ("local" ))
1083
+ .metaData (MetaData .builder ()
1084
+ .putCustom (AutoFollowMetadata .TYPE ,
1085
+ new AutoFollowMetadata (Map .of (pattern , new AutoFollowPattern ("remote" , List .of ("docs-*" ), null ,
1086
+ null , null , null , null , null , null , null , null , null , null )),
1087
+ Map .of (pattern , List .of ()), Map .of (pattern , Map .of ()))))
1088
+ .build ();
1089
+
1090
+ ClusterState remoteState = null ;
1091
+ final int nbLeaderIndices = randomInt (15 );
1092
+ for (int i = 0 ; i < nbLeaderIndices ; i ++) {
1093
+ String indexName = "docs-" + i ;
1094
+ if (remoteState == null ) {
1095
+ remoteState = createRemoteClusterState (indexName , true );
1096
+ } else {
1097
+ remoteState = createRemoteClusterState (remoteState , indexName );
1098
+ }
1099
+ if (randomBoolean ()) {
1100
+ // randomly close the index
1101
+ remoteState = ClusterState .builder (remoteState .getClusterName ())
1102
+ .routingTable (remoteState .routingTable ())
1103
+ .metaData (MetaData .builder (remoteState .metaData ())
1104
+ .put (IndexMetaData .builder (remoteState .metaData ().index (indexName )).state (IndexMetaData .State .CLOSE ).build (), true )
1105
+ .build ())
1106
+ .build ();
1107
+ }
1108
+ }
1109
+
1110
+ final ClusterState finalRemoteState = remoteState ;
1111
+ final AtomicReference <ClusterState > lastModifiedClusterState = new AtomicReference <>(localState );
1112
+ final List <AutoFollowCoordinator .AutoFollowResult > results = new ArrayList <>();
1113
+ final Set <Object > followedIndices = ConcurrentCollections .newConcurrentSet ();
1114
+ final AutoFollower autoFollower =
1115
+ new AutoFollower ("remote" , results ::addAll , localClusterStateSupplier (localState ), () -> 1L , Runnable ::run ) {
1116
+ @ Override
1117
+ void getRemoteClusterState (String remoteCluster ,
1118
+ long metadataVersion ,
1119
+ BiConsumer <ClusterStateResponse , Exception > handler ) {
1120
+ assertThat (remoteCluster , equalTo ("remote" ));
1121
+ handler .accept (new ClusterStateResponse (new ClusterName ("remote" ), finalRemoteState , false ), null );
1122
+ }
1123
+
1124
+ @ Override
1125
+ void createAndFollow (Map <String , String > headers ,
1126
+ PutFollowAction .Request followRequest ,
1127
+ Runnable successHandler ,
1128
+ Consumer <Exception > failureHandler ) {
1129
+ followedIndices .add (followRequest .getLeaderIndex ());
1130
+ successHandler .run ();
1131
+ }
1132
+
1133
+ @ Override
1134
+ void updateAutoFollowMetadata (Function <ClusterState , ClusterState > updateFunction , Consumer <Exception > handler ) {
1135
+ lastModifiedClusterState .updateAndGet (updateFunction ::apply );
1136
+ handler .accept (null );
1137
+ }
1138
+
1139
+ @ Override
1140
+ void cleanFollowedRemoteIndices (ClusterState remoteClusterState , List <String > patterns ) {
1141
+ // Ignore, to avoid invoking updateAutoFollowMetadata(...) twice
1142
+ }
1143
+ };
1144
+ autoFollower .start ();
1145
+
1146
+ assertThat (results , notNullValue ());
1147
+ assertThat (results .size (), equalTo (1 ));
1148
+
1149
+ for (ObjectObjectCursor <String , IndexMetaData > index : remoteState .metaData ().indices ()) {
1150
+ boolean expect = index .value .getState () == IndexMetaData .State .OPEN ;
1151
+ assertThat (results .get (0 ).autoFollowExecutionResults .containsKey (index .value .getIndex ()), is (expect ));
1152
+ assertThat (followedIndices .contains (index .key ), is (expect ));
1153
+ }
1154
+ }
1155
+
1048
1156
private static ClusterState createRemoteClusterState (String indexName , boolean enableSoftDeletes ) {
1049
1157
Settings .Builder indexSettings ;
1050
1158
indexSettings = settings (Version .CURRENT ).put (IndexSettings .INDEX_SOFT_DELETES_SETTING .getKey (), enableSoftDeletes );
@@ -1067,19 +1175,21 @@ private static ClusterState createRemoteClusterState(String indexName, boolean e
1067
1175
1068
1176
private static ClusterState createRemoteClusterState (ClusterState previous , String indexName ) {
1069
1177
IndexMetaData indexMetaData = IndexMetaData .builder (indexName )
1070
- .settings (settings (Version .CURRENT ).put (IndexSettings .INDEX_SOFT_DELETES_SETTING .getKey (), true ))
1178
+ .settings (settings (Version .CURRENT )
1179
+ .put (IndexSettings .INDEX_SOFT_DELETES_SETTING .getKey (), true )
1180
+ .put (IndexMetaData .SETTING_INDEX_UUID , UUIDs .randomBase64UUID (random ())))
1071
1181
.numberOfShards (1 )
1072
1182
.numberOfReplicas (0 )
1073
1183
.build ();
1074
- ClusterState .Builder csBuilder = ClusterState .builder (new ClusterName ( "remote" ))
1184
+ ClusterState .Builder csBuilder = ClusterState .builder (previous . getClusterName ( ))
1075
1185
.metaData (MetaData .builder (previous .metaData ())
1076
1186
.version (previous .metaData ().version () + 1 )
1077
1187
.put (indexMetaData , true ));
1078
1188
1079
1189
ShardRouting shardRouting =
1080
1190
TestShardRouting .newShardRouting (indexName , 0 , "1" , true , ShardRoutingState .INITIALIZING ).moveToStarted ();
1081
1191
IndexRoutingTable indexRoutingTable = IndexRoutingTable .builder (indexMetaData .getIndex ()).addShard (shardRouting ).build ();
1082
- csBuilder .routingTable (RoutingTable .builder ().add (indexRoutingTable ).build ()).build ();
1192
+ csBuilder .routingTable (RoutingTable .builder (previous . routingTable () ).add (indexRoutingTable ).build ()).build ();
1083
1193
1084
1194
return csBuilder .build ();
1085
1195
}
0 commit comments