diff --git a/changelogs/fragments/connection-pipeline.yml b/changelogs/fragments/connection-pipeline.yml
new file mode 100644
index 00000000000..8736a18e703
--- /dev/null
+++ b/changelogs/fragments/connection-pipeline.yml
@@ -0,0 +1,6 @@
+deprecated_features:
+  - >-
+    connection plugins - Added a soft deprecation on the connection attributes ``has_native_async`` and
+    ``always_pipeline_modules``. Connection plugins that wish to apply custom behaviour around pipelining should
+    instead override the method ``is_pipelining_enabled(self, wrap_async=False)`` added in Ansible 2.19. For backwards
+    compatibility no runtime deprecation warning is emitted but will be in the future.
diff --git a/lib/ansible/plugins/action/command.py b/lib/ansible/plugins/action/command.py
index df4dbe9f7c6..e0f60b42f53 100644
--- a/lib/ansible/plugins/action/command.py
+++ b/lib/ansible/plugins/action/command.py
@@ -14,7 +14,7 @@ class ActionModule(ActionBase):
         results = super(ActionModule, self).run(tmp, task_vars)
         del tmp  # tmp no longer has any effect
 
-        wrap_async = self._task.async_val and not self._connection.has_native_async
+        wrap_async = self._task.async_val
         # explicitly call `ansible.legacy.command` for backcompat to allow library/ override of `command` while not allowing
         # collections search for an unqualified `command` module
         results = merge_hash(results, self._execute_module(module_name='ansible.legacy.command', task_vars=task_vars, wrap_async=wrap_async))
diff --git a/lib/ansible/plugins/action/normal.py b/lib/ansible/plugins/action/normal.py
index 0476f9aacdf..96664288283 100644
--- a/lib/ansible/plugins/action/normal.py
+++ b/lib/ansible/plugins/action/normal.py
@@ -32,7 +32,7 @@ class ActionModule(ActionBase):
         result = super(ActionModule, self).run(tmp, task_vars)
         del tmp  # tmp no longer has any effect
 
-        wrap_async = self._task.async_val and not self._connection.has_native_async
+        wrap_async = self._task.async_val
 
         # do work!
         result = merge_hash(result, self._execute_module(task_vars=task_vars, wrap_async=wrap_async))
