private void runQuery ( ) { if ( this . rowKeysList != null && this . rowKeysList . size ( ) > 0 ) { // Check if there are rowkeys to query Cassandra if ( threadCount > 1 ) { // When thread count greater than 1 enables parallelism, use threads to query Cassandra // multiple times ExecutorService executor = Executors . newFixedThreadPool ( threadCount ) ; List < Future < ? > > futures = new LinkedList < Future < ? > > ( ) ; for ( final List < K > param : this . rowKeysList ) { Future < ? > future = executor . submit ( new Runnable ( ) { public void run ( ) { // Query Cassandra with the input keys provided runMultigetSliceQuery ( param ) ; } } ) ; futures . add ( future ) ; } for ( Future < ? > f : futures ) { // iterate through thread results try { f . get ( ) ; // wait for thread to complete } catch ( InterruptedException e ) { throw new HectorException ( "Failed to retrieve rows from Cassandra." , e ) ; } catch ( ExecutionException e ) { throw new HectorException ( "Failed to retrieve rows from Cassandra." , e ) ; } } // Safe to shutdown the threadpool and release the resources executor . shutdown ( ) ; // set the rowKeysIndex to size of input keys so as no further calls // will be made to Cassandra. // This ensures iterator.hasNext() returns false when all keys are // queried rowKeysIndex = this . rowKeysList . size ( ) ; } else { // When thread count less than or equal to 1 (0 or negative) disables // parallelism, set of(maxRowCountPerQuery) keys queries // Cassandra at a time runMultigetSliceQuery ( this . rowKeysList . get ( rowKeysIndex ) ) ; // Increment the rowKeyIndex instead of setting it to this.rowKeysList.size(); rowKeysIndex ++ ; } } ArrayList < Row < K , N , V > > resultList = new ArrayList < Row < K , N , V > > ( queryResult . size ( ) ) ; synchronized ( queryResult ) { // Ensure that runMultigetSliceQuery() method call updates global // variable queryResult with query result (if exists) if ( queryResult != null && queryResult . size ( ) > 0 ) { for ( Rows < K , N , V > rows : queryResult ) { if ( rows != null && rows . getCount ( ) > 0 ) { for ( Row < K , N , V > row : rows ) { // prepare List<Row<K, N, V>> to return // the iterator of <Row<K,N,V>> to the caller resultList . add ( row ) ; } } } } } // assign global iterator with the result of multigetSliceQuery iterator = resultList . iterator ( ) ; }
This method prepares keys for execution determines whether to use parallelism or not to query Cassandra executes the query and collects the result