diff --git a/core/src/main/java/org/apache/sqoop/connector/ConnectorHandler.java b/core/src/main/java/org/apache/sqoop/connector/ConnectorHandler.java
index d17ebef3..87822091 100644
--- a/core/src/main/java/org/apache/sqoop/connector/ConnectorHandler.java
+++ b/core/src/main/java/org/apache/sqoop/connector/ConnectorHandler.java
@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 import java.net.URL;
+import java.util.ArrayList;
 import java.util.Properties;
 
 import org.apache.log4j.Logger;
@@ -29,6 +30,7 @@
 import org.apache.sqoop.model.MConnector;
 import org.apache.sqoop.common.SqoopException;
 import org.apache.sqoop.connector.spi.SqoopConnector;
+import org.apache.sqoop.model.MForm;
 import org.apache.sqoop.model.MJobForms;
 
 public final class ConnectorHandler {
@@ -91,13 +93,23 @@ public ConnectorHandler(URL configFileUrl) {
     }
 
     // Initialize Metadata
-    MJobForms fromJobForms = new MJobForms(FormUtils.toForms(
-      connector.getJobConfigurationClass(Direction.FROM)));
+    MJobForms fromJobForms = null;
+    MJobForms toJobForms = null;
+    if (connector.getSupportedDirections().contains(Direction.FROM)) {
+      fromJobForms = new MJobForms(FormUtils.toForms(
+          connector.getJobConfigurationClass(Direction.FROM)));
+    } else {
+      fromJobForms = new MJobForms(new ArrayList<MForm>());
+    }
+
+    if (connector.getSupportedDirections().contains(Direction.TO)) {
+      toJobForms = new MJobForms(FormUtils.toForms(
+          connector.getJobConfigurationClass(Direction.TO)));
+    } else {
+      toJobForms = new MJobForms(new ArrayList<MForm>());
+    }
+
     MConnectionForms connectionForms = new MConnectionForms(
-      FormUtils.toForms(connector.getConnectionConfigurationClass()));
-    MJobForms toJobForms = new MJobForms(FormUtils.toForms(
-        connector.getJobConfigurationClass(Direction.TO)));
-    MConnectionForms toConnectionForms = new MConnectionForms(
         FormUtils.toForms(connector.getConnectionConfigurationClass()));
 
     String connectorVersion = connector.getVersion();
diff --git a/core/src/main/java/org/apache/sqoop/framework/FrameworkError.java b/core/src/main/java/org/apache/sqoop/framework/FrameworkError.java
index f7a6b67f..8ecb1972 100644
--- a/core/src/main/java/org/apache/sqoop/framework/FrameworkError.java
+++ b/core/src/main/java/org/apache/sqoop/framework/FrameworkError.java
@@ -46,6 +46,8 @@ public enum FrameworkError implements ErrorCode {
 
   FRAMEWORK_0010("Connection for this job has been disabled. Cannot submit this job."),
 
+  FRAMEWORK_0011("Connector does not support direction. Cannot submit this job."),
+
   ;
 
   private final String message;
diff --git a/core/src/main/java/org/apache/sqoop/framework/JobManager.java b/core/src/main/java/org/apache/sqoop/framework/JobManager.java
index 48360ea5..55719286 100644
--- a/core/src/main/java/org/apache/sqoop/framework/JobManager.java
+++ b/core/src/main/java/org/apache/sqoop/framework/JobManager.java
@@ -298,6 +298,17 @@ public MSubmission submit(long jobId, HttpEventContext ctx) {
     SqoopConnector toConnector =
         ConnectorManager.getInstance().getConnector(job.getConnectorId(Direction.TO));
 
+    // Make sure that connectors support the directions they will be used from.
+    if (!fromConnector.getSupportedDirections().contains(Direction.FROM)) {
+      throw new SqoopException(FrameworkError.FRAMEWORK_0011,
+          "Connector: " + fromConnector.getClass().getCanonicalName());
+    }
+
+    if (!toConnector.getSupportedDirections().contains(Direction.TO)) {
+      throw new SqoopException(FrameworkError.FRAMEWORK_0011,
+          "Connector: " + toConnector.getClass().getCanonicalName());
+    }
+
     // Transform forms to fromConnector specific classes
     Object fromConnectorConnection = ClassUtils.instantiate(
         fromConnector.getConnectionConfigurationClass());
diff --git a/server/src/main/java/org/apache/sqoop/handler/JobRequestHandler.java b/server/src/main/java/org/apache/sqoop/handler/JobRequestHandler.java
index 6f2df93d..fba5b1c9 100644
--- a/server/src/main/java/org/apache/sqoop/handler/JobRequestHandler.java
+++ b/server/src/main/java/org/apache/sqoop/handler/JobRequestHandler.java
@@ -186,6 +186,16 @@ private JsonBean createUpdateJob(RequestContext ctx, boolean update) {
     SqoopConnector toConnector =
       ConnectorManager.getInstance().getConnector(job.getConnectorId(Direction.TO));
 
+    if (!fromConnector.getSupportedDirections().contains(Direction.FROM)) {
+      throw new SqoopException(ServerError.SERVER_0004, "Connector " + fromConnector.getClass().getCanonicalName()
+          + " does not support FROM direction.");
+    }
+
+    if (!toConnector.getSupportedDirections().contains(Direction.TO)) {
+      throw new SqoopException(ServerError.SERVER_0004, "Connector " + toConnector.getClass().getCanonicalName()
+          + " does not support TO direction.");
+    }
+
     // Get validator objects
     Validator fromConnectorValidator = fromConnector.getValidator();
     Validator frameworkValidator = FrameworkManager.getInstance().getValidator();
diff --git a/spi/src/main/java/org/apache/sqoop/connector/spi/SqoopConnector.java b/spi/src/main/java/org/apache/sqoop/connector/spi/SqoopConnector.java
index 2112bf70..2774e8be 100644
--- a/spi/src/main/java/org/apache/sqoop/connector/spi/SqoopConnector.java
+++ b/spi/src/main/java/org/apache/sqoop/connector/spi/SqoopConnector.java
@@ -17,6 +17,8 @@
  */
 package org.apache.sqoop.connector.spi;
 
+import java.util.Arrays;
+import java.util.List;
 import java.util.Locale;
 import java.util.ResourceBundle;
 
@@ -45,6 +47,16 @@ public abstract class SqoopConnector {
    */
   public abstract ResourceBundle getBundle(Locale locale);
 
+  /**
+   * @return The supported directions
+   */
+  public List<Direction> getSupportedDirections() {
+    return Arrays.asList(new Direction[]{
+        Direction.FROM,
+        Direction.TO
+    });
+  }
+
   /**
    * @return Get connection configuration class
    */