diff --git a/lib/ansible/plugins/connection/__init__.py b/lib/ansible/plugins/connection/__init__.py
index c592ed1d037..16a359e387d 100644
--- a/lib/ansible/plugins/connection/__init__.py
+++ b/lib/ansible/plugins/connection/__init__.py
@@ -55,8 +55,13 @@ class ConnectionBase(AnsiblePlugin):
     """
 
     has_pipelining = False
-    has_native_async = False  # eg, winrm
-    always_pipeline_modules = False  # eg, winrm
+
+    # The next 2 options are obsolete and should not be set to True. Plugins should instead override
+    # is_pipelining_enabled if they need to change the pipelining behaviour. We cannot mark as deprecated yet as some
+    # plugins may still need to set these for compatibility with older Ansible versions still in use.
+    has_native_async = False
+    always_pipeline_modules = False
+
     has_tty = True  # for interacting with become plugins
     # When running over this connection type, prefer modules written in a certain language
     # as discovered by the specified file extension.  An empty string as the
@@ -308,7 +313,12 @@ class ConnectionBase(AnsiblePlugin):
             except KeyError:
                 is_enabled = getattr(self._play_context, 'pipelining', False)
 
-        # TODO: deprecate always_pipeline_modules and has_native_async in favor for each plugin overriding this function
+        # We should deprecate always_pipeline_modules and has_native_async in favor of plugins just overriding this
+        # method. We cannot add it now as plugins will need to be compatible with older Ansible versions that still need
+        # to set this to be true. This function was added in 2.19 and when we expect 2.19 being a baseline for plugins
+        # we can add a runtime deprecation that checks whether these are True and if so warn the function should be
+        # overridden instead.
+        # deprecate: description='Revisit adding a deprecation or bump the core_version' core_version='2.24'
         conditions = [
             is_enabled or self.always_pipeline_modules,       # enabled via config or forced via connection (eg winrm)
             not wrap_async or self.has_native_async,          # async does not normally support pipelining unless it does (eg winrm)
diff --git a/lib/ansible/plugins/connection/psrp.py b/lib/ansible/plugins/connection/psrp.py
index 67911649cac..6ec36728fb6 100644
--- a/lib/ansible/plugins/connection/psrp.py
+++ b/lib/ansible/plugins/connection/psrp.py
@@ -369,9 +369,6 @@ class Connection(ConnectionBase):
     _shell: PowerShellPlugin
 
     def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
-        self.always_pipeline_modules = True
-        self.has_native_async = True
-
         self.runspace: RunspacePool | None = None
         self.host: PSHost | None = None
         self._last_pipeline: PowerShell | None = None
@@ -790,3 +787,6 @@ class Connection(ConnectionBase):
         self.host.ui.stderr = []
 
         return rc, to_bytes(stdout, encoding='utf-8'), to_bytes(stderr, encoding='utf-8')
+
+    def is_pipelining_enabled(self, wrap_async: bool = False) -> bool:
+        return True
diff --git a/lib/ansible/plugins/connection/ssh.py b/lib/ansible/plugins/connection/ssh.py
index b6ae4994a83..3fa2a1b2357 100644
--- a/lib/ansible/plugins/connection/ssh.py
+++ b/lib/ansible/plugins/connection/ssh.py
@@ -660,8 +660,6 @@ class Connection(ConnectionBase):
         # we need to set various properties to ensure SSH on Windows continues
         # to work
         if getattr(self._shell, "_IS_WINDOWS", False):
-            self.has_native_async = True
-            self.always_pipeline_modules = True
             self.module_implementation_preferences = ('.ps1', '.exe', '')
             self.allow_executable = False
 
@@ -1618,7 +1616,10 @@ class Connection(ConnectionBase):
     def is_pipelining_enabled(self, wrap_async=False):
         """ override parent method and ensure we don't request a tty """
 
-        if self._is_tty_requested():
+        if getattr(self._shell, "_IS_WINDOWS", False):
+            # pipelining is always used on Windows
+            return True
+        elif self._is_tty_requested():
             return False
         else:
             return super(Connection, self).is_pipelining_enabled(wrap_async)
diff --git a/lib/ansible/plugins/connection/winrm.py b/lib/ansible/plugins/connection/winrm.py
index a46ff8e6830..df89bdcd1bc 100644
--- a/lib/ansible/plugins/connection/winrm.py
+++ b/lib/ansible/plugins/connection/winrm.py
@@ -245,9 +245,6 @@ class Connection(ConnectionBase):
 
     def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
 
-        self.always_pipeline_modules = True
-        self.has_native_async = True
-
         self.protocol: winrm.Protocol | None = None
         self.shell_id: str | None = None
         self.delegate = None
@@ -879,3 +876,6 @@ class Connection(ConnectionBase):
         self.shell_id = None
         self.protocol = None
         self._connected = False
+
+    def is_pipelining_enabled(self, wrap_async: bool = False) -> bool:
+        return True
diff --git a/test/integration/targets/async_fail/action_plugins/normal.py b/test/integration/targets/async_fail/action_plugins/normal.py
index 22e2a3faeea..5d70d3f5a55 100644
--- a/test/integration/targets/async_fail/action_plugins/normal.py
+++ b/test/integration/targets/async_fail/action_plugins/normal.py
@@ -40,7 +40,7 @@ class ActionModule(ActionBase):
                 del result['invocation']['module_args']
 
             # FUTURE: better to let _execute_module calculate this internally?
-            wrap_async = self._task.async_val and not self._connection.has_native_async
+            wrap_async = self._task.async_val
 
             # do work!
             result = merge_hash(result, self._execute_module(task_vars=task_vars, wrap_async=wrap_async))