diff --git a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/NLJoinPOperator.java b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/NLJoinPOperator.java
index b35c6f9..a2e697d 100644
--- a/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/NLJoinPOperator.java
+++ b/algebricks/algebricks-core/src/main/java/edu/uci/ics/hyracks/algebricks/core/algebra/operators/physical/NLJoinPOperator.java
@@ -57,13 +57,8 @@
 import edu.uci.ics.hyracks.dataflow.std.join.NestedLoopJoinOperatorDescriptor;
 
 /**
- * <<<<<<< .working
- * Left input is broadcast and preserves its local properties. Right input can
- * be partitioned in any way.
- * =======
  * Left input is broadcast and preserves its local properties.
  * Right input can be partitioned in any way.
- * >>>>>>> .merge-right.r3014
  */
 public class NLJoinPOperator extends AbstractJoinPOperator {
 
@@ -290,4 +285,4 @@
         }
 
     }
-}
\ No newline at end of file
+}
diff --git a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/NestedPlansAccumulatingAggregatorFactory.java b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/NestedPlansAccumulatingAggregatorFactory.java
index 09354a5..af4bff2 100644
--- a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/NestedPlansAccumulatingAggregatorFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/NestedPlansAccumulatingAggregatorFactory.java
@@ -128,8 +128,8 @@
             }
 
             @Override
-            public boolean outputPartialResult(ArrayTupleBuilder tupleBuilder, IFrameTupleAccessor accessor, int tIndex,
-                    AggregateState state) throws HyracksDataException {
+            public boolean outputPartialResult(ArrayTupleBuilder tupleBuilder, IFrameTupleAccessor accessor,
+                    int tIndex, AggregateState state) throws HyracksDataException {
                 throw new IllegalStateException("this method should not be called");
             }
 
diff --git a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/NestedPlansRunningAggregatorFactory.java b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/NestedPlansRunningAggregatorFactory.java
index 8973c75..918be11 100644
--- a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/NestedPlansRunningAggregatorFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/NestedPlansRunningAggregatorFactory.java
@@ -239,7 +239,7 @@
 
         @Override
         public void close() throws HyracksDataException {
-            if(outputAppender.getTupleCount() > 0){
+            if (outputAppender.getTupleCount() > 0) {
                 FrameUtils.flushFrame(outputFrame, outputWriter);
                 outputAppender.reset(outputFrame, true);
             }
diff --git a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/SimpleAlgebricksAccumulatingAggregatorFactory.java b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/SimpleAlgebricksAccumulatingAggregatorFactory.java
index 9b638b4..11a7a5c 100644
--- a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/SimpleAlgebricksAccumulatingAggregatorFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/aggreg/SimpleAlgebricksAccumulatingAggregatorFactory.java
@@ -119,8 +119,8 @@
             }
 
             @Override
-            public boolean outputPartialResult(ArrayTupleBuilder tupleBuilder, IFrameTupleAccessor accessor, int tIndex,
-                    AggregateState state) throws HyracksDataException {
+            public boolean outputPartialResult(ArrayTupleBuilder tupleBuilder, IFrameTupleAccessor accessor,
+                    int tIndex, AggregateState state) throws HyracksDataException {
                 throw new IllegalStateException("this method should not be called");
             }
 
diff --git a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/sort/InMemorySortRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/sort/InMemorySortRuntimeFactory.java
index 131eea0..54ed192 100644
--- a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/sort/InMemorySortRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/sort/InMemorySortRuntimeFactory.java
@@ -57,8 +57,8 @@
             @Override
             public void open() throws HyracksDataException {
                 if (frameSorter == null) {
-                    frameSorter = new FrameSorterMergeSort(ctx, sortFields, firstKeyNormalizerFactory, comparatorFactories,
-                            outputRecordDesc);
+                    frameSorter = new FrameSorterMergeSort(ctx, sortFields, firstKeyNormalizerFactory,
+                            comparatorFactories, outputRecordDesc);
                 }
                 frameSorter.reset();
                 writer.open();
diff --git a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/std/PartitioningSplitOperatorDescriptor.java b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/std/PartitioningSplitOperatorDescriptor.java
index 048dbc9..d19dd34 100644
--- a/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/std/PartitioningSplitOperatorDescriptor.java
+++ b/algebricks/algebricks-runtime/src/main/java/edu/uci/ics/hyracks/algebricks/runtime/operators/std/PartitioningSplitOperatorDescriptor.java
@@ -41,11 +41,11 @@
 public class PartitioningSplitOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
     private static final long serialVersionUID = 1L;
     public static int NO_DEFAULT_BRANCH = -1;
-    
+
     private final ICopyEvaluatorFactory[] evalFactories;
     private final IBinaryBooleanInspector boolInspector;
     private final int defaultBranchIndex;
-    
+
     public PartitioningSplitOperatorDescriptor(IOperatorDescriptorRegistry spec, ICopyEvaluatorFactory[] evalFactories,
             IBinaryBooleanInspector boolInspector, int defaultBranchIndex, RecordDescriptor rDesc) {
         super(spec, 1, (defaultBranchIndex == evalFactories.length) ? evalFactories.length + 1 : evalFactories.length);
@@ -66,14 +66,15 @@
             private final ByteBuffer[] writeBuffers = new ByteBuffer[outputArity];
             private final ICopyEvaluator[] evals = new ICopyEvaluator[outputArity];
             private final ArrayBackedValueStorage evalBuf = new ArrayBackedValueStorage();
-            private final RecordDescriptor inOutRecDesc = recordDescProvider.getInputRecordDescriptor(getActivityId(), 0);
+            private final RecordDescriptor inOutRecDesc = recordDescProvider.getInputRecordDescriptor(getActivityId(),
+                    0);
             private final FrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), inOutRecDesc);
             private final FrameTupleReference frameTuple = new FrameTupleReference();
-            
+
             private final FrameTupleAppender tupleAppender = new FrameTupleAppender(ctx.getFrameSize());
             private final ArrayTupleBuilder tupleBuilder = new ArrayTupleBuilder(inOutRecDesc.getFieldCount());
             private final DataOutput tupleDos = tupleBuilder.getDataOutput();
-            
+
             @Override
             public void close() throws HyracksDataException {
                 // Flush (possibly not full) buffers that have data, and close writers.
@@ -102,28 +103,28 @@
                     boolean found = false;
                     for (int j = 0; j < evals.length; j++) {
                         try {
-                        	evalBuf.reset();
+                            evalBuf.reset();
                             evals[j].evaluate(frameTuple);
                         } catch (AlgebricksException e) {
                             throw new HyracksDataException(e);
                         }
                         found = boolInspector.getBooleanValue(evalBuf.getByteArray(), 0, 1);
                         if (found) {
-                        	copyAndAppendTuple(j);
-                        	break;
+                            copyAndAppendTuple(j);
+                            break;
                         }
                     }
                     // Optionally write to default output branch.
                     if (!found && defaultBranchIndex != NO_DEFAULT_BRANCH) {
-                    	copyAndAppendTuple(defaultBranchIndex);
+                        copyAndAppendTuple(defaultBranchIndex);
                     }
                 }
             }
 
             private void copyAndAppendTuple(int outputIndex) throws HyracksDataException {
-            	// Copy tuple into tuple builder.
+                // Copy tuple into tuple builder.
                 try {
-                	tupleBuilder.reset();
+                    tupleBuilder.reset();
                     for (int i = 0; i < frameTuple.getFieldCount(); i++) {
                         tupleDos.write(frameTuple.getFieldData(i), frameTuple.getFieldStart(i),
                                 frameTuple.getFieldLength(i));
@@ -134,15 +135,17 @@
                 }
                 // Append to frame.
                 tupleAppender.reset(writeBuffers[outputIndex], false);
-                if (!tupleAppender.append(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray(), 0, tupleBuilder.getSize())) {
+                if (!tupleAppender.append(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray(), 0,
+                        tupleBuilder.getSize())) {
                     FrameUtils.flushFrame(writeBuffers[outputIndex], writers[outputIndex]);
                     tupleAppender.reset(writeBuffers[outputIndex], true);
-                    if (!tupleAppender.append(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray(), 0, tupleBuilder.getSize())) {
+                    if (!tupleAppender.append(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray(), 0,
+                            tupleBuilder.getSize())) {
                         throw new IllegalStateException();
                     }
                 }
             }
-            
+
             @Override
             public void open() throws HyracksDataException {
                 for (IFrameWriter writer : writers) {
@@ -155,13 +158,13 @@
                     tupleAppender.reset(writeBuffers[i], true);
                 }
                 // Create evaluators for partitioning.
-				try {
-					for (int i = 0; i < evalFactories.length; i++) {
-						evals[i] = evalFactories[i].createEvaluator(evalBuf);
-					}
-				} catch (AlgebricksException e) {
-					throw new HyracksDataException(e);
-				}
+                try {
+                    for (int i = 0; i < evalFactories.length; i++) {
+                        evals[i] = evalFactories[i].createEvaluator(evalBuf);
+                    }
+                } catch (AlgebricksException e) {
+                    throw new HyracksDataException(e);
+                }
             }
 
             @Override
@@ -171,4 +174,3 @@
         };
     }
 }
-
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/application/ICCApplicationContext.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/application/ICCApplicationContext.java
index 7a795da..1ca4c6d 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/application/ICCApplicationContext.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/application/ICCApplicationContext.java
@@ -58,4 +58,5 @@
      * @return The Cluster Controller Context.
      */
     public ICCContext getCCContext();
+
 }
\ No newline at end of file
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/application/IClusterLifecycleListener.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/application/IClusterLifecycleListener.java
index 51db13e..5a567b4 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/application/IClusterLifecycleListener.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/application/IClusterLifecycleListener.java
@@ -5,7 +5,7 @@
  * you may obtain a copy of the License from
  * 
  *     http://www.apache.org/licenses/LICENSE-2.0
- * 
+ *     
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/HyracksClientInterfaceFunctions.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/HyracksClientInterfaceFunctions.java
index 55d80e2..9ff741e 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/HyracksClientInterfaceFunctions.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/HyracksClientInterfaceFunctions.java
@@ -31,6 +31,7 @@
         GET_CLUSTER_TOPOLOGY,
         CREATE_JOB,
         GET_JOB_STATUS,
+        GET_JOB_INFO,
         START_JOB,
         GET_DATASET_DIRECTORY_SERIVICE_INFO,
         GET_DATASET_RESULT_STATUS,
@@ -76,6 +77,25 @@
         }
     }
 
