diff --git a/awx/main/models/unified_jobs.py b/awx/main/models/unified_jobs.py
index 926555fbab..e786ab9de5 100644
--- a/awx/main/models/unified_jobs.py
+++ b/awx/main/models/unified_jobs.py
@@ -1374,14 +1374,13 @@ class UnifiedJob(PolymorphicModel, PasswordFieldsModel, CommonModelNameNotUnique
 
         created_by = getattr_dne(self, 'created_by')
 
-        if not created_by:
-            wj = self.get_workflow_job()
-            if wj:
-                for name in ('awx', 'tower'):
-                    r['{}_workflow_job_id'.format(name)] = wj.pk
-                    r['{}_workflow_job_name'.format(name)] = wj.name
-                created_by = getattr_dne(wj, 'created_by')
+        wj = self.get_workflow_job()
+        if wj:
+            for name in ('awx', 'tower'):
+                r['{}_workflow_job_id'.format(name)] = wj.pk
+                r['{}_workflow_job_name'.format(name)] = wj.name
 
+        if not created_by:
             schedule = getattr_dne(self, 'schedule')
             if schedule:
                 for name in ('awx', 'tower'):
diff --git a/awx/main/models/workflow.py b/awx/main/models/workflow.py
index 17b869f418..2c0beba698 100644
--- a/awx/main/models/workflow.py
+++ b/awx/main/models/workflow.py
@@ -276,6 +276,8 @@ class WorkflowJobNode(WorkflowNodeBase):
             data['extra_vars'] = extra_vars
         # ensure that unified jobs created by WorkflowJobs are marked
         data['_eager_fields'] = {'launch_type': 'workflow'}
+        if self.workflow_job and self.workflow_job.created_by:
+            data['_eager_fields']['created_by'] = self.workflow_job.created_by
         # Extra processing in the case that this is a slice job
         if 'job_slice' in self.ancestor_artifacts and is_root_node:
             data['_eager_fields']['allow_simultaneous'] = True
diff --git a/awx/main/tests/functional/models/test_unified_job.py b/awx/main/tests/functional/models/test_unified_job.py
index 18e80b63cc..8596055ce0 100644
--- a/awx/main/tests/functional/models/test_unified_job.py
+++ b/awx/main/tests/functional/models/test_unified_job.py
@@ -112,15 +112,15 @@ class TestMetaVars:
         for key in user_vars:
             assert key not in job.awx_meta_vars()
 
-    def test_workflow_job_metavars(self, admin_user):
+    def test_workflow_job_metavars(self, admin_user, job_template):
         workflow_job = WorkflowJob.objects.create(
             name='workflow-job',
             created_by=admin_user
         )
-        job = Job.objects.create(
-            name='fake-job',
-            launch_type='workflow'
-        )
+        node = workflow_job.workflow_nodes.create(unified_job_template=job_template)
+        job_kv = node.get_job_kwargs()
+        job = node.unified_job_template.create_unified_job(**job_kv)
+
         workflow_job.workflow_nodes.create(job=job)
         data = job.awx_meta_vars()
         assert data['awx_user_id'] == admin_user.id