protected < C extends FlagAffectedCommand & TopologyAffectedCommand > CompletionStage < Void > remoteGetSingleKey ( InvocationContext ctx , C command , Object key , boolean isWrite ) { LocalizedCacheTopology cacheTopology = checkTopologyId ( command ) ; int topologyId = cacheTopology . getTopologyId ( ) ; DistributionInfo info = retrieveDistributionInfo ( cacheTopology , command , key ) ; if ( info . isReadOwner ( ) ) { if ( trace ) { log . tracef ( "Key %s became local after wrapping, retrying command. Command topology is %d, current topology is %d" , key , command . getTopologyId ( ) , topologyId ) ; } // The topology has changed between EWI and BDI, let's retry if ( command . getTopologyId ( ) == topologyId ) { throw new IllegalStateException ( ) ; } throw OutdatedTopologyException . RETRY_NEXT_TOPOLOGY ; } if ( trace ) { log . tracef ( "Perform remote get for key %s. currentTopologyId=%s, owners=%s" , key , topologyId , info . readOwners ( ) ) ; } ClusteredGetCommand getCommand = cf . buildClusteredGetCommand ( key , info . segmentId ( ) , command . getFlagsBitSet ( ) ) ; getCommand . setTopologyId ( topologyId ) ; getCommand . setWrite ( isWrite ) ; return rpcManager . invokeCommandStaggered ( info . readOwners ( ) , getCommand , new RemoteGetSingleKeyCollector ( ) , rpcManager . getSyncRpcOptions ( ) ) . thenAccept ( response -> { Object responseValue = response . getResponseValue ( ) ; if ( responseValue == null ) { if ( rvrl != null ) { rvrl . remoteValueNotFound ( key ) ; } wrapRemoteEntry ( ctx , key , NullCacheEntry . getInstance ( ) , isWrite ) ; return ; } InternalCacheEntry ice = ( ( InternalCacheValue ) responseValue ) . toInternalCacheEntry ( key ) ; if ( rvrl != null ) { rvrl . remoteValueFound ( ice ) ; } wrapRemoteEntry ( ctx , key , ice , isWrite ) ; } ) ; }
Fetch a key from its remote owners and store it in the context .