+    public static class GetJobInfoFunction extends Function {
+        private static final long serialVersionUID = 1L;
+
+        private final JobId jobId;
+
+        public GetJobInfoFunction(JobId jobId) {
+            this.jobId = jobId;
+        }
+
+        @Override
+        public FunctionId getFunctionId() {
+            return FunctionId.GET_JOB_INFO;
+        }
+
+        public JobId getJobId() {
+            return jobId;
+        }
+    }
+
     public static class StartJobFunction extends Function {
         private static final long serialVersionUID = 1L;
 
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/HyracksClientInterfaceRemoteProxy.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/HyracksClientInterfaceRemoteProxy.java
index 2bac49f..98f27f2 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/HyracksClientInterfaceRemoteProxy.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/HyracksClientInterfaceRemoteProxy.java
@@ -23,6 +23,7 @@
 import edu.uci.ics.hyracks.api.deployment.DeploymentId;
 import edu.uci.ics.hyracks.api.job.JobFlag;
 import edu.uci.ics.hyracks.api.job.JobId;
+import edu.uci.ics.hyracks.api.job.JobInfo;
 import edu.uci.ics.hyracks.api.job.JobStatus;
 import edu.uci.ics.hyracks.api.topology.ClusterTopology;
 import edu.uci.ics.hyracks.ipc.api.IIPCHandle;
@@ -103,4 +104,11 @@
                 deploymentId);
         rpci.call(ipcHandle, dbf);
     }
+
+    @Override
+    public JobInfo getJobInfo(JobId jobId) throws Exception {
+        HyracksClientInterfaceFunctions.GetJobInfoFunction gjsf = new HyracksClientInterfaceFunctions.GetJobInfoFunction(
+                jobId);
+        return (JobInfo) rpci.call(ipcHandle, gjsf);
+    }
 }
