svn commit: r1679339 - in /hama/trunk/core/src: main/java/org/apache/hama/bsp/BSPJobClient.java test/java/org/apache/hama/bsp/TestKeyValueTextInputFormat.java

[edwardyoon]

Author: edwardyoon
Date: Thu May 14 11:02:47 2015
New Revision: 1679339

URL: http://svn.apache.org/r1679339
Log:
debugging

Modified:
    hama/trunk/core/src/main/java/org/apache/hama/bsp/BSPJobClient.java
    
hama/trunk/core/src/test/java/org/apache/hama/bsp/TestKeyValueTextInputFormat.java

Modified: hama/trunk/core/src/main/java/org/apache/hama/bsp/BSPJobClient.java
URL: 
http://svn.apache.org/viewvc/hama/trunk/core/src/main/java/org/apache/hama/bsp/BSPJobClient.java?rev=1679339&r1=1679338&r2=1679339&view=diff
==============================================================================
--- hama/trunk/core/src/main/java/org/apache/hama/bsp/BSPJobClient.java 
(original)
+++ hama/trunk/core/src/main/java/org/apache/hama/bsp/BSPJobClient.java Thu May 
14 11:02:47 2015
@@ -58,6 +58,11 @@ import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.util.ToolRunner;
 import org.apache.hama.Constants;
 import org.apache.hama.HamaConfiguration;
+import org.apache.hama.bsp.message.MessageManager;
+import org.apache.hama.bsp.message.OutgoingMessageManager;
+import org.apache.hama.bsp.message.OutgoingPOJOMessageBundle;
+import org.apache.hama.bsp.message.queue.MemoryQueue;
+import org.apache.hama.bsp.message.queue.MessageQueue;
 import org.apache.hama.ipc.HamaRPCProtocolVersion;
 import org.apache.hama.ipc.JobSubmissionProtocol;
 import org.apache.hama.ipc.RPC;
@@ -460,6 +465,13 @@ public class BSPJobClient extends Config
         LOG.debug("partitioningJob input: "
             + partitioningJob.get(Constants.JOB_INPUT_DIR));
 
+        partitioningJob.getConfiguration().setClass(
+            MessageManager.OUTGOING_MESSAGE_MANAGER_CLASS,
+            OutgoingPOJOMessageBundle.class, OutgoingMessageManager.class);
+        partitioningJob.getConfiguration().setClass(
+            MessageManager.RECEIVE_QUEUE_TYPE_CLASS, MemoryQueue.class,
+            MessageQueue.class);
+
         partitioningJob.setBoolean(Constants.FORCE_SET_BSP_TASKS, true);
         partitioningJob.setInputFormat(job.getInputFormat().getClass());
         partitioningJob.setInputKeyClass(job.getInputKeyClass());

Modified: 
hama/trunk/core/src/test/java/org/apache/hama/bsp/TestKeyValueTextInputFormat.java
URL: 
http://svn.apache.org/viewvc/hama/trunk/core/src/test/java/org/apache/hama/bsp/TestKeyValueTextInputFormat.java?rev=1679339&r1=1679338&r2=1679339&view=diff
==============================================================================
--- 
hama/trunk/core/src/test/java/org/apache/hama/bsp/TestKeyValueTextInputFormat.java
 (original)
+++ 
hama/trunk/core/src/test/java/org/apache/hama/bsp/TestKeyValueTextInputFormat.java
 Thu May 14 11:02:47 2015
@@ -93,7 +93,7 @@ public class TestKeyValueTextInputFormat
       } //while (peer.readNext(key, value) != false)
       message.put(new Text(KeyValueHashPartitionedBSP.TEST_INPUT_VALUES), 
expectedKeys);
       
-      int master = peer.getNumPeers()/2;
+      int master = peer.getNumPeers() / 2;
       String masterName = peer.getPeerName(master);
       peer.send(masterName, message);
       peer.sync();
@@ -106,6 +106,7 @@ public class TestKeyValueTextInputFormat
 
         while ( (msg = peer.getCurrentMessage()) != null ) {
           blValue = (BooleanWritable) msg.get(new 
Text(KeyValueHashPartitionedBSP.TEST_UNEXPECTED_KEYS));
+          System.out.println(">>>>> " + peer.getPeerName() + ", "+ 
blValue.get());
           assertEquals(false, blValue.get());
           values = (MapWritable) msg.get(new 
Text(KeyValueHashPartitionedBSP.TEST_INPUT_VALUES));
           for (Map.Entry<Writable,Writable> w : values.entrySet()) {