diff --git a/awx/api/views.py b/awx/api/views.py
index 05f59b32ae..8146381e1b 100644
--- a/awx/api/views.py
+++ b/awx/api/views.py
@@ -2969,7 +2969,7 @@ class WorkflowJobTemplateLaunch(WorkflowsEnforcementMixin, RetrieveAPIView):
 
         prompted_fields, ignored_fields = obj._accept_or_ignore_job_kwargs(**request.data)
 
-        new_job = obj.create_workflow_job(**prompted_fields)
+        new_job = obj.create_unified_job(**prompted_fields)
         new_job.signal_start(**prompted_fields)
 
         data = OrderedDict()
diff --git a/awx/main/models/jobs.py b/awx/main/models/jobs.py
index 514b37db36..b141843b96 100644
--- a/awx/main/models/jobs.py
+++ b/awx/main/models/jobs.py
@@ -4,7 +4,6 @@
 # Python
 import datetime
 import hmac
-import json
 import logging
 import time
 from urlparse import urljoin
@@ -589,23 +588,6 @@ class Job(UnifiedJob, JobOptions, SurveyJobMixin, JobNotificationMixin):
                          hosts=all_hosts))
         return data
 
-    def handle_extra_data(self, extra_data):
-        extra_vars = {}
-        if isinstance(extra_data, dict):
-            extra_vars = extra_data
-        elif extra_data is None:
-            return
-        else:
-            if extra_data == "":
-                return
-            try:
-                extra_vars = json.loads(extra_data)
-            except Exception as e:
-                logger.warn("Exception deserializing extra vars: " + str(e))
-        evars = self.extra_vars_dict
-        evars.update(extra_vars)
-        self.update_fields(extra_vars=json.dumps(evars))
-
     def _resources_sufficient_for_launch(self):
         if self.job_type == PERM_INVENTORY_SCAN:
             return self.inventory_id is not None
@@ -1356,23 +1338,6 @@ class SystemJob(UnifiedJob, SystemJobOptions, JobNotificationMixin):
     def get_ui_url(self):
         return urljoin(settings.TOWER_URL_BASE, "/#/management_jobs/{}".format(self.pk))
 
-    def handle_extra_data(self, extra_data):
-        extra_vars = {}
-        if isinstance(extra_data, dict):
-            extra_vars = extra_data
-        elif extra_data is None:
-            return
-        else:
-            if extra_data == "":
-                return
-            try:
-                extra_vars = json.loads(extra_data)
-            except Exception as e:
-                logger.warn("Exception deserializing extra vars: " + str(e))
-        evars = self.extra_vars_dict
-        evars.update(extra_vars)
-        self.update_fields(extra_vars=json.dumps(evars))
-
     @property
     def task_impact(self):
         return 150
diff --git a/awx/main/models/unified_jobs.py b/awx/main/models/unified_jobs.py
index 1a6fa0351e..b210b3ce6d 100644
--- a/awx/main/models/unified_jobs.py
+++ b/awx/main/models/unified_jobs.py
@@ -819,7 +819,22 @@ class UnifiedJob(PolymorphicModel, PasswordFieldsModel, CommonModelNameNotUnique
         return []
 
     def handle_extra_data(self, extra_data):
-        return
+        if hasattr(self, 'extra_vars'):
+            extra_vars = {}
+            if isinstance(extra_data, dict):
+                extra_vars = extra_data
+            elif extra_data is None:
+                return
+            else:
+                if extra_data == "":
+                    return
+                try:
+                    extra_vars = json.loads(extra_data)
+                except Exception as e:
+                    logger.warn("Exception deserializing extra vars: " + str(e))
+            evars = self.extra_vars_dict
+            evars.update(extra_vars)
+            self.update_fields(extra_vars=json.dumps(evars))
 
     @property
     def can_start(self):
diff --git a/awx/main/models/workflow.py b/awx/main/models/workflow.py
index 112bdcfc76..a864e57390 100644
--- a/awx/main/models/workflow.py
+++ b/awx/main/models/workflow.py
@@ -392,8 +392,8 @@ class WorkflowJobTemplate(UnifiedJobTemplate, WorkflowJobOptions, SurveyJobTempl
                     success=list(success_notification_templates),
                     any=list(any_notification_templates))
 
-    def create_workflow_job(self, **kwargs):
-        workflow_job = self.create_unified_job(**kwargs)
+    def create_unified_job(self, **kwargs):
+        workflow_job = super(WorkflowJobTemplate, self).create_unified_job(**kwargs)
         workflow_job.copy_nodes_from_original(original=self)
         return workflow_job