@ SuppressWarnings ( { "unchecked" , "rawtypes" } ) protected void addStoreToSession ( String store ) { Exception initializationException = null ; storeNames . add ( store ) ; for ( Node node : nodesToStream ) { SocketDestination destination = null ; SocketAndStreams sands = null ; try { destination = new SocketDestination ( node . getHost ( ) , node . getAdminPort ( ) , RequestFormatType . ADMIN_PROTOCOL_BUFFERS ) ; sands = streamingSocketPool . checkout ( destination ) ; DataOutputStream outputStream = sands . getOutputStream ( ) ; DataInputStream inputStream = sands . getInputStream ( ) ; nodeIdStoreToSocketRequest . put ( new Pair ( store , node . getId ( ) ) , destination ) ; nodeIdStoreToOutputStreamRequest . put ( new Pair ( store , node . getId ( ) ) , outputStream ) ; nodeIdStoreToInputStreamRequest . put ( new Pair ( store , node . getId ( ) ) , inputStream ) ; nodeIdStoreToSocketAndStreams . put ( new Pair ( store , node . getId ( ) ) , sands ) ; nodeIdStoreInitialized . put ( new Pair ( store , node . getId ( ) ) , false ) ; remoteStoreDefs = adminClient . metadataMgmtOps . getRemoteStoreDefList ( node . getId ( ) ) . getValue ( ) ; } catch ( Exception e ) { logger . error ( e ) ; try { close ( sands . getSocket ( ) ) ; streamingSocketPool . checkin ( destination , sands ) ; } catch ( Exception ioE ) { logger . error ( ioE ) ; } if ( ! faultyNodes . contains ( node . getId ( ) ) ) faultyNodes . add ( node . getId ( ) ) ; initializationException = e ; } } if ( initializationException != null ) throw new VoldemortException ( initializationException ) ; if ( store . equals ( "slop" ) ) return ; boolean foundStore = false ; for ( StoreDefinition remoteStoreDef : remoteStoreDefs ) { if ( remoteStoreDef . getName ( ) . equals ( store ) ) { RoutingStrategyFactory factory = new RoutingStrategyFactory ( ) ; RoutingStrategy storeRoutingStrategy = factory . updateRoutingStrategy ( remoteStoreDef , adminClient . getAdminClientCluster ( ) ) ; storeToRoutingStrategy . put ( store , storeRoutingStrategy ) ; validateSufficientNodesAvailable ( blackListedNodes , remoteStoreDef ) ; foundStore = true ; break ; } } if ( ! foundStore ) { logger . error ( "Store Name not found on the cluster" ) ; throw new VoldemortException ( "Store Name not found on the cluster" ) ; } }
Add another store destination to an existing streaming session