private static RestRepository initSingleIndex ( Settings settings , long currentInstance , Resource resource , Log log ) { if ( log . isDebugEnabled ( ) ) { log . debug ( String . format ( "Resource [%s] resolves as a single index" , resource ) ) ; } RestRepository repository = new RestRepository ( settings ) ; // create the index if needed if ( repository . touch ( ) ) { if ( repository . waitForYellow ( ) ) { log . warn ( String . format ( "Timed out waiting for index [%s] to reach yellow health" , resource ) ) ; } } // if WAN mode is used, use an already selected node if ( settings . getNodesWANOnly ( ) ) { String node = SettingsUtils . getPinnedNode ( settings ) ; if ( log . isDebugEnabled ( ) ) { log . debug ( String . format ( "Partition writer instance [%s] assigned to [%s]" , currentInstance , node ) ) ; } return repository ; } // if client-nodes are used, simply use the underlying nodes if ( settings . getNodesClientOnly ( ) ) { String clientNode = repository . getRestClient ( ) . getCurrentNode ( ) ; if ( log . isDebugEnabled ( ) ) { log . debug ( String . format ( "Client-node routing detected; partition writer instance [%s] assigned to [%s]" , currentInstance , clientNode ) ) ; } return repository ; } // no routing necessary; select the relevant target shard/node Map < ShardInfo , NodeInfo > targetShards = repository . getWriteTargetPrimaryShards ( settings . getNodesClientOnly ( ) ) ; repository . close ( ) ; Assert . isTrue ( ! targetShards . isEmpty ( ) , String . format ( "Cannot determine write shards for [%s]; likely its format is incorrect (maybe it contains illegal characters? or all shards failed?)" , resource ) ) ; List < ShardInfo > orderedShards = new ArrayList < ShardInfo > ( targetShards . keySet ( ) ) ; // make sure the order is strict Collections . sort ( orderedShards ) ; if ( log . isTraceEnabled ( ) ) { log . trace ( String . format ( "Partition writer instance [%s] discovered [%s] primary shards %s" , currentInstance , orderedShards . size ( ) , orderedShards ) ) ; } // if there's no task info, just pick a random bucket if ( currentInstance <= 0 ) { currentInstance = new Random ( ) . nextInt ( targetShards . size ( ) ) + 1 ; } int bucket = ( int ) ( currentInstance % targetShards . size ( ) ) ; ShardInfo chosenShard = orderedShards . get ( bucket ) ; NodeInfo targetNode = targetShards . get ( chosenShard ) ; // pin settings SettingsUtils . pinNode ( settings , targetNode . getPublishAddress ( ) ) ; String node = SettingsUtils . getPinnedNode ( settings ) ; repository = new RestRepository ( settings ) ; if ( log . isDebugEnabled ( ) ) { log . debug ( String . format ( "Partition writer instance [%s] assigned to primary shard [%s] at address [%s]" , currentInstance , chosenShard . getName ( ) , node ) ) ; } return repository ; }
Validate and configure a rest repository for writing to an index . The index is potentially created if it does not exist and the client is pinned to a node that hosts one of the index s primary shards based on its currentInstance number .