public void run()
        {
            outer:
            while (run || !queue.isEmpty())
            {
                List<ByteBuffer> bindVariables;
                try
                {
                    bindVariables = queue.take();
                }
                catch (InterruptedException e)
                {
                    // re-check loop condition after interrupt
                    continue;
                }

                Iterator<InetAddress> iter = endpoints.iterator();
                while (true)
                {
                    // send the mutation to the last-used endpoint.  first time through, this will NPE harmlessly.
                    try
                    {
                        int i = 0;
                        int itemId = preparedStatement(client);
                        while (bindVariables != null)
                        {
                            client.execute_prepared_cql3_query(itemId, bindVariables, ConsistencyLevel.ONE);
                            i++;
                            
                            if (i >= batchThreshold)
                                break;
                            
                            bindVariables = queue.poll();
                        }
                        
                        break;
                    }
                    catch (Exception e)
                    {
                        closeInternal();
                        if (!iter.hasNext())
                        {
                            lastException = new IOException(e);
                            break outer;
                        }
                    }

                    // attempt to connect to a different endpoint
                    try
                    {
                        InetAddress address = iter.next();
                        String host = address.getHostName();
                        int port = ConfigHelper.getOutputRpcPort(conf);
                        client = CqlOutputFormat.createAuthenticatedClient(host, port, conf);
                    }
                    catch (Exception e)
                    {
                        closeInternal();
                        // TException means something unexpected went wrong to that endpoint, so
                        // we should try again to another.  Other exceptions (auth or invalid request) are fatal.
                        if ((!(e instanceof TException)) || !iter.hasNext())
                        {
                            lastException = new IOException(e);
                            break outer;
                        }
                    }
                }
            }

            // close all our connections once we are done.
            closeInternal();
        }
public void run ( ) { outer : while ( run || ! queue . isEmpty ( ) ) { List < ByteBuffer > bindVariables ; try { bindVariables = queue . take ( ) ; } catch ( InterruptedException e ) { continue ; } Iterator < InetAddress > iter = endpoints . iterator ( ) ; while ( true ) { try { int i = 0 ; int itemId = preparedStatement ( client ) ; while ( bindVariables != null ) { client . execute_prepared_cql3_query ( itemId , bindVariables , ConsistencyLevel . ONE ) ; i ++ ; if ( i >= batchThreshold ) break ; bindVariables = queue . poll ( ) ; } break ; } catch ( Exception e ) { closeInternal ( ) ; if ( ! iter . hasNext ( ) ) { lastException = new IOException ( e ) ; break outer ; } } try { InetAddress address = iter . next ( ) ; String host = address . getHostName ( ) ; int port = ConfigHelper . getOutputRpcPort ( conf ) ; client = CqlOutputFormat . createAuthenticatedClient ( host , port , conf ) ; } catch ( Exception e ) { closeInternal ( ) ; if ( ( ! ( e instanceof TException ) ) || ! iter . hasNext ( ) ) { lastException = new IOException ( e ) ; break outer ; } } } } closeInternal ( ) ; }
Loops collecting cql binded variable values from the queue and sending to Cassandra