svn commit: r1570508 - in /hama/trunk: core/src/main/java/org/apache/hama/bsp/PartitioningRunner.java graph/src/main/java/org/apache/hama/graph/VertexInputReader.java

[edwardyoon]

Author: edwardyoon
Date: Fri Feb 21 10:07:24 2014
New Revision: 1570508

URL: http://svn.apache.org/r1570508
Log:
HAMA-874: The convertRecord() method should throw the Exception

Modified:
    hama/trunk/core/src/main/java/org/apache/hama/bsp/PartitioningRunner.java
    hama/trunk/graph/src/main/java/org/apache/hama/graph/VertexInputReader.java

Modified: 
hama/trunk/core/src/main/java/org/apache/hama/bsp/PartitioningRunner.java
URL: 
http://svn.apache.org/viewvc/hama/trunk/core/src/main/java/org/apache/hama/bsp/PartitioningRunner.java?rev=1570508&r1=1570507&r2=1570508&view=diff
==============================================================================
--- hama/trunk/core/src/main/java/org/apache/hama/bsp/PartitioningRunner.java 
(original)
+++ hama/trunk/core/src/main/java/org/apache/hama/bsp/PartitioningRunner.java 
Fri Feb 21 10:07:24 2014
@@ -92,9 +92,10 @@ public class PartitioningRunner extends
      * @param conf Configuration of the job.
      * @return the Key-Value pair instance of the expected sequential format.
      *         Should return null if the conversion was not successful.
+     * @throws IOException 
      */
     public KeyValuePair<Writable, Writable> convertRecord(
-        KeyValuePair<Writable, Writable> inputRecord, Configuration conf);
+        KeyValuePair<Writable, Writable> inputRecord, Configuration conf) 
throws IOException;
 
     public int getPartitionId(KeyValuePair<Writable, Writable> inputRecord,
         @SuppressWarnings("rawtypes")
@@ -157,8 +158,9 @@ public class PartitioningRunner extends
       convertedRecord = converter.convertRecord(rawRecord, conf);
 
       if (convertedRecord == null) {
-        continue;
+        throw new IOException("The converted record can't be null.");
       }
+      
       Writable convertedKey = convertedRecord.getKey();
       convertedKeyClass = convertedKey.getClass();
 

Modified: 
hama/trunk/graph/src/main/java/org/apache/hama/graph/VertexInputReader.java
URL: 
http://svn.apache.org/viewvc/hama/trunk/graph/src/main/java/org/apache/hama/graph/VertexInputReader.java?rev=1570508&r1=1570507&r2=1570508&view=diff
==============================================================================
--- hama/trunk/graph/src/main/java/org/apache/hama/graph/VertexInputReader.java 
(original)
+++ hama/trunk/graph/src/main/java/org/apache/hama/graph/VertexInputReader.java 
Fri Feb 21 10:07:24 2014
@@ -17,8 +17,8 @@
  */
 package org.apache.hama.graph;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+import java.io.IOException;
+
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableComparable;
@@ -41,8 +41,6 @@ import org.apache.hama.commons.util.KeyV
 public abstract class VertexInputReader<KEYIN extends Writable, VALUEIN 
extends Writable, V extends WritableComparable, E extends Writable, M extends 
Writable>
     implements RecordConverter {
 
-  private static final Log LOG = LogFactory.getLog(VertexInputReader.class);
-
   @SuppressWarnings("unchecked")
   @Override
   public void setup(Configuration conf) {
@@ -63,7 +61,8 @@ public abstract class VertexInputReader<
   @SuppressWarnings("unchecked")
   @Override
   public KeyValuePair<Writable, Writable> convertRecord(
-      KeyValuePair<Writable, Writable> inputRecord, Configuration conf) {
+      KeyValuePair<Writable, Writable> inputRecord, Configuration conf)
+      throws IOException {
     Class<Vertex<V, E, M>> vertexClass = (Class<Vertex<V, E, M>>) conf
         .getClass(GraphJob.VERTEX_CLASS_ATTR, Vertex.class);
     boolean vertexCreation;
@@ -73,11 +72,11 @@ public abstract class VertexInputReader<
       vertexCreation = parseVertex((KEYIN) inputRecord.getKey(),
           (VALUEIN) inputRecord.getValue(), vertex);
     } catch (Exception e) {
-      LOG.error("Error parsing vertex.", e);
       vertexCreation = false;
     }
     if (!vertexCreation) {
-      return null;
+      throw new IOException(
+          "Error parsing vertex. Please check your vertex input reader.");
     }
 
     outputRecord.setKey(vertex.getVertexID());