\ No newline at end of file
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/HyracksConnection.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/HyracksConnection.java
index 98836cd..1916360 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/HyracksConnection.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/HyracksConnection.java
@@ -36,6 +36,7 @@
 import edu.uci.ics.hyracks.api.job.IActivityClusterGraphGeneratorFactory;
 import edu.uci.ics.hyracks.api.job.JobFlag;
 import edu.uci.ics.hyracks.api.job.JobId;
+import edu.uci.ics.hyracks.api.job.JobInfo;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.api.job.JobStatus;
 import edu.uci.ics.hyracks.api.topology.ClusterTopology;
@@ -148,7 +149,7 @@
                 binaryURLs.add(new URL(url));
             }
         }
-        /**deploy the URLs to the CC and NCs*/
+        /** deploy the URLs to the CC and NCs */
         hci.deployBinary(binaryURLs, deploymentId);
         return deploymentId;
     }
@@ -176,4 +177,9 @@
             EnumSet<JobFlag> jobFlags) throws Exception {
         return hci.startJob(deploymentId, JavaSerializationUtils.serialize(acggf), jobFlags);
     }
+
+    @Override
+    public JobInfo getJobInfo(JobId jobId) throws Exception {
+        return hci.getJobInfo(jobId);
+    }
 }
\ No newline at end of file
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/IHyracksClientConnection.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/IHyracksClientConnection.java
index 2820cdf..1e44e91 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/IHyracksClientConnection.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/IHyracksClientConnection.java
@@ -23,6 +23,7 @@
 import edu.uci.ics.hyracks.api.job.IActivityClusterGraphGeneratorFactory;
 import edu.uci.ics.hyracks.api.job.JobFlag;
 import edu.uci.ics.hyracks.api.job.JobId;
+import edu.uci.ics.hyracks.api.job.JobInfo;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.api.job.JobStatus;
 import edu.uci.ics.hyracks.api.topology.ClusterTopology;
@@ -44,6 +45,16 @@
     public JobStatus getJobStatus(JobId jobId) throws Exception;
 
     /**
+     * Gets detailed information about the specified Job.
+     * 
+     * @param jobId
+     *            JobId of the Job
+     * @return {@link JobStatus}
+     * @throws Exception
+     */
+    public JobInfo getJobInfo(JobId jobId) throws Exception;
+
+    /**
      * Start the specified Job.
      * 
      * @param appName
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/IHyracksClientInterface.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/IHyracksClientInterface.java
index 737152b..d0eeada 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/IHyracksClientInterface.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/IHyracksClientInterface.java
@@ -23,6 +23,7 @@
 import edu.uci.ics.hyracks.api.deployment.DeploymentId;
 import edu.uci.ics.hyracks.api.job.JobFlag;
 import edu.uci.ics.hyracks.api.job.JobId;
+import edu.uci.ics.hyracks.api.job.JobInfo;
 import edu.uci.ics.hyracks.api.job.JobStatus;
 import edu.uci.ics.hyracks.api.topology.ClusterTopology;
 
@@ -46,4 +47,7 @@
     public void unDeployBinary(DeploymentId deploymentId) throws Exception;
 
     public JobId startJob(DeploymentId deploymentId, byte[] acggfBytes, EnumSet<JobFlag> jobFlags) throws Exception;
+
+    public JobInfo getJobInfo(JobId jobId) throws Exception;
+
 }
\ No newline at end of file
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/impl/JobSpecificationActivityClusterGraphGeneratorFactory.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/impl/JobSpecificationActivityClusterGraphGeneratorFactory.java
index 79efaa0..48d7275 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/impl/JobSpecificationActivityClusterGraphGeneratorFactory.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/client/impl/JobSpecificationActivityClusterGraphGeneratorFactory.java
@@ -43,6 +43,11 @@
     }
 
     @Override
+    public JobSpecification getJobSpecification() {
+        return spec;
+    }
+
+    @Override
     public IActivityClusterGraphGenerator createActivityClusterGraphGenerator(JobId jobId,
             final ICCApplicationContext ccAppCtx, EnumSet<JobFlag> jobFlags) throws HyracksException {
         final JobActivityGraphBuilder builder = new JobActivityGraphBuilder(spec, jobFlags);
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/comm/IFrameWriter.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/comm/IFrameWriter.java
index 83aa0f2..8e35dda 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/comm/IFrameWriter.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/comm/IFrameWriter.java
@@ -19,8 +19,7 @@
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 
 /**
- * {@link IFrameWriter} is the interface implemented by a stream consumer. An
- * {@link IFrameWriter} could be in one of the following states:
+ * {@link IFrameWriter} is the interface implemented by a stream consumer. An {@link IFrameWriter} could be in one of the following states:
  * <ul>
  * <li>INITIAL</li>
  * <li>OPENED</li>
@@ -29,33 +28,21 @@
  * </ul>
  * A producer follows the following protocol when using an {@link IFrameWriter}.
  * Initially, the {@link IFrameWriter} is in the INITIAL state.
- * The first valid call to an {@link IFrameWriter} is always the
- * {@link IFrameWriter#open()}. This call provides the opportunity for the
- * {@link IFrameWriter} implementation to allocate any resources for its
+ * The first valid call to an {@link IFrameWriter} is always the {@link IFrameWriter#open()}. This call provides the opportunity for the {@link IFrameWriter} implementation to allocate any resources for its
  * processing. Once this call returns, the {@link IFrameWriter} is in the OPENED
  * state. If an error occurs
- * during the {@link IFrameWriter#open()} call, a {@link HyracksDataException}
- * is thrown and it stays in the INITIAL state.
+ * during the {@link IFrameWriter#open()} call, a {@link HyracksDataException} is thrown and it stays in the INITIAL state.
  * While the {@link IFrameWriter} is in the OPENED state, the producer can call
  * one of:
  * <ul>
- * <li> {@link IFrameWriter#close()} to give up any resources owned by the
- * {@link IFrameWriter} and enter the CLOSED state.</li>
- * <li> {@link IFrameWriter#nextFrame(ByteBuffer)} to provide data to the
- * {@link IFrameWriter}. The call returns normally on success and the
- * {@link IFrameWriter} remains in the OPENED state. On failure, the call throws
- * a {@link HyracksDataException}, and the {@link IFrameWriter} enters the ERROR
- * state.</li>
- * <li> {@link IFrameWriter#fail()} to indicate that stream is to be aborted. The
- * {@link IFrameWriter} enters the FAILED state.</li>
+ * <li> {@link IFrameWriter#close()} to give up any resources owned by the {@link IFrameWriter} and enter the CLOSED state.</li>
+ * <li> {@link IFrameWriter#nextFrame(ByteBuffer)} to provide data to the {@link IFrameWriter}. The call returns normally on success and the {@link IFrameWriter} remains in the OPENED state. On failure, the call throws a {@link HyracksDataException}, and the {@link IFrameWriter} enters the ERROR state.</li>
+ * <li> {@link IFrameWriter#fail()} to indicate that stream is to be aborted. The {@link IFrameWriter} enters the FAILED state.</li>
  * </ul>
- * In the FAILED state, the only call allowed is the
- * {@link IFrameWriter#close()} to move the {@link IFrameWriter} into the CLOSED
+ * In the FAILED state, the only call allowed is the {@link IFrameWriter#close()} to move the {@link IFrameWriter} into the CLOSED
  * state and give up all resources.
  * No calls are allowed when the {@link IFrameWriter} is in the CLOSED state.
- * 
- * Note: If the call to {@link IFrameWriter#open()} failed, the
- * {@link IFrameWriter#close()} is not called by the producer. So an exceptional
+ * Note: If the call to {@link IFrameWriter#open()} failed, the {@link IFrameWriter#close()} is not called by the producer. So an exceptional
  * return from the {@link IFrameWriter#open()} call must clean up all partially
  * allocated resources.
  * 
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/context/IHyracksCommonContext.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/context/IHyracksCommonContext.java
index 1af32de..965ade7 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/context/IHyracksCommonContext.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/context/IHyracksCommonContext.java
@@ -25,6 +25,6 @@
     public IIOManager getIOManager();
 
     public ByteBuffer allocateFrame() throws HyracksDataException;
-    
+
     public void deallocateFrames(int frameCount);
 }
\ No newline at end of file
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/ILinearizeComparatorFactory.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/ILinearizeComparatorFactory.java
index a06b311f..d4b158b 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/ILinearizeComparatorFactory.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/ILinearizeComparatorFactory.java
@@ -14,7 +14,6 @@
  */
 package edu.uci.ics.hyracks.api.dataflow.value;
 
