diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index 3e524afc06..3f7c09a0a5 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -1019,7 +1019,7 @@ class TaskExecutor:
         Returns the correct action plugin to handle the requestion task action
         '''
 
-        module_prefix = self._task.action.split('_')[0]
+        module_prefix = self._task.action.split('.')[-1].split('_')[0]
 
         collections = self._task.collections
 
diff --git a/test/units/executor/test_task_executor.py b/test/units/executor/test_task_executor.py
index 6d44d7685d..d7baa00d75 100644
--- a/test/units/executor/test_task_executor.py
+++ b/test/units/executor/test_task_executor.py
@@ -19,6 +19,8 @@
 from __future__ import (absolute_import, division, print_function)
 __metaclass__ = type
 
+import mock
+
 from units.compat import unittest
 from units.compat.mock import patch, MagicMock
 from ansible.errors import AnsibleError
@@ -355,6 +357,108 @@ class TestTaskExecutor(unittest.TestCase):
         self.assertEqual(new_items, items)
         self.assertEqual(mock_task.args, {'name': '{{item.name}}', 'state': '{{item.state}}'})
 
+    def test_task_executor_get_action_handler(self):
+        te = TaskExecutor(
+            host=MagicMock(),
+            task=MagicMock(),
+            job_vars={},
+            play_context=MagicMock(),
+            new_stdin=None,
+            loader=DictDataLoader({}),
+            shared_loader_obj=MagicMock(),
+            final_q=MagicMock(),
+        )
+
+        action_loader = te._shared_loader_obj.action_loader
+        action_loader.has_plugin.return_value = True
+        action_loader.get.return_value = mock.sentinel.handler
+
+        mock_connection = MagicMock()
+        mock_templar = MagicMock()
+        action = 'namespace.prefix_sufix'
+        te._task.action = action
+
+        handler = te._get_action_handler(mock_connection, mock_templar)
+
+        self.assertIs(mock.sentinel.handler, handler)
+
+        action_loader.has_plugin.assert_called_once_with(
+            action, collection_list=te._task.collections)
+
+        action_loader.get.assert_called_once_with(
+            te._task.action, task=te._task, connection=mock_connection,
+            play_context=te._play_context, loader=te._loader,
+            templar=mock_templar, shared_loader_obj=te._shared_loader_obj,
+            collection_list=te._task.collections)
+
+    def test_task_executor_get_handler_prefix(self):
+        te = TaskExecutor(
+            host=MagicMock(),
+            task=MagicMock(),
+            job_vars={},
+            play_context=MagicMock(),
+            new_stdin=None,
+            loader=DictDataLoader({}),
+            shared_loader_obj=MagicMock(),
+            final_q=MagicMock(),
+        )
+
+        action_loader = te._shared_loader_obj.action_loader
+        action_loader.has_plugin.return_value = False
+        action_loader.get.return_value = mock.sentinel.handler
+        action_loader.__contains__.return_value = True
+
+        mock_connection = MagicMock()
+        mock_templar = MagicMock()
+        action = 'namespace.netconf_sufix'
+        te._task.action = action
+
+        handler = te._get_action_handler(mock_connection, mock_templar)
+
+        self.assertIs(mock.sentinel.handler, handler)
+        action_loader.has_plugin.assert_called_once_with(
+            action, collection_list=te._task.collections)
+
+        action_loader.get.assert_called_once_with(
+            'netconf', task=te._task, connection=mock_connection,
+            play_context=te._play_context, loader=te._loader,
+            templar=mock_templar, shared_loader_obj=te._shared_loader_obj,
+            collection_list=te._task.collections)
+
+    def test_task_executor_get_handler_normal(self):
+        te = TaskExecutor(
+            host=MagicMock(),
+            task=MagicMock(),
+            job_vars={},
+            play_context=MagicMock(),
+            new_stdin=None,
+            loader=DictDataLoader({}),
+            shared_loader_obj=MagicMock(),
+            final_q=MagicMock(),
+        )
+
+        action_loader = te._shared_loader_obj.action_loader
+        action_loader.has_plugin.return_value = False
+        action_loader.get.return_value = mock.sentinel.handler
+        action_loader.__contains__.return_value = False
+
+        mock_connection = MagicMock()
+        mock_templar = MagicMock()
+        action = 'namespace.prefix_sufix'
+        te._task.action = action
+
+        handler = te._get_action_handler(mock_connection, mock_templar)
+
+        self.assertIs(mock.sentinel.handler, handler)
+        action_loader.has_plugin.assert_called_once_with(
+            action, collection_list=te._task.collections)
+
+        action_loader.get.assert_called_once_with(
+            'normal', task=te._task, connection=mock_connection,
+            play_context=te._play_context, loader=te._loader,
+            templar=mock_templar, shared_loader_obj=te._shared_loader_obj,
+            collection_list=None)
+
     def test_task_executor_execute(self):
         fake_loader = DictDataLoader({})