svn commit: r731856 - /incubator/hama/trunk/src/java/org/apache/hama/mapred/VectorInputFormat.java

[ed...@apache.org]

Author: edwardyoon
Date: Mon Jan  5 23:24:14 2009
New Revision: 731856

URL: http://svn.apache.org/viewvc?rev=731856&view=rev
Log:
Sorry for my trivial updates.

Modified:
    incubator/hama/trunk/src/java/org/apache/hama/mapred/VectorInputFormat.java

Modified: incubator/hama/trunk/src/java/org/apache/hama/mapred/VectorInputFormat.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/mapred/VectorInputFormat.java?rev=731856&r1=731855&r2=731856&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/mapred/VectorInputFormat.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/mapred/VectorInputFormat.java Mon Jan  5 23:24:14 2009
@@ -21,6 +21,9 @@
 
 import java.io.IOException;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.UnknownScannerException;
 import org.apache.hadoop.hbase.io.RowResult;
 import org.apache.hadoop.hbase.mapred.TableSplit;
 import org.apache.hadoop.hbase.util.Writables;
@@ -31,11 +34,13 @@
 import org.apache.hadoop.mapred.JobConfigurable;
 import org.apache.hadoop.mapred.RecordReader;
 import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.util.StringUtils;
 import org.apache.hama.io.VectorWritable;
 import org.apache.hama.util.BytesUtil;
 
 public class VectorInputFormat extends TableInputFormatBase implements
     InputFormat<IntWritable, VectorWritable>, JobConfigurable {
+  static final Log LOG = LogFactory.getLog(VectorInputFormat.class);
   private TableRecordReader tableRecordReader;
   
   /**
@@ -73,10 +78,21 @@
      */
     public boolean next(IntWritable key, VectorWritable value)
         throws IOException {
-      RowResult result = this.scanner.next();
+      RowResult result;
+      try {
+        result = this.scanner.next();
+      } catch (UnknownScannerException e) {
+        LOG.debug("recovered from " + StringUtils.stringifyException(e));  
+        restart(lastRow);
+        this.scanner.next();    // skip presumed already mapped row
+        result = this.scanner.next();
+      }
+      
       boolean hasMore = result != null && result.size() > 0;
       if (hasMore) {
-        key.set(BytesUtil.bytesToInt(result.getRow()));
+        byte[] row = result.getRow();
+        key.set(BytesUtil.bytesToInt(row));
+        lastRow = row;
         Writables.copyWritable(result, value);
       }
       return hasMore;