cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jo...@apache.org
Subject svn commit: r946564 - /cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java
Date Thu, 20 May 2010 08:20:35 GMT
Author: johan
Date: Thu May 20 08:20:35 2010
New Revision: 946564

URL: http://svn.apache.org/viewvc?rev=946564&view=rev
Log:
Close thrift connections properly in the Hadoop record reader. Patch by johan, review by jbellis.
CASSANDRA-1081

Modified:
    cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java

Modified: cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java?rev=946564&r1=946563&r2=946564&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java
(original)
+++ cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java
Thu May 20 08:20:35 2010
@@ -56,7 +56,11 @@ public class ColumnFamilyRecordReader ex
     private String cfName;
     private String keyspace;
 
-    public void close() {}
+    public void close() 
+    {
+        if (iter != null)
+            iter.close();
+    }
     
     public String getCurrentKey()
     {
@@ -102,6 +106,7 @@ public class ColumnFamilyRecordReader ex
         private int totalRead = 0;
         private int i = 0;
         private AbstractType comparator = DatabaseDescriptor.getComparator(keyspace, cfName);
+        private TSocket socket;
 
         private void maybeInit()
         {
@@ -111,7 +116,11 @@ public class ColumnFamilyRecordReader ex
             
             if (rows != null)
                 return;
-            TSocket socket = new TSocket(getLocation(),
+            
+            // close previous connection if one is open
+            close();
+            
+            socket = new TSocket(getLocation(),
                                          DatabaseDescriptor.getThriftPort());
             TBinaryProtocol binaryProtocol = new TBinaryProtocol(socket, false, false);
             Cassandra.Client client = new Cassandra.Client(binaryProtocol);
@@ -226,6 +235,14 @@ public class ColumnFamilyRecordReader ex
             }
             return new Pair<String, SortedMap<byte[], IColumn>>(ks.key, map);
         }
+        
+        public void close() 
+        {
+            if (socket != null && socket.isOpen())
+            {
+                socket.close();
+            }
+        }
     }
 
     private IColumn unthriftify(ColumnOrSuperColumn cosc)



Mime
View raw message