diff --git a/changelogs/fragments/extras_fix.yml b/changelogs/fragments/extras_fix.yml
new file mode 100644
index 00000000000..9d8e24594f4
--- /dev/null
+++ b/changelogs/fragments/extras_fix.yml
@@ -0,0 +1,4 @@
+bugfixes:
+  - connection plugins using the 'extras' option feature would need variables to match the plugin's loaded name,
+    sometimes requiring fqcn, which is not the same as the documented/declared/expected variables.
+    Now we fall back to the 'basename' of the fqcn, but plugin authors can still set the expected value directly.
diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index d2bee161864..a400df6781e 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -1068,7 +1068,7 @@ class TaskExecutor:
         # add extras if plugin supports them
         if getattr(self._connection, 'allow_extras', False):
             for k in variables:
-                if k.startswith('ansible_%s_' % self._connection._load_name) and k not in options:
+                if k.startswith('ansible_%s_' % self._connection.extras_prefix) and k not in options:
                     options['_extras'][k] = templar.template(variables[k])
 
         task_keys = self._task.dump_attrs()
diff --git a/lib/ansible/plugins/__init__.py b/lib/ansible/plugins/__init__.py
index 63d087b0806..23f11d170e0 100644
--- a/lib/ansible/plugins/__init__.py
+++ b/lib/ansible/plugins/__init__.py
@@ -50,16 +50,23 @@ def get_plugin_class(obj):
 
 class AnsiblePlugin(ABC):
 
-    # allow extra passthrough parameters
-    allow_extras = False
-
     # Set by plugin loader
     _load_name: str
 
+    # allow extra passthrough parameters
+    allow_extras: bool = False
+    _extras_prefix: str | None = None
+
     def __init__(self):
         self._options = {}
         self._defs = None
 
+    @property
+    def extras_prefix(self):
+        if not self._extras_prefix:
+            self._extras_prefix = self._load_name.split('.')[-1]
+        return self._extras_prefix
+
     def matches_name(self, possible_names):
         possible_fqcns = set()
         for name in possible_names:
diff --git a/test/units/plugins/connection/test_psrp.py b/test/units/plugins/connection/test_psrp.py
index d1c5fe821a5..de0def01fc0 100644
--- a/test/units/plugins/connection/test_psrp.py
+++ b/test/units/plugins/connection/test_psrp.py
@@ -218,12 +218,13 @@ class TestConnectionPSRP(object):
         pc = PlayContext()
         new_stdin = StringIO()
 
-        conn = connection_loader.get('psrp', pc, new_stdin)
-        conn.set_options(var_options={'_extras': {'ansible_psrp_mock_test3': True}})
+        for conn_name in ('psrp', 'ansible.legacy.psrp'):
+            conn = connection_loader.get(conn_name, pc, new_stdin)
+            conn.set_options(var_options={'_extras': {'ansible_psrp_mock_test3': True}})
 
-        mock_display = MagicMock()
-        monkeypatch.setattr(Display, "warning", mock_display)
-        conn._build_kwargs()
+            mock_display = MagicMock()
+            monkeypatch.setattr(Display, "warning", mock_display)
+            conn._build_kwargs()
 
-        assert mock_display.call_args[0][0] == \
-            'ansible_psrp_mock_test3 is unsupported by the current psrp version installed'
+            assert mock_display.call_args[0][0] == \
+                'ansible_psrp_mock_test3 is unsupported by the current psrp version installed'