src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
/**
         * Loops collecting mutations from the queue and sending to Cassandra
         */
Loops collecting mutations from the queue and sending to Cassandra
Loops collecting mutations from the queue and sending to Cassandra
@Override
        public void run()
        {
            outer:
            while (run || !queue.isEmpty())
            {
                Pair<ByteBuffer, Mutation> mutation;
                try
                {
                    mutation = queue.take();
                }
                catch (InterruptedException e)
                {
                    continue;
                }
                Map<ByteBuffer, Map<String, List<Mutation>>> batch = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
                while (mutation != null)
                {
                    Map<String, List<Mutation>> subBatch = batch.get(mutation.left);
                    if (subBatch == null)
                    {
                        subBatch = Collections.singletonMap(columnFamily, (List<Mutation>) new ArrayList<Mutation>());
                        batch.put(mutation.left, subBatch);
                    }
                    subBatch.get(columnFamily).add(mutation.right);
                    if (batch.size() >= batchThreshold)
                        break;
                    mutation = queue.poll();
                }
                Iterator<InetAddress> iter = endpoints.iterator();
                while (true)
                {
                    try
                    {
                        thriftClient.batch_mutate(batch, consistencyLevel, null, LamportClock.COPS_UNSUPPORTED);
                        break;
                    }
                    catch (Exception e)
                    {
                        closeInternal();
                        if (!iter.hasNext())
                        {
                            lastException = new IOException(e);
                            break outer;
                        }
                    }
                    try
                    {
                        InetAddress address = iter.next();
                        thriftSocket = new TSocket(address.getHostName(), ConfigHelper.getOutputRpcPort(conf));
                        thriftClient = ColumnFamilyOutputFormat.createAuthenticatedClient(thriftSocket, conf);
                    }
                    catch (Exception e)
                    {
                        closeInternal();
                        if ((!(e instanceof TException)) || !iter.hasNext())
                        {
                            lastException = new IOException(e);
                            break outer;
                        }
                    }
                }
            }
        }
@ Override public void run ( ) { outer : while ( run || ! queue . isEmpty ( ) ) { Pair < ByteBuffer , Mutation > mutation ; try { mutation = queue . take ( ) ; } catch ( InterruptedException e ) { continue ; } Map < ByteBuffer , Map < String , List < Mutation > > > batch = new HashMap < ByteBuffer , Map < String , List < Mutation > > > ( ) ; while ( mutation != null ) { Map < String , List < Mutation > > subBatch = batch . get ( mutation . left ) ; if ( subBatch == null ) { subBatch = Collections . singletonMap ( columnFamily , ( List < Mutation > ) new ArrayList < Mutation > ( ) ) ; batch . put ( mutation . left , subBatch ) ; } subBatch . get ( columnFamily ) . add ( mutation . right ) ; if ( batch . size ( ) >= batchThreshold ) break ; mutation = queue . poll ( ) ; } Iterator < InetAddress > iter = endpoints . iterator ( ) ; while ( true ) { try { thriftClient . batch_mutate ( batch , consistencyLevel , null , LamportClock . COPS_UNSUPPORTED ) ; break ; } catch ( Exception e ) { closeInternal ( ) ; if ( ! iter . hasNext ( ) ) { lastException = new IOException ( e ) ; break outer ; } } try { InetAddress address = iter . next ( ) ; thriftSocket = new TSocket ( address . getHostName ( ) , ConfigHelper . getOutputRpcPort ( conf ) ) ; thriftClient = ColumnFamilyOutputFormat . createAuthenticatedClient ( thriftSocket , conf ) ; } catch ( Exception e ) { closeInternal ( ) ; if ( ( ! ( e instanceof TException ) ) || ! iter . hasNext ( ) ) { lastException = new IOException ( e ) ; break outer ; } } } } }
Loops collecting mutations from the queue and sending to Cassandra
Loops collecting mutations from the queue and sending to Cassandra
// re-check loop condition after interrupt // send the mutation to the last-used endpoint.  first time through, this will NPE harmlessly. //WL TODO: don't call thrift client directly if we want to support this hadoop stuff // attempt to connect to a different endpoint // TException means something unexpected went wrong to that endpoint, so // we should try again to another.  Other exceptions (auth or invalid request) are fatal.