svn commit: r1075627 - /cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java

[jbellis]

Author: jbellis
Date: Tue Mar  1 02:52:03 2011
New Revision: 1075627

URL: http://svn.apache.org/viewvc?rev=1075627&view=rev
Log:
fix Hadoop ColumnFamilyOutputFormat droppingof mutations
patch by Eldon Stegall and Jeremy Hanna; reviewed by jbellis for CASSANDRA-2255

Modified:
    
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java

Modified: 
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
URL: 
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java?rev=1075627&r1=1075626&r2=1075627&view=diff
==============================================================================
--- 
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
 (original)
+++ 
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
 Tue Mar  1 02:52:03 2011
@@ -325,7 +325,7 @@ implements org.apache.hadoop.mapred.Reco
                 }
 
                 Map<ByteBuffer, Map<String, List<Mutation>>> batch = new 
HashMap<ByteBuffer, Map<String, List<Mutation>>>();
-                while (batch.size() < batchThreshold)
+                while (mutation != null)
                 {
                     Map<String, List<Mutation>> subBatch = 
batch.get(mutation.left);
                     if (subBatch == null)
@@ -333,10 +333,12 @@ implements org.apache.hadoop.mapred.Reco
                         subBatch = Collections.singletonMap(columnFamily, 
(List<Mutation>) new ArrayList<Mutation>());
                         batch.put(mutation.left, subBatch);
                     }
-                    
+
                     subBatch.get(columnFamily).add(mutation.right);
-                    if ((mutation = queue.poll()) == null)
+                    if (batch.size() >= batchThreshold)
                         break;
+
+                    mutation = queue.poll();
                 }
 
                 Iterator<InetAddress> iter = endpoints.iterator();