-
 public interface ILinearizeComparatorFactory extends IBinaryComparatorFactory {
     public ILinearizeComparator createBinaryComparator();
 }
\ No newline at end of file
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/IPredicateEvaluator.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/IPredicateEvaluator.java
index 575472c..d42d714 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/IPredicateEvaluator.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/IPredicateEvaluator.java
@@ -21,5 +21,5 @@
  * Compares two tuples to make sure that records, whose comparison keys are NULL do not pass comparator filter  
  */
 public interface IPredicateEvaluator {
-	public boolean evaluate(IFrameTupleAccessor fta0, int tupId0, IFrameTupleAccessor fta1, int tupId1);
+    public boolean evaluate(IFrameTupleAccessor fta0, int tupId0, IFrameTupleAccessor fta1, int tupId1);
 }
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/IPredicateEvaluatorFactory.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/IPredicateEvaluatorFactory.java
index bc2f339..41cd6f6 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/IPredicateEvaluatorFactory.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/IPredicateEvaluatorFactory.java
@@ -22,5 +22,5 @@
  */
 
 public interface IPredicateEvaluatorFactory extends Serializable {
-	public IPredicateEvaluator createPredicateEvaluator();
+    public IPredicateEvaluator createPredicateEvaluator();
 }
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/IPredicateEvaluatorFactoryProvider.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/IPredicateEvaluatorFactoryProvider.java
index 029ab211..09127b5 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/IPredicateEvaluatorFactoryProvider.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/dataflow/value/IPredicateEvaluatorFactoryProvider.java
@@ -21,6 +21,6 @@
  * Provides PredicateEvaluatorFactory based on (equi-join) keys 		
  */
 
