diff --git a/cli/src/main/java/org/apache/falcon/cli/FalconEntityCLI.java b/cli/src/main/java/org/apache/falcon/cli/FalconEntityCLI.java
index a8aea52..dcac8e8 100644
--- a/cli/src/main/java/org/apache/falcon/cli/FalconEntityCLI.java
+++ b/cli/src/main/java/org/apache/falcon/cli/FalconEntityCLI.java
@@ -369,7 +369,7 @@
         OUT.get().println(result);
     }
 
-    private void validateColo(Set<String> optionsList) {
+    static void validateColo(Set<String> optionsList) {
         if (optionsList.contains(FalconCLIConstants.COLO_OPT)) {
             throw new FalconCLIException("Invalid argument : " + FalconCLIConstants.COLO_OPT);
         }
diff --git a/cli/src/main/java/org/apache/falcon/cli/FalconExtensionCLI.java b/cli/src/main/java/org/apache/falcon/cli/FalconExtensionCLI.java
index 0343aa8..2a105dc 100644
--- a/cli/src/main/java/org/apache/falcon/cli/FalconExtensionCLI.java
+++ b/cli/src/main/java/org/apache/falcon/cli/FalconExtensionCLI.java
@@ -33,38 +33,44 @@
 import org.apache.falcon.resource.ExtensionInstanceList;
 import org.apache.falcon.resource.ExtensionJobList;
 
+import java.io.IOException;
 import java.io.PrintStream;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static org.apache.falcon.cli.FalconEntityCLI.validateColo;
+import static org.apache.falcon.client.FalconCLIConstants.COLO_OPT;
+import static org.apache.falcon.client.FalconCLIConstants.COLO_OPT_DESCRIPTION;
+
 /**
  * Falcon extensions Command Line Interface - wraps the RESTful API for extensions.
  */
-public class FalconExtensionCLI {
+public class FalconExtensionCLI extends FalconCLI{
     public static final AtomicReference<PrintStream> OUT = new AtomicReference<>(System.out);
 
     // Extension commands
-    public static final String ENUMERATE_OPT = "enumerate";
-    public static final String DEFINITION_OPT = "definition";
-    public static final String DESCRIBE_OPT = "describe";
-    public static final String INSTANCES_OPT = "instances";
-    public static final String UNREGISTER_OPT = "unregister";
-    public static final String DETAIL_OPT = "detail";
-    public static final String REGISTER_OPT = "register";
-    public static final String ENABLE_OPT = "enable";
-    public static final String DISABLE_OPT = "disable";
+    private static final String ENUMERATE_OPT = "enumerate";
+    private static final String DEFINITION_OPT = "definition";
+    private static final String DESCRIBE_OPT = "describe";
+    private static final String INSTANCES_OPT = "instances";
+    private static final String UNREGISTER_OPT = "unregister";
+    private static final String DETAIL_OPT = "detail";
+    private static final String REGISTER_OPT = "register";
+    private static final String ENABLE_OPT = "enable";
+    private static final String DISABLE_OPT = "disable";
 
     // Input parameters
-    public static final String EXTENSION_NAME_OPT = "extensionName";
-    public static final String JOB_NAME_OPT = "jobName";
+    private static final String EXTENSION_NAME_OPT = "extensionName";
+    private static final String JOB_NAME_OPT = "jobName";
     public static final String DESCRIPTION = "description";
-    public static final String PATH = "path";
+    private static final String PATH = "path";
 
-    public FalconExtensionCLI() {
+    FalconExtensionCLI() throws Exception {
+        super();
     }
 
-    public void extensionCommand(CommandLine commandLine, FalconClient client) {
+    void extensionCommand(CommandLine commandLine, FalconClient client) throws IOException {
         Set<String> optionsList = new HashSet<>();
         for (Option option : commandLine.getOptions()) {
             optionsList.add(option.getOpt());
@@ -77,6 +83,8 @@
         String doAsUser = commandLine.getOptionValue(FalconCLIConstants.DO_AS_OPT);
         String path = commandLine.getOptionValue(FalconCLIConstants.PATH);
         String description = commandLine.getOptionValue(FalconCLIConstants.DESCRIPTION);
+        String colo = commandLine.getOptionValue(FalconCLIConstants.COLO_OPT);
+        colo = getColo(colo);
 
         if (optionsList.contains(ENUMERATE_OPT)) {
             result = client.enumerateExtensions().getMessage();
@@ -105,6 +113,7 @@
             validateRequiredParameter(extensionName, EXTENSION_NAME_OPT);
             validateRequiredParameter(jobName, JOB_NAME_OPT);
             validateRequiredParameter(filePath, FalconCLIConstants.FILE_PATH_OPT);
+            validateColo(optionsList);
             result = client.submitExtensionJob(extensionName, jobName, filePath, doAsUser).getMessage();
         } else if (optionsList.contains(REGISTER_OPT)) {
             validateRequiredParameter(extensionName, EXTENSION_NAME_OPT);
@@ -114,6 +123,7 @@
             validateRequiredParameter(extensionName, EXTENSION_NAME_OPT);
             validateRequiredParameter(jobName, JOB_NAME_OPT);
             validateRequiredParameter(filePath, FalconCLIConstants.FILE_PATH_OPT);
+            validateColo(optionsList);
             result = client.submitAndScheduleExtensionJob(extensionName, jobName, filePath, doAsUser).getMessage();
         } else if (optionsList.contains(FalconCLIConstants.UPDATE_OPT)) {
             validateRequiredParameter(jobName, JOB_NAME_OPT);
@@ -125,7 +135,8 @@
             result = client.validateExtensionJob(extensionName, jobName, filePath, doAsUser).getMessage();
         } else if (optionsList.contains(FalconCLIConstants.SCHEDULE_OPT)) {
             validateRequiredParameter(jobName, JOB_NAME_OPT);
-            result = client.scheduleExtensionJob(jobName, doAsUser).getMessage();
+            colo = getColo(colo);
+            result = client.scheduleExtensionJob(jobName, colo, doAsUser).getMessage();
         } else if (optionsList.contains(FalconCLIConstants.SUSPEND_OPT)) {
             validateRequiredParameter(jobName, JOB_NAME_OPT);
             result = client.suspendExtensionJob(jobName, doAsUser).getMessage();
@@ -170,7 +181,7 @@
         OUT.get().println(result);
     }
 
-    public Options createExtensionOptions() {
+    Options createExtensionOptions() {
         Options extensionOptions = new Options();
 
         Option enumerate = new Option(ENUMERATE_OPT, false, "Enumerate all extensions");
@@ -192,6 +203,8 @@
         Option detail = new Option(FalconCLIConstants.DETAIL, false, "Show details of a given extension");
         Option register = new Option(FalconCLIConstants.REGISTER, false, "Register an extension with Falcon. This will "
                 + "make the extension available for instantiation for all users.");
+        Option colo = new Option(COLO_OPT, true, COLO_OPT_DESCRIPTION);
+        colo.setRequired(false);
 
         OptionGroup group = new OptionGroup();
         group.addOption(enumerate);
@@ -249,6 +262,7 @@
         extensionOptions.addOption(filePath);
         extensionOptions.addOption(path);
         extensionOptions.addOption(description);
+        extensionOptions.addOption(colo);
 
         return extensionOptions;
     }
diff --git a/client/src/main/java/org/apache/falcon/client/AbstractFalconClient.java b/client/src/main/java/org/apache/falcon/client/AbstractFalconClient.java
index 3181b64..7b8a606 100644
--- a/client/src/main/java/org/apache/falcon/client/AbstractFalconClient.java
+++ b/client/src/main/java/org/apache/falcon/client/AbstractFalconClient.java
@@ -222,6 +222,13 @@
                                                 String doAsUser);
 
     /**
+     * Schedules the set of entities that are part of the extension.
+     * @param jobName extensionJob that needs to be scheduled.
+     * @return APIResult stating status of scheduling the extension.
+     */
+    public abstract APIResult scheduleExtensionJob(String jobName, String coloExpr, String doAsUser);
+
+    /**
      * Prepares set of entities the extension has implemented and stage them to a local directory and submits and
      * schedules them.
      * @param extensionName extension which is available in the store.
diff --git a/client/src/main/java/org/apache/falcon/client/FalconClient.java b/client/src/main/java/org/apache/falcon/client/FalconClient.java
index 8401c9c..2772085 100644
--- a/client/src/main/java/org/apache/falcon/client/FalconClient.java
+++ b/client/src/main/java/org/apache/falcon/client/FalconClient.java
@@ -1200,9 +1200,10 @@
         }
     }
 
-    public APIResult scheduleExtensionJob(final String jobName, final String doAsUser)  {
+    public APIResult scheduleExtensionJob(String jobName, final String coloExpr, final String doAsUser)  {
         ClientResponse clientResponse = new ResourceBuilder()
                 .path(ExtensionOperations.SCHEDULE.path, jobName)
+                .addQueryParam(COLO, coloExpr)
                 .addQueryParam(DO_AS_OPT, doAsUser)
                 .call(ExtensionOperations.SCHEDULE);
         return getResponse(APIResult.class, clientResponse);
diff --git a/extensions/src/main/java/org/apache/falcon/extensions/jdbc/ExtensionMetaStore.java b/extensions/src/main/java/org/apache/falcon/extensions/jdbc/ExtensionMetaStore.java
index e53069a..9126b67 100644
--- a/extensions/src/main/java/org/apache/falcon/extensions/jdbc/ExtensionMetaStore.java
+++ b/extensions/src/main/java/org/apache/falcon/extensions/jdbc/ExtensionMetaStore.java
@@ -19,6 +19,7 @@
 
 import org.apache.falcon.extensions.ExtensionStatus;
 import org.apache.falcon.extensions.ExtensionType;
+import org.apache.falcon.extensions.store.ExtensionStore;
 import org.apache.falcon.persistence.ExtensionBean;
 import org.apache.falcon.persistence.ExtensionJobsBean;
 import org.apache.falcon.persistence.PersistenceConstants;
@@ -145,6 +146,11 @@
 
     public void storeExtensionJob(String jobName, String extensionName, List<String> feeds, List<String> processes,
                                   byte[] config) {
+        ExtensionMetaStore metaStore = ExtensionStore.getMetaStore();
+        boolean alreadySubmitted = false;
+        if (metaStore.getExtensionJobDetails(jobName) != null){
+            alreadySubmitted = true;
+        }
         ExtensionJobsBean extensionJobsBean = new ExtensionJobsBean();
         Date currentTime = new Date(System.currentTimeMillis());
         extensionJobsBean.setJobName(jobName);
@@ -157,7 +163,11 @@
         EntityManager entityManager = getEntityManager();
         try {
             beginTransaction(entityManager);
-            entityManager.persist(extensionJobsBean);
+            if (alreadySubmitted) {
+                entityManager.merge(extensionJobsBean);
+            } else {
+                entityManager.persist(extensionJobsBean);
+            }
         } finally {
             commitAndCloseTransaction(entityManager);
         }
diff --git a/extensions/src/test/java/org/apache/falcon/extensions/jdbc/ExtensionMetaStoreTest.java b/extensions/src/test/java/org/apache/falcon/extensions/jdbc/ExtensionMetaStoreTest.java
index 1688abb..e3327e8 100644
--- a/extensions/src/test/java/org/apache/falcon/extensions/jdbc/ExtensionMetaStoreTest.java
+++ b/extensions/src/test/java/org/apache/falcon/extensions/jdbc/ExtensionMetaStoreTest.java
@@ -83,6 +83,8 @@
 
         byte[] config = new byte[0];
         stateStore.storeExtensionJob("job1", "test2", feeds, processes, config);
+        //storing again to check for entity manager merge to let submission go forward.
+        stateStore.storeExtensionJob("job1", "test2", feeds, processes, config);
 
         Assert.assertEquals(stateStore.getAllExtensionJobs().size(), 1);
         Assert.assertEquals(stateStore.getExtensionJobDetails("job1").getFeeds().get(0), "testFeed");
diff --git a/prism/src/main/java/org/apache/falcon/resource/AbstractEntityManager.java b/prism/src/main/java/org/apache/falcon/resource/AbstractEntityManager.java
index 81b0448..8daf8c7 100644
--- a/prism/src/main/java/org/apache/falcon/resource/AbstractEntityManager.java
+++ b/prism/src/main/java/org/apache/falcon/resource/AbstractEntityManager.java
@@ -128,12 +128,21 @@
     }
 
     protected Set<String> getColosFromExpression(String coloExpr, String type, String entity) {
-        Set<String> colos;
         final Set<String> applicableColos = getApplicableColos(type, entity);
+        return getColosFromExpression(coloExpr, applicableColos);
+    }
+
+    protected Set<String> getColosFromExpression(String coloExpr, String type, Entity entity) {
+        final Set<String> applicableColos = getApplicableColos(type, entity);
+        return getColosFromExpression(coloExpr, applicableColos);
+    }
+
+    private Set<String> getColosFromExpression(String coloExpr, Set<String> applicableColos) {
+        Set<String> colos;
         if (coloExpr == null || coloExpr.equals("*") || coloExpr.isEmpty()) {
             colos = applicableColos;
         } else {
-            colos = new HashSet<String>(Arrays.asList(coloExpr.split(",")));
+            colos = new HashSet<>(Arrays.asList(coloExpr.split(",")));
             if (!applicableColos.containsAll(colos)) {
                 throw FalconWebException.newAPIException("Given colos not applicable for entity operation");
             }
diff --git a/prism/src/main/java/org/apache/falcon/resource/proxy/ExtensionManagerProxy.java b/prism/src/main/java/org/apache/falcon/resource/proxy/ExtensionManagerProxy.java
index 2b5cbe7..6f75dc7 100644
--- a/prism/src/main/java/org/apache/falcon/resource/proxy/ExtensionManagerProxy.java
+++ b/prism/src/main/java/org/apache/falcon/resource/proxy/ExtensionManagerProxy.java
@@ -191,21 +191,26 @@
     @Consumes({MediaType.TEXT_XML, MediaType.TEXT_PLAIN})
     @Produces({MediaType.TEXT_XML, MediaType.TEXT_PLAIN, MediaType.APPLICATION_JSON})
     public APIResult schedule(@PathParam("job-name") String jobName,
+                              @Context HttpServletRequest request,
+                              @QueryParam("colo") final String coloExpr,
                               @DefaultValue("") @QueryParam("doAs") String doAsUser) {
         checkIfExtensionServiceIsEnabled();
         checkIfExtensionIsEnabled(ExtensionStore.getMetaStore().getExtensionJobDetails(jobName).getExtensionName());
-        try {
-            List<Entity> entities = getEntityList("", "", "", TAG_PREFIX_EXTENSION_JOB + jobName, "", doAsUser);
-            if (entities.isEmpty()) {
-                // return failure if the extension job doesn't exist
-                return new APIResult(APIResult.Status.FAILED, "Extension job " + jobName + " doesn't exist.");
-            }
+        ExtensionMetaStore metaStore = ExtensionStore.getMetaStore();
+        ExtensionJobsBean extensionJobsBean = metaStore.getExtensionJobDetails(jobName);
+        if (extensionJobsBean == null) {
+            // return failure if the extension job doesn't exist
+            LOG.error("Extension Job not found:" + jobName);
+            throw FalconWebException.newAPIException("ExtensionJob not found:" + jobName,
+                    Response.Status.NOT_FOUND);
+        }
 
-            for (Entity entity : entities) {
-                scheduleInternal(entity.getEntityType().name(), entity.getName(), null, null);
-            }
-        } catch (FalconException | IOException e) {
-            LOG.error("Error when scheduling extension job: " + jobName + ": ", e);
+        SortedMap<EntityType, List<Entity>> entityMap;
+        try {
+            entityMap = getJobEntities(extensionJobsBean);
+            scheduleEntities(entityMap, request, coloExpr);
+        } catch (FalconException | IOException | JAXBException e) {
+            LOG.error("Error while scheduling entities of the extension: " + jobName + ": ", e);
             throw FalconWebException.newAPIException(e, Response.Status.INTERNAL_SERVER_ERROR);
         }
         return new APIResult(APIResult.Status.SUCCEEDED, "Extension job " + jobName + " scheduled successfully");
@@ -375,6 +380,7 @@
             @Context HttpServletRequest request,
             @DefaultValue("") @QueryParam("doAs") String doAsUser,
             @QueryParam("jobName") String jobName,
+            @QueryParam("colo") final String coloExpr,
             @FormDataParam("processes") List<FormDataBodyPart> processForms,
             @FormDataParam("feeds") List<FormDataBodyPart> feedForms,
             @FormDataParam("config") InputStream config) {
@@ -385,7 +391,7 @@
         try {
             entityMap = getEntityList(extensionName, jobName, feedForms, processForms, config);
             submitEntities(extensionName, jobName, entityMap, config, request);
-            scheduleEntities(entityMap, request);
+            scheduleEntities(entityMap, request, coloExpr);
         } catch (FalconException | IOException | JAXBException e) {
             LOG.error("Error while submitting extension job: ", e);
             throw FalconWebException.newAPIException(e, Response.Status.INTERNAL_SERVER_ERROR);
@@ -393,13 +399,13 @@
         return new APIResult(APIResult.Status.SUCCEEDED, "Extension job submitted and scheduled successfully");
     }
 
-    protected void scheduleEntities(Map<EntityType, List<Entity>> entityMap, HttpServletRequest request)
+    private void scheduleEntities(Map<EntityType, List<Entity>> entityMap, HttpServletRequest request, String coloExpr)
         throws FalconException, JAXBException, IOException {
         for (Map.Entry<EntityType, List<Entity>> entry : entityMap.entrySet()) {
             for (final Entity entity : entry.getValue()) {
                 final HttpServletRequest httpServletRequest = getEntityStream(entity, entity.getEntityType(), request);
                 final HttpServletRequest bufferedRequest = getBufferedRequest(httpServletRequest);
-                final Set<String> colos = getApplicableColos(entity.getEntityType().toString(), entity);
+                final Set<String> colos = getColosFromExpression(coloExpr, entity.getEntityType().name(), entity);
 
                 new EntityProxy(entity.getEntityType().toString(), entity.getName()) {
                     @Override
@@ -441,8 +447,9 @@
     }
 
     private void submitEntities(String extensionName, String jobName,
-                                  SortedMap<EntityType, List<Entity>> entityMap, InputStream configStream,
-                                  HttpServletRequest request) throws FalconException, IOException, JAXBException {
+                                SortedMap<EntityType, List<Entity>> entityMap, InputStream configStream,
+                                HttpServletRequest request)
+        throws FalconException, IOException, JAXBException {
         List<Entity> feeds = entityMap.get(EntityType.FEED);
         List<Entity> processes = entityMap.get(EntityType.PROCESS);
         validateFeeds(feeds);
@@ -766,6 +773,7 @@
 
     private static void checkIfExtensionServiceIsEnabled() {
         if (!Services.get().isRegistered(ExtensionService.SERVICE_NAME)) {
+            LOG.error(ExtensionService.SERVICE_NAME + " is not enabled.");
             throw FalconWebException.newAPIException(
                     ExtensionService.SERVICE_NAME + " is not enabled.", Response.Status.NOT_FOUND);
         }
@@ -773,7 +781,8 @@
 
     private static void checkIfExtensionIsEnabled(String extensionName) {
         ExtensionMetaStore metaStore = ExtensionStore.getMetaStore();
-        if (metaStore.getDetail(extensionName).getStatus().equals(ExtensionStatus.ENABLED)) {
+        if (!metaStore.getDetail(extensionName).getStatus().equals(ExtensionStatus.ENABLED)) {
+            LOG.error("Extension: " + extensionName + " is in disabled state.");
             throw FalconWebException.newAPIException("Extension: " + extensionName + " is in disabled state.",
                     Response.Status.INTERNAL_SERVER_ERROR);
         }
@@ -783,6 +792,7 @@
         ExtensionMetaStore metaStore = ExtensionStore.getMetaStore();
         ExtensionJobsBean extensionJobsBean = metaStore.getExtensionJobDetails(jobName);
         if (extensionJobsBean != null && !extensionJobsBean.getExtensionName().equals(extensionName)) {
+            LOG.error("Extension job with name: " + extensionName + " already exists.");
             throw FalconWebException.newAPIException("Extension job with name: " + extensionName + " already exists.",
                     Response.Status.INTERNAL_SERVER_ERROR);
         }
diff --git a/unit/src/main/java/org/apache/falcon/unit/FalconUnitClient.java b/unit/src/main/java/org/apache/falcon/unit/FalconUnitClient.java
index 2a40611..6a65d2c 100644
--- a/unit/src/main/java/org/apache/falcon/unit/FalconUnitClient.java
+++ b/unit/src/main/java/org/apache/falcon/unit/FalconUnitClient.java
@@ -334,6 +334,15 @@
     }
 
     @Override
+    public APIResult scheduleExtensionJob(String jobName, String coloExpr, String doAsUser) {
+        try {
+            return localExtensionManager.scheduleExtensionJob(jobName, coloExpr, doAsUser);
+        } catch (FalconException | IOException e) {
+            throw new FalconCLIException("Failed to delete the extension job:" + coloExpr);
+        }
+    }
+
+    @Override
     public APIResult submitAndScheduleExtensionJob(String extensionName, String jobName, String configPath,
                                                    String doAsUser) {
         InputStream configStream = getServletInputStream(configPath);
diff --git a/unit/src/main/java/org/apache/falcon/unit/LocalExtensionManager.java b/unit/src/main/java/org/apache/falcon/unit/LocalExtensionManager.java
index 1e9b15a..20ccfca 100644
--- a/unit/src/main/java/org/apache/falcon/unit/LocalExtensionManager.java
+++ b/unit/src/main/java/org/apache/falcon/unit/LocalExtensionManager.java
@@ -40,55 +40,75 @@
  * A proxy implementation of the extension operations in local mode.
  */
 public class LocalExtensionManager extends AbstractExtensionManager {
-    public LocalExtensionManager() {}
+    LocalExtensionManager() {}
 
-    public APIResult submitExtensionJob(String extensionName, String jobName, InputStream config,
-                                        SortedMap<EntityType, List<Entity>> entityMap)
+    APIResult submitExtensionJob(String extensionName, String jobName, InputStream configStream,
+                                 SortedMap<EntityType, List<Entity>> entityMap)
         throws FalconException, IOException {
-
-        for(Map.Entry<EntityType, List<Entity>> entry : entityMap.entrySet()){
-            for(Entity entity : entry.getValue()){
+        for (Map.Entry<EntityType, List<Entity>> entry : entityMap.entrySet()) {
+            for (Entity entity : entry.getValue()) {
                 submitInternal(entity, "falconUser");
             }
         }
+        storeExtension(extensionName, jobName, configStream, entityMap);
+
         return new APIResult(APIResult.Status.SUCCEEDED, "Extension job submitted successfully" + jobName);
     }
 
-    public APIResult submitAndSchedulableExtensionJob(String extensionName, String jobName, InputStream configStream,
-                                                      SortedMap<EntityType, List<Entity>> entityMap)
+    APIResult submitAndSchedulableExtensionJob(String extensionName, String jobName, InputStream configStream,
+                                               SortedMap<EntityType, List<Entity>> entityMap)
         throws FalconException, IOException {
-        List<String> feedNames = new ArrayList<>();
-        List<String> processNames = new ArrayList<>();
         for(Map.Entry<EntityType, List<Entity>> entry : entityMap.entrySet()){
             for(Entity entity : entry.getValue()){
                 submitInternal(entity, "falconUser");
             }
         }
 
-        for(Map.Entry<EntityType, List<Entity>> entry : entityMap.entrySet()){
-            for(Entity entity : entry.getValue()){
+        storeExtension(extensionName, jobName, configStream, entityMap);
+
+        for (Map.Entry<EntityType, List<Entity>> entry : entityMap.entrySet()) {
+            for (Entity entity : entry.getValue()) {
                 scheduleInternal(entry.getKey().name(), entity.getName(), null, null);
             }
         }
+        return new APIResult(APIResult.Status.SUCCEEDED, "Extension job submitted and scheduled successfully"
+                + jobName);
+    }
+
+    private void storeExtension(String extensionName, String jobName, InputStream configStream, SortedMap<EntityType,
+            List<Entity>> entityMap) throws IOException {
         byte[] configBytes = null;
         if (configStream != null) {
             configBytes = IOUtils.toByteArray(configStream);
         }
-        for(Map.Entry<EntityType, List<Entity>> entry : entityMap.entrySet()){
-            for(final Entity entity : entry.getValue()){
-                if (entity.getEntityType().equals(EntityType.FEED)){
+        List<String> feedNames = new ArrayList<>();
+        List<String> processNames = new ArrayList<>();
+        for (Map.Entry<EntityType, List<Entity>> entry : entityMap.entrySet()) {
+            for (final Entity entity : entry.getValue()) {
+                if (entity.getEntityType().equals(EntityType.FEED)) {
                     feedNames.add(entity.getName());
-                }else{
+                } else {
                     processNames.add(entity.getName());
                 }
             }
         }
         ExtensionStore.getMetaStore().storeExtensionJob(jobName, extensionName, feedNames, processNames, configBytes);
-
-        return new APIResult(APIResult.Status.SUCCEEDED, "Extension job submitted successfully" + jobName);
     }
 
-    public APIResult deleteExtensionJob(String jobName) throws FalconException, IOException{
+    APIResult scheduleExtensionJob(String jobName, String coloExpr, String doAsUser)
+        throws FalconException, IOException{
+        ExtensionMetaStore metaStore = ExtensionStore.getMetaStore();
+        ExtensionJobsBean extensionJobsBean = metaStore.getExtensionJobDetails(jobName);
+        SortedMap<EntityType, List<Entity>> entityMap = getJobEntities(extensionJobsBean);
+        for (Map.Entry<EntityType, List<Entity>> entry : entityMap.entrySet()) {
+            for (Entity entity : entry.getValue()) {
+                scheduleInternal(entity.getEntityType().name(), entity.getName(), true, null);
+            }
+        }
+        return new APIResult(APIResult.Status.SUCCEEDED, "Extension job " + jobName + " scheduled successfully");
+    }
+
+    APIResult deleteExtensionJob(String jobName) throws FalconException, IOException {
         ExtensionMetaStore metaStore = ExtensionStore.getMetaStore();
         ExtensionJobsBean extensionJobsBean = metaStore.getExtensionJobDetails(jobName);
         SortedMap<EntityType, List<Entity>> entityMap = getJobEntities(extensionJobsBean);
@@ -101,8 +121,8 @@
         return new APIResult(APIResult.Status.SUCCEEDED, "Extension job " + jobName + " deleted successfully");
     }
 
-    public APIResult updateExtensionJob(String extensionName, String jobName, InputStream configStream,
-                                        SortedMap<EntityType, List<Entity>> entityMap)
+    APIResult updateExtensionJob(String extensionName, String jobName, InputStream configStream,
+                                 SortedMap<EntityType, List<Entity>> entityMap)
         throws FalconException, IOException {
         List<String> feedNames = new ArrayList<>();
         List<String> processNames = new ArrayList<>();
@@ -128,27 +148,27 @@
         return new APIResult(APIResult.Status.SUCCEEDED, "Updated successfully");
     }
 
-    public APIResult registerExtensionMetadata(String extensionName, String packagePath , String description) {
+    APIResult registerExtensionMetadata(String extensionName, String packagePath, String description) {
         return super.registerExtensionMetadata(extensionName, packagePath, description, CurrentUser.getUser());
     }
 
-    public APIResult unRegisterExtension(String extensionName) {
+    APIResult unRegisterExtension(String extensionName) {
         return super.deleteExtensionMetadata(extensionName);
     }
 
-    public APIResult getExtensionJobDetails(String jobName){
+    APIResult getExtensionJobDetails(String jobName){
         return super.getExtensionJobDetail(jobName);
     }
 
-    public APIResult disableExtension(String extensionName) {
+    APIResult disableExtension(String extensionName) {
         return new APIResult(APIResult.Status.SUCCEEDED, super.disableExtension(extensionName, CurrentUser.getUser()));
     }
 
-    public APIResult enableExtension(String extensionName) {
+    APIResult enableExtension(String extensionName) {
         return new APIResult(APIResult.Status.SUCCEEDED, super.disableExtension(extensionName, CurrentUser.getUser()));
     }
 
-    public APIResult getExtensionDetails(String extensionName){
+    APIResult getExtensionDetails(String extensionName){
         return super.getExtensionDetail(extensionName);
     }
 
diff --git a/unit/src/test/java/org/apache/falcon/unit/TestFalconUnit.java b/unit/src/test/java/org/apache/falcon/unit/TestFalconUnit.java
index a41743d..508a7bb 100644
--- a/unit/src/test/java/org/apache/falcon/unit/TestFalconUnit.java
+++ b/unit/src/test/java/org/apache/falcon/unit/TestFalconUnit.java
@@ -444,7 +444,7 @@
 
         createDir(PROCESS_APP_PATH);
         copyExtensionJar(packageBuildLib);
-        APIResult apiResult = submitAndScheduleExtensionJob(TEST_EXTENSION, TEST_JOB, null, null);
+        APIResult apiResult = submitExtensionJob(TEST_EXTENSION, TEST_JOB, null, null);
         assertStatus(apiResult);
         result = getExtensionJobDetails(TEST_JOB);
         JSONObject resultJson = new JSONObject(result);
@@ -452,6 +452,8 @@
         Process process = (Process) getClient().getDefinition(EntityType.PROCESS.toString(), "sample", null);
         Assert.assertEquals(process.getPipelines(), "testPipeline");
 
+        apiResult = getClient().scheduleExtensionJob(TEST_JOB, null, null);
+        assertStatus(apiResult);
         apiResult = getClient().getStatus(EntityType.PROCESS, "sample", CLUSTER_NAME, null, false);
         assertStatus(apiResult);
         Assert.assertEquals(apiResult.getMessage(), "RUNNING");
diff --git a/webapp/src/main/java/org/apache/falcon/resource/ExtensionManager.java b/webapp/src/main/java/org/apache/falcon/resource/ExtensionManager.java
index 2160320..3a6c9c0 100644
--- a/webapp/src/main/java/org/apache/falcon/resource/ExtensionManager.java
+++ b/webapp/src/main/java/org/apache/falcon/resource/ExtensionManager.java
@@ -17,39 +17,93 @@
  */
 package org.apache.falcon.resource;
 
+import com.sun.jersey.multipart.FormDataBodyPart;
+import com.sun.jersey.multipart.FormDataParam;
 import org.apache.falcon.FalconWebException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.DefaultValue;
 import javax.ws.rs.GET;
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
 import javax.ws.rs.POST;
 import javax.ws.rs.Consumes;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.Context;
 import javax.ws.rs.core.MediaType;
-import javax.ws.rs.core.Response;
+import java.io.InputStream;
+import java.util.List;
 
 /**
  * This class provides RESTful API for the extensions.
  */
 @Path("extension")
-public class ExtensionManager {
-    public static final Logger LOG = LoggerFactory.getLogger(ExtensionManager.class);
+public class ExtensionManager extends AbstractExtensionManager {
+    private static final Logger LOG = LoggerFactory.getLogger(ExtensionManager.class);
 
     @GET
     @Path("enumerate")
     @Produces({MediaType.APPLICATION_JSON})
-    public Response getExtensions() {
+    public APIResult getExtensions() {
         LOG.error("Enumerate is not supported on Server.Please run your operation on Prism ");
         throw FalconWebException.newAPIException("Enumerate is not supported on Server. Please run your operation "
                 + "on Prism.");
     }
 
+    @POST
+    @Path("schedule/{job-name}")
+    @Consumes({MediaType.TEXT_XML, MediaType.TEXT_PLAIN})
+    @Produces({MediaType.TEXT_XML, MediaType.TEXT_PLAIN, MediaType.APPLICATION_JSON})
+    public APIResult schedule(@PathParam("job-name") String jobName,
+                              @Context HttpServletRequest request,
+                              @DefaultValue("") @QueryParam("doAs") String doAsUser) {
+        LOG.error("schedule is not supported on Server.Please run your operation on Prism ");
+        throw FalconWebException.newAPIException("schedule is not supported on Server. Please run your operation "
+                + "on Prism.");
+    }
+
+    @POST
+    @Path("submit/{extension-name}")
+    @Consumes({MediaType.TEXT_XML, MediaType.TEXT_PLAIN, MediaType.MULTIPART_FORM_DATA,
+            MediaType.APPLICATION_OCTET_STREAM})
+    @Produces({MediaType.TEXT_XML, MediaType.TEXT_PLAIN, MediaType.APPLICATION_JSON})
+    public APIResult submit(
+            @PathParam("extension-name") String extensionName,
+            @Context HttpServletRequest request,
+            @DefaultValue("") @QueryParam("doAs") String doAsUser,
+            @QueryParam("jobName") String jobName,
+            @FormDataParam("processes") List<FormDataBodyPart> processForms,
+            @FormDataParam("feeds") List<FormDataBodyPart> feedForms,
+            @FormDataParam("config") InputStream config) {
+        LOG.error("submit is not supported on Server.Please run your operation on Prism ");
+        throw FalconWebException.newAPIException("submit is not supported on Server. Please run your operation "
+                + "on Prism.");
+    }
+
+    @POST
+    @Path("submitAndSchedule/{extension-name}")
+    @Consumes({MediaType.TEXT_XML, MediaType.TEXT_PLAIN, MediaType.MULTIPART_FORM_DATA})
+    @Produces({MediaType.TEXT_XML, MediaType.TEXT_PLAIN, MediaType.APPLICATION_JSON})
+    public APIResult submitAndSchedule(
+            @PathParam("extension-name") String extensionName,
+            @Context HttpServletRequest request,
+            @DefaultValue("") @QueryParam("doAs") String doAsUser,
+            @QueryParam("jobName") String jobName,
+            @FormDataParam("processes") List<FormDataBodyPart> processForms,
+            @FormDataParam("feeds") List<FormDataBodyPart> feedForms,
+            @FormDataParam("config") InputStream config) {
+        LOG.error("submitAndSchedule is not supported on Server.Please run your operation on Prism ");
+        throw FalconWebException.newAPIException("submitAndSchedule is not supported on Server. Please run your "
+                + "operation on Prism.");
+    }
+
     @GET
     @Path("describe/{extension-name}")
     @Produces(MediaType.TEXT_PLAIN)
-    public String getExtensionDescription(
+    public APIResult getExtensionDescription(
             @PathParam("extension-name") String extensionName) {
         LOG.error("Describe is not supported on Server.Please run your operation on Prism ");
         throw FalconWebException.newAPIException("Describe is not supported on Server. Please run your operation "
@@ -59,7 +113,7 @@
     @GET
     @Path("detail/{extension-name}")
     @Produces({MediaType.APPLICATION_JSON})
-    public Response getDetail(@PathParam("extension-name") String extensionName) {
+    public APIResult getDetail(@PathParam("extension-name") String extensionName) {
         LOG.error("Detail is not supported on Server.Please run your operation on Prism ");
         throw FalconWebException.newAPIException("Detail is not supported on Server. Please run your operation "
                 + "on Prism.");
@@ -69,8 +123,8 @@
     @Path("unregister/{extension-name}")
     @Consumes({MediaType.TEXT_XML, MediaType.TEXT_PLAIN})
     @Produces(MediaType.TEXT_PLAIN)
-    public String deleteExtensionMetadata(
-            @PathParam("extension-name") String extensionName){
+    public APIResult deleteExtensionMetadata(
+            @PathParam("extension-name") String extensionName) {
         LOG.error("Unregister is not supported on Server.Please run your operation on Prism ");
         throw FalconWebException.newAPIException("Unregister is not supported on Server. Please run your operation "
                 + "on Prism.");
@@ -79,7 +133,7 @@
     @GET
     @Path("definition/{extension-name}")
     @Produces({MediaType.APPLICATION_JSON})
-    public String getExtensionDefinition(
+    public APIResult getExtensionDefinition(
             @PathParam("extension-name") String extensionName) {
         LOG.error("Definition is not supported on Server.Please run your operation on Prism ");
         throw FalconWebException.newAPIException("Definition is not supported on Server. Please run your operation "