-public interface IPredicateEvaluatorFactoryProvider extends Serializable{
-	public IPredicateEvaluatorFactory getPredicateEvaluatorFactory(int[] keys0, int[] keys1);
+public interface IPredicateEvaluatorFactoryProvider extends Serializable {
+    public IPredicateEvaluatorFactory getPredicateEvaluatorFactory(int[] keys0, int[] keys1);
 }
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/deployment/DeploymentId.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/deployment/DeploymentId.java
index 8d8d728..6eab7a7 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/deployment/DeploymentId.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/deployment/DeploymentId.java
@@ -21,7 +21,6 @@
  * The representation of a deployment id
  * 
  * @author yingyib
- *
  */
 public class DeploymentId implements Serializable {
     private static final long serialVersionUID = 1L;
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/io/FileReference.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/io/FileReference.java
index 4eda148..2270c6a 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/io/FileReference.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/io/FileReference.java
@@ -34,13 +34,13 @@
     }
 
     public File getFile() {
-    	return file;
+        return file;
     }
 
     public IODeviceHandle getDeviceHandle() {
-    	return dev;
+        return dev;
     }
-    
+
     @Override
     public String toString() {
         return file.getAbsolutePath();
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/job/IActivityClusterGraphGeneratorFactory.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/job/IActivityClusterGraphGeneratorFactory.java
index 978348b..99a0712 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/job/IActivityClusterGraphGeneratorFactory.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/job/IActivityClusterGraphGeneratorFactory.java
@@ -23,4 +23,6 @@
 public interface IActivityClusterGraphGeneratorFactory extends Serializable {
     public IActivityClusterGraphGenerator createActivityClusterGraphGenerator(JobId jobId,
             ICCApplicationContext ccAppCtx, EnumSet<JobFlag> jobFlags) throws HyracksException;
+
+    public JobSpecification getJobSpecification();
 }
\ No newline at end of file
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/job/IJobSerializerDeserializer.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/job/IJobSerializerDeserializer.java
index a105953..dcd8860 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/job/IJobSerializerDeserializer.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/job/IJobSerializerDeserializer.java
@@ -60,7 +60,6 @@
     public Class<?> loadClass(String className) throws HyracksException;
 
     /**
-     * 
      * @param binaryURLs
      * @throws HyracksException
      */
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/job/JobInfo.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/job/JobInfo.java
new file mode 100644
index 0000000..7d3230e
--- /dev/null
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/job/JobInfo.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.uci.ics.hyracks.api.job;
+
+import java.io.Serializable;
+import java.util.Map;
+
+import edu.uci.ics.hyracks.api.dataflow.OperatorDescriptorId;
+
+public class JobInfo implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private final JobId jobId;
+
+    private final JobStatus status;
+
+    private final Map<OperatorDescriptorId, Map<Integer, String>> operatorLocations;
+
+    public JobInfo(JobId jobId, JobStatus jobStatus, Map<OperatorDescriptorId, Map<Integer, String>> operatorLocations) {
+        this.jobId = jobId;
+        this.operatorLocations = operatorLocations;
+        this.status = jobStatus;
+    }
+
+    public JobId getJobId() {
+        return jobId;
+    }
+
+    public JobStatus getStatus() {
+        return status;
+    }
+
+    public Map<OperatorDescriptorId, Map<Integer, String>> getOperatorLocations() {
+        return operatorLocations;
+    }
+
+}
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/messages/IMessage.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/messages/IMessage.java
index e9ddd7e..7f9b5ea 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/messages/IMessage.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/messages/IMessage.java
@@ -18,7 +18,6 @@
 
 /**
  * @author rico
- * 
  */
 public interface IMessage extends Serializable {
 
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/messages/IMessageBroker.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/messages/IMessageBroker.java
index 7b33eb4..f4f770f 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/messages/IMessageBroker.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/messages/IMessageBroker.java
@@ -16,7 +16,6 @@
 
 /**
  * @author rico
- * 
  */
 public interface IMessageBroker {
 
diff --git a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/partitions/ResultSetPartitionId.java b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/partitions/ResultSetPartitionId.java
index 21f8d98..0c2e0cc 100644
--- a/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/partitions/ResultSetPartitionId.java
+++ b/hyracks/hyracks-api/src/main/java/edu/uci/ics/hyracks/api/partitions/ResultSetPartitionId.java
@@ -25,7 +25,7 @@
     private final JobId jobId;
 
     private final ResultSetId resultSetId;
-    
+
     private final int partition;
 
     public ResultSetPartitionId(JobId jobId, ResultSetId resultSetId, int partition) {
diff --git a/hyracks/hyracks-client/src/main/java/edu/uci/ics/hyracks/client/stats/Counters.java b/hyracks/hyracks-client/src/main/java/edu/uci/ics/hyracks/client/stats/Counters.java
index 7a75b2a..da30e20 100644
--- a/hyracks/hyracks-client/src/main/java/edu/uci/ics/hyracks/client/stats/Counters.java
+++ b/hyracks/hyracks-client/src/main/java/edu/uci/ics/hyracks/client/stats/Counters.java
@@ -16,6 +16,8 @@
 package edu.uci.ics.hyracks.client.stats;
 
 public class Counters {
+    public static final String NUM_PROCESSOR = "num-processors";
+
     public static final String SYSTEM_LOAD = "system-load-averages";
 
     public static final String MEMORY_USAGE = "heap-used-sizes";
diff --git a/hyracks/hyracks-client/src/main/java/edu/uci/ics/hyracks/client/stats/IClusterCounterContext.java b/hyracks/hyracks-client/src/main/java/edu/uci/ics/hyracks/client/stats/IClusterCounterContext.java
new file mode 100644
index 0000000..356ef34
--- /dev/null
+++ b/hyracks/hyracks-client/src/main/java/edu/uci/ics/hyracks/client/stats/IClusterCounterContext.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package edu.uci.ics.hyracks.client.stats;
+
+import edu.uci.ics.hyracks.api.job.profiling.counters.ICounter;
+import edu.uci.ics.hyracks.api.job.profiling.counters.ICounterContext;
+
+public interface IClusterCounterContext extends ICounterContext {
+
+    public ICounter getCounter(String machineName, String counterName, boolean create);
+    
+}
diff --git a/hyracks/hyracks-client/src/main/java/edu/uci/ics/hyracks/client/stats/impl/ClientCounterContext.java b/hyracks/hyracks-client/src/main/java/edu/uci/ics/hyracks/client/stats/impl/ClientCounterContext.java
index be35555..a527f87 100644
--- a/hyracks/hyracks-client/src/main/java/edu/uci/ics/hyracks/client/stats/impl/ClientCounterContext.java
+++ b/hyracks/hyracks-client/src/main/java/edu/uci/ics/hyracks/client/stats/impl/ClientCounterContext.java
@@ -29,17 +29,17 @@
 import org.json.simple.parser.JSONParser;
 
 import edu.uci.ics.hyracks.api.job.profiling.counters.ICounter;
-import edu.uci.ics.hyracks.api.job.profiling.counters.ICounterContext;
 import edu.uci.ics.hyracks.client.stats.AggregateCounter;
 import edu.uci.ics.hyracks.client.stats.Counters;
+import edu.uci.ics.hyracks.client.stats.IClusterCounterContext;
 import edu.uci.ics.hyracks.control.common.job.profiling.counters.Counter;
 
 /**
  * @author yingyib
  */
-public class ClientCounterContext implements ICounterContext {
+public class ClientCounterContext implements IClusterCounterContext {
     private static String[] RESET_COUNTERS = { Counters.NETWORK_IO_READ, Counters.NETWORK_IO_WRITE,
-            Counters.MEMORY_USAGE, Counters.DISK_READ, Counters.DISK_WRITE };
+            Counters.MEMORY_USAGE, Counters.DISK_READ, Counters.DISK_WRITE, Counters.NUM_PROCESSOR };
     private static String[] AGG_COUNTERS = { Counters.SYSTEM_LOAD };
     private static int UPDATE_INTERVAL = 10000;
 
@@ -58,6 +58,14 @@
         for (String aggCounterName : AGG_COUNTERS) {
             counterMap.put(aggCounterName, new AggregateCounter(aggCounterName));
         }
+        for (String slave : slaveMachines) {
+            for (String restCounterName : RESET_COUNTERS) {
+                counterMap.put(slave + "$" + restCounterName, new AggregateCounter(restCounterName));
+            }
+            for (String aggCounterName : AGG_COUNTERS) {
+                counterMap.put(slave + "$" + aggCounterName, new AggregateCounter(aggCounterName));
+            }
+        }
         requestCounters();
         updateThread.start();
     }
@@ -70,6 +78,12 @@
             AggregateCounter aggCounter = (AggregateCounter) counterMap.get(aggCounterName);
             aggCounter.reset();
         }
+        for (String slave : slaveMachines) {
+            for (String aggCounterName : RESET_COUNTERS) {
+                AggregateCounter aggCounter = (AggregateCounter) counterMap.get(slave + "$" + aggCounterName);
+                aggCounter.reset();
+            }
+        }
     }
 
     public void resetAll() {
@@ -78,6 +92,12 @@
             AggregateCounter aggCounter = (AggregateCounter) counterMap.get(aggCounterName);
             aggCounter.reset();
         }
+        for (String slave : slaveMachines) {
+            for (String aggCounterName : AGG_COUNTERS) {
+                AggregateCounter aggCounter = (AggregateCounter) counterMap.get(slave + "$" + aggCounterName);
+                aggCounter.reset();
+            }
+        }
     }
 
     @Override
@@ -89,6 +109,16 @@
         return counter;
     }
 
+    @Override
+    public ICounter getCounter(String machineName, String counterName, boolean create) {
+        Counter counter = counterMap.get(machineName + "$" + counterName);
+        if (counter == null) {
+            throw new IllegalStateException("request an unknown counter: " + counterName + " on slave machine "
+                    + machineName + "!");
+        }
+        return counter;
+    }
+
     /**
      * request to each slave machine for all the counters
      */
@@ -99,7 +129,10 @@
                 String slaveProfile = requestProfile(slave);
                 JSONParser parser = new JSONParser();
                 JSONObject jo = (JSONObject) parser.parse(slaveProfile);
-                updateCounterMap((JSONObject) jo.get("result"));
+                Object counterObject = jo.get("result");
+                if (counterObject instanceof JSONObject) {
+                    updateCounterMapWithJSONArray(slave, (JSONObject) counterObject);
+                }
             }
         } catch (Exception e) {
             throw new IllegalStateException(e);
@@ -112,14 +145,32 @@
      * @param jo
      *            the Profile JSON object
      */
-    private void updateCounterMap(JSONObject jo) {
+    private void updateCounterMapWithJSONArray(String slave, JSONObject jo) {
         for (String counterName : RESET_COUNTERS) {
-            JSONArray jArray = (JSONArray) jo.get(counterName);
-            if(jArray==null){
-                continue; 
-            }
+            updateCounter(slave, jo, counterName);
+        }
+        
+        for (String counterName : AGG_COUNTERS) {
+            updateCounter(slave, jo, counterName);
+        }
+    }
+
+    private void updateCounter(String slave, JSONObject jo, String counterName) {
+        Object counterObject = jo.get(counterName);
+        long counterValue = extractCounterValue(counterObject);
+        // global counter
+        ICounter counter = getCounter(counterName, true);
+        counter.set(counterValue);
+        //local counters
+        ICounter localCounter = getCounter(slave, counterName, true);
+        localCounter.set(counterValue);
+    }
+
+    private long extractCounterValue(Object counterObject) {
+        long counterValue = 0;
+        if (counterObject instanceof JSONArray) {
+            JSONArray jArray = (JSONArray) counterObject;
             Object[] values = jArray.toArray();
-            long counterValue = 0;
             for (Object value : values) {
                 if (value instanceof Double) {
                     Double dValue = (Double) value;
@@ -130,9 +181,11 @@
                 }
             }
             counterValue /= values.length;
-            ICounter counter = getCounter(counterName, true);
-            counter.set(counterValue);
+        } else {
+            Long val = (Long) counterObject;
+            counterValue = val.longValue();
         }
+        return counterValue;
     }
 
     /**
@@ -203,4 +256,4 @@
 
     }
 
-}
+}
\ No newline at end of file
diff --git a/hyracks/hyracks-client/src/test/java/edu/uci/ics/hyracks/client/stats/ClientCounterContextTest.java b/hyracks/hyracks-client/src/test/java/edu/uci/ics/hyracks/client/stats/ClientCounterContextTest.java
index 8394daf..bbf212f 100644
--- a/hyracks/hyracks-client/src/test/java/edu/uci/ics/hyracks/client/stats/ClientCounterContextTest.java
+++ b/hyracks/hyracks-client/src/test/java/edu/uci/ics/hyracks/client/stats/ClientCounterContextTest.java
@@ -34,10 +34,16 @@
             wait(20000);
         }
         String[] counters = { Counters.MEMORY_USAGE, Counters.NETWORK_IO_READ, Counters.NETWORK_IO_WRITE,
-                Counters.SYSTEM_LOAD, Counters.DISK_READ, Counters.DISK_WRITE };
+                Counters.SYSTEM_LOAD, Counters.NUM_PROCESSOR, Counters.DISK_READ, Counters.DISK_WRITE };
         for (String counterName : counters) {
             ICounter counter = ccContext.getCounter(counterName, false);
-            System.out.println(counter.get());
+            System.out.println(counterName + ": " + counter.get());
+        }
+        for (String slave : ncs) {
+            for (String counterName : counters) {
+                ICounter counter = ccContext.getCounter(slave, counterName, false);
+                System.out.println(slave + " " + counterName + ": " + counter.get());
+            }
         }
         ccContext.stop();
         HyracksUtils.deinit();
diff --git a/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/ClusterControllerService.java b/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/ClusterControllerService.java
index 0463350..5ec9c19 100644
--- a/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/ClusterControllerService.java
+++ b/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/ClusterControllerService.java
@@ -42,6 +42,7 @@
 import edu.uci.ics.hyracks.api.dataset.DatasetJobRecord.Status;
 import edu.uci.ics.hyracks.api.deployment.DeploymentId;
 import edu.uci.ics.hyracks.api.job.JobId;
+import edu.uci.ics.hyracks.api.job.JobInfo;
 import edu.uci.ics.hyracks.api.job.JobStatus;
 import edu.uci.ics.hyracks.api.topology.ClusterTopology;
 import edu.uci.ics.hyracks.api.topology.TopologyDefinitionParser;
@@ -55,6 +56,7 @@
 import edu.uci.ics.hyracks.control.cc.work.CliUnDeployBinaryWork;
 import edu.uci.ics.hyracks.control.cc.work.GetDatasetDirectoryServiceInfoWork;
 import edu.uci.ics.hyracks.control.cc.work.GetIpAddressNodeNameMapWork;
+import edu.uci.ics.hyracks.control.cc.work.GetJobInfoWork;
 import edu.uci.ics.hyracks.control.cc.work.GetJobStatusWork;
 import edu.uci.ics.hyracks.control.cc.work.GetNodeControllersInfoWork;
 import edu.uci.ics.hyracks.control.cc.work.GetResultPartitionLocationsWork;
@@ -350,6 +352,13 @@
                     return;
                 }
 
+                case GET_JOB_INFO: {
+                    HyracksClientInterfaceFunctions.GetJobInfoFunction gjsf = (HyracksClientInterfaceFunctions.GetJobInfoFunction) fn;
+                    workQueue.schedule(new GetJobInfoWork(ClusterControllerService.this, gjsf.getJobId(),
+                            new IPCResponder<JobInfo>(handle, mid)));
+                    return;
+                }
+
                 case START_JOB: {
                     HyracksClientInterfaceFunctions.StartJobFunction sjf = (HyracksClientInterfaceFunctions.StartJobFunction) fn;
                     JobId jobId = createJobId();
diff --git a/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/adminconsole/pages/AbstractPage.java b/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/adminconsole/pages/AbstractPage.java
index ecae1cd..dd4642d 100644
--- a/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/adminconsole/pages/AbstractPage.java
+++ b/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/adminconsole/pages/AbstractPage.java
@@ -20,7 +20,7 @@
 
 public class AbstractPage extends WebPage {
     private static final long serialVersionUID = 1L;
-    
+
     public HyracksAdminConsoleApplication getAdminConsoleApplication() {
         return (HyracksAdminConsoleApplication) getApplication();
     }
diff --git a/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/job/JobRun.java b/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/job/JobRun.java
index bae0eb5..8e47735 100644
--- a/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/job/JobRun.java
+++ b/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/job/JobRun.java
@@ -14,6 +14,7 @@
  */
 package edu.uci.ics.hyracks.control.cc.job;
 
+import java.util.ArrayList;
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.util.EnumSet;
@@ -29,6 +30,7 @@
 
 import edu.uci.ics.hyracks.api.dataflow.ActivityId;
 import edu.uci.ics.hyracks.api.dataflow.ConnectorDescriptorId;
+import edu.uci.ics.hyracks.api.dataflow.OperatorDescriptorId;
 import edu.uci.ics.hyracks.api.dataflow.TaskId;
 import edu.uci.ics.hyracks.api.dataflow.connectors.IConnectorPolicy;
 import edu.uci.ics.hyracks.api.deployment.DeploymentId;
@@ -87,6 +89,8 @@
 
     private List<Exception> pendingExceptions;
 
+    private Map<OperatorDescriptorId, Map<Integer, String>> operatorLocations;
+
     public JobRun(ClusterControllerService ccs, DeploymentId deploymentId, JobId jobId,
             IActivityClusterGraphGenerator acgg, EnumSet<JobFlag> jobFlags) {
         this.deploymentId = deploymentId;
@@ -101,6 +105,7 @@
         cleanupPendingNodeIds = new HashSet<String>();
         profile = new JobProfile(jobId);
         connectorPolicyMap = new HashMap<ConnectorDescriptorId, IConnectorPolicy>();
+        operatorLocations = new HashMap<OperatorDescriptorId, Map<Integer, String>>();
     }
 
     public DeploymentId getDeploymentId() {
@@ -178,6 +183,15 @@
         this.endTime = endTime;
     }
 
+    public void registerOperatorLocation(OperatorDescriptorId op, int partition, String location) {
+        Map<Integer, String> locations = operatorLocations.get(op);
+        if (locations == null) {
+            locations = new HashMap<Integer, String>();
+            operatorLocations.put(op, locations);
+        }
+        locations.put(partition, location);
+    }
+
     @Override
     public synchronized void waitForCompletion() throws Exception {
         while (status != JobStatus.TERMINATED && status != JobStatus.FAILURE) {
@@ -348,8 +362,7 @@
                                 taskAttempt.put("end-time", ta.getEndTime());
                                 List<Exception> exceptions = ta.getExceptions();
                                 if (exceptions != null && !exceptions.isEmpty()) {
-                                    List<Exception> filteredExceptions = ExceptionUtils
-                                            .getActualExceptions(exceptions);
+                                    List<Exception> filteredExceptions = ExceptionUtils.getActualExceptions(exceptions);
                                     for (Exception exception : filteredExceptions) {
                                         StringWriter exceptionWriter = new StringWriter();
                                         exception.printStackTrace(new PrintWriter(exceptionWriter));
@@ -379,4 +392,8 @@
 
         return result;
     }
-}
\ No newline at end of file
+
+    public Map<OperatorDescriptorId, Map<Integer, String>> getOperatorLocations() {
+        return operatorLocations;
+    }
+}
diff --git a/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/scheduler/JobScheduler.java b/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/scheduler/JobScheduler.java
index fd6360a..ad4744b 100644
--- a/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/scheduler/JobScheduler.java
+++ b/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/scheduler/JobScheduler.java
@@ -35,6 +35,7 @@
 import edu.uci.ics.hyracks.api.constraints.expressions.PartitionLocationExpression;
 import edu.uci.ics.hyracks.api.dataflow.ActivityId;
 import edu.uci.ics.hyracks.api.dataflow.ConnectorDescriptorId;
+import edu.uci.ics.hyracks.api.dataflow.OperatorDescriptorId;
 import edu.uci.ics.hyracks.api.dataflow.TaskAttemptId;
 import edu.uci.ics.hyracks.api.dataflow.TaskId;
 import edu.uci.ics.hyracks.api.dataflow.connectors.IConnectorPolicy;
@@ -92,6 +93,7 @@
 
     public void startJob() throws HyracksException {
         startRunnableActivityClusters();
+        ccs.getApplicationContext().notifyJobStart(jobRun.getJobId());
     }
 
     private void findRunnableTaskClusterRoots(Set<TaskCluster> frontier, Collection<ActivityCluster> roots)
@@ -326,6 +328,8 @@
                 tads = new ArrayList<TaskAttemptDescriptor>();
                 taskAttemptMap.put(nodeId, tads);
             }
+            OperatorDescriptorId opId = tid.getActivityId().getOperatorDescriptorId();
+            jobRun.registerOperatorLocation(opId, tid.getPartition(), nodeId);
             ActivityPartitionDetails apd = ts.getActivityPlan().getActivityPartitionDetails();
             TaskAttemptDescriptor tad = new TaskAttemptDescriptor(taskAttempt.getTaskAttemptId(),
                     apd.getPartitionCount(), apd.getInputPartitionCounts(), apd.getOutputPartitionCounts());
diff --git a/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/work/CliUnDeployBinaryWork.java b/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/work/CliUnDeployBinaryWork.java
index d54444c..ef82862 100644
--- a/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/work/CliUnDeployBinaryWork.java
+++ b/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/work/CliUnDeployBinaryWork.java
@@ -71,7 +71,7 @@
                 nodeIds.add(nc);
             }
             final DeploymentRun dRun = new DeploymentRun(nodeIds);
-            
+
             /** The following call prevents a user to undeploy with the same deployment id simultaneously. */
             ccs.addDeploymentRun(deploymentId, dRun);
 
diff --git a/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/work/GetJobInfoWork.java b/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/work/GetJobInfoWork.java
new file mode 100644
index 0000000..72f8811
--- /dev/null
+++ b/hyracks/hyracks-control/hyracks-control-cc/src/main/java/edu/uci/ics/hyracks/control/cc/work/GetJobInfoWork.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.uci.ics.hyracks.control.cc.work;
+
+import edu.uci.ics.hyracks.api.job.JobId;
+import edu.uci.ics.hyracks.api.job.JobInfo;
+import edu.uci.ics.hyracks.control.cc.ClusterControllerService;
+import edu.uci.ics.hyracks.control.cc.job.JobRun;
+import edu.uci.ics.hyracks.control.common.work.IResultCallback;
+import edu.uci.ics.hyracks.control.common.work.SynchronizableWork;
+
+public class GetJobInfoWork extends SynchronizableWork {
+    private final ClusterControllerService ccs;
+    private final JobId jobId;
+    private final IResultCallback<JobInfo> callback;
+
+    public GetJobInfoWork(ClusterControllerService ccs, JobId jobId, IResultCallback<JobInfo> callback) {
+        this.ccs = ccs;
+        this.jobId = jobId;
+        this.callback = callback;
+    }
+
+    @Override
+    protected void doRun() throws Exception {
+        try {
+            JobRun run = ccs.getActiveRunMap().get(jobId);
+            if (run == null) {
+                run = ccs.getRunMapArchive().get(jobId);
+            }
+            JobInfo info = (run != null) ? new JobInfo(run.getJobId(), run.getStatus(), run.getOperatorLocations())
+                    : null;
+            callback.setValue(info);
+        } catch (Exception e) {
+            callback.setException(e);
+        }
+    }
+}
\ No newline at end of file
diff --git a/hyracks/hyracks-control/hyracks-control-common/src/main/java/edu/uci/ics/hyracks/control/common/controllers/CCConfig.java b/hyracks/hyracks-control/hyracks-control-common/src/main/java/edu/uci/ics/hyracks/control/common/controllers/CCConfig.java
index c7244fb..f6c1e54 100644
--- a/hyracks/hyracks-control/hyracks-control-common/src/main/java/edu/uci/ics/hyracks/control/common/controllers/CCConfig.java
+++ b/hyracks/hyracks-control/hyracks-control-common/src/main/java/edu/uci/ics/hyracks/control/common/controllers/CCConfig.java
@@ -113,4 +113,4 @@
             }
         }
     }
-}
\ No newline at end of file
+}
diff --git a/hyracks/hyracks-control/hyracks-control-common/src/main/java/edu/uci/ics/hyracks/control/common/controllers/NCConfig.java b/hyracks/hyracks-control/hyracks-control-common/src/main/java/edu/uci/ics/hyracks/control/common/controllers/NCConfig.java
index 43129f5..01525e4 100644
--- a/hyracks/hyracks-control/hyracks-control-common/src/main/java/edu/uci/ics/hyracks/control/common/controllers/NCConfig.java
+++ b/hyracks/hyracks-control/hyracks-control-common/src/main/java/edu/uci/ics/hyracks/control/common/controllers/NCConfig.java
@@ -123,4 +123,4 @@
         }
 
     }
-}
\ No newline at end of file
+}
diff --git a/pregelix/pregelix-core/src/main/java/edu/uci/ics/pregelix/core/driver/Driver.java b/pregelix/pregelix-core/src/main/java/edu/uci/ics/pregelix/core/driver/Driver.java
index 4604bc7..b3a90e9 100644
--- a/pregelix/pregelix-core/src/main/java/edu/uci/ics/pregelix/core/driver/Driver.java
+++ b/pregelix/pregelix-core/src/main/java/edu/uci/ics/pregelix/core/driver/Driver.java
@@ -59,8 +59,8 @@
 
 @SuppressWarnings("rawtypes")
 public class Driver implements IDriver {
-    public static final String[] COUNTERS = { Counters.SYSTEM_LOAD, Counters.MEMORY_USAGE, Counters.DISK_READ,
-            Counters.DISK_WRITE, Counters.NETWORK_IO_READ, Counters.NETWORK_IO_WRITE };
+    public static final String[] COUNTERS = { Counters.NUM_PROCESSOR, Counters.SYSTEM_LOAD, Counters.MEMORY_USAGE,
+            Counters.DISK_READ, Counters.DISK_WRITE, Counters.NETWORK_IO_READ, Counters.NETWORK_IO_WRITE };
     private static final Log LOG = LogFactory.getLog(Driver.class);
     private IHyracksClientConnection hcc;
     private Class exampleClass;