diff --git a/plugins/module_utils/k8s/client.py b/plugins/module_utils/k8s/client.py
index 4cfac198..0cef6c50 100644
--- a/plugins/module_utils/k8s/client.py
+++ b/plugins/module_utils/k8s/client.py
@@ -203,8 +203,10 @@ class K8SClient:
             params["dry_run"] = True
         return params
 
-    def validate(self, resource, **params):
-        pass
+    def validate(
+        self, resource, version: Optional[str] = None, strict: Optional[bool] = False
+    ):
+        return self.client.validate(resource, version, strict)
 
     def get(self, resource, **params):
         return resource.get(**params)
diff --git a/plugins/module_utils/k8s/core.py b/plugins/module_utils/k8s/core.py
index 4365922a..ac495ec9 100644
--- a/plugins/module_utils/k8s/core.py
+++ b/plugins/module_utils/k8s/core.py
@@ -109,6 +109,13 @@ def gather_versions() -> dict:
     except ImportError:
         pass
 
+    try:
+        import kubernetes_validate
+
+        versions["kubernetes-validate"] = kubernetes_validate.__version__
+    except ImportError:
+        pass
+
     try:
         import yaml
 
diff --git a/plugins/module_utils/k8s/runner.py b/plugins/module_utils/k8s/runner.py
index 2c9379d4..bce2f7ff 100644
--- a/plugins/module_utils/k8s/runner.py
+++ b/plugins/module_utils/k8s/runner.py
@@ -3,6 +3,8 @@
 
 from typing import Dict
 
+from ansible.module_utils._text import to_native
+
 from ansible_collections.kubernetes.core.plugins.module_utils.k8s.client import (
     get_api_client,
 )
@@ -15,7 +17,6 @@ from ansible_collections.kubernetes.core.plugins.module_utils.k8s.service import
 from ansible_collections.kubernetes.core.plugins.module_utils.k8s.exceptions import (
     CoreException,
 )
-
 from ansible_collections.kubernetes.core.plugins.module_utils.selector import (
     LabelSelectorFilter,
 )
@@ -42,32 +43,40 @@ def validate(client, module, resource):
 
 def run_module(module) -> None:
     results = []
-
+    changed = False
     client = get_api_client(module)
     svc = K8sService(client, module)
     definitions = create_definitions(module.params)
 
     for definition in definitions:
-        module.warnings = []
+        result = {"changed": False, "result": {}, "warnings": []}
+        warnings = []
 
-        if module.params["validate"] is not None:
-            module.warnings = validate(client, module, definition)
+        if module.params.get("validate") is not None:
+            warnings = validate(client, module, definition)
 
         try:
             result = perform_action(svc, definition, module.params)
         except CoreException as e:
-            if module.warnings:
-                e["msg"] += "\n" + "\n    ".join(module.warnings)
+            msg = to_native(e)
+            if warnings:
+                msg += "\n" + "\n    ".join(warnings)
             if module.params.get("continue_on_error"):
-                result = {"error": "{0}".format(e)}
+                result["error"] = {"msg": msg}
             else:
-                module.fail_json(msg=e)
-        if module.warnings:
-            result["warnings"] = module.warnings
+                module.fail_json(msg=msg)
 
+        if warnings:
+            result.setdefault("warnings", [])
+            result["warnings"] += warnings
+
+        changed |= result["changed"]
         results.append(result)
 
-    module.exit_json(**results)
+    if len(results) == 1:
+        module.exit_json(**results[0])
+
+    module.exit_json(**{"changed": changed, "result": {"results": results}})
 
 
 def perform_action(svc, definition: Dict, params: Dict) -> Dict:
@@ -75,9 +84,13 @@ def perform_action(svc, definition: Dict, params: Dict) -> Dict:
     namespace = definition["metadata"].get("namespace")
     label_selectors = params.get("label_selectors")
     state = params.get("state", None)
-    result = {}
+    kind = definition.get("kind")
+    api_version = definition.get("apiVersion")
+    result = {"changed": False, "result": {}}
 
-    resource = svc.find_resource(definition)
+    resource = svc.find_resource(kind, api_version, fail=True)
+    definition["kind"] = resource.kind
+    definition["apiVersion"] = resource.group_version
     existing = svc.retrieve(resource, definition)
 
     if state == "absent":
@@ -91,7 +104,7 @@ def perform_action(svc, definition: Dict, params: Dict) -> Dict:
                 result["msg"] = (
                     "resource 'kind={kind},name={name},namespace={namespace}' "
                     "filtered by label_selectors.".format(
-                        kind=definition["kind"], name=origin_name, namespace=namespace,
+                        kind=kind, name=origin_name, namespace=namespace,
                     )
                 )
                 return result
@@ -100,6 +113,14 @@ def perform_action(svc, definition: Dict, params: Dict) -> Dict:
             result = svc.apply(resource, definition, existing)
             result["method"] = "apply"
         elif not existing:
+            if state == "patched":
+                result.setdefault("warnings", []).append(
+                    "resource 'kind={kind},name={name}' was not found but will not be "
+                    "created as 'state' parameter has been set to '{state}'".format(
+                        kind=kind, name=definition["metadata"].get("name"), state=state
+                    )
+                )
+                return result
             result = svc.create(resource, definition)
             result["method"] = "create"
         elif params.get("force", False):
diff --git a/plugins/module_utils/k8s/service.py b/plugins/module_utils/k8s/service.py
index ffdd289f..6a23d770 100644
--- a/plugins/module_utils/k8s/service.py
+++ b/plugins/module_utils/k8s/service.py
@@ -171,29 +171,35 @@ class K8sService:
             msg = "Failed to patch object: {0}".format(reason)
             raise CoreException(msg) from e
 
-    def retrieve(self, resource: Resource, definition: Dict) -> Dict:
+    def retrieve(self, resource: Resource, definition: Dict) -> ResourceInstance:
         state = self.module.params.get("state", None)
         append_hash = self.module.params.get("append_hash", False)
         name = definition["metadata"].get("name")
+        generate_name = definition["metadata"].get("generateName")
         namespace = definition["metadata"].get("namespace")
         label_selectors = self.module.params.get("label_selectors")
-        results = {
-            "changed": False,
-            "result": {},
-        }
-        existing = None
+        existing: ResourceInstance = None
 
         try:
             # ignore append_hash for resources other than ConfigMap and Secret
             if append_hash and definition["kind"] in ["ConfigMap", "Secret"]:
-                name = "%s-%s" % (name, generate_hash(definition))
-                definition["metadata"]["name"] = name
-            params = dict(name=name)
+                if name:
+                    name = "%s-%s" % (name, generate_hash(definition))
+                    definition["metadata"]["name"] = name
+                elif generate_name:
+                    definition["metadata"]["generateName"] = "%s-%s" % (
+                        generate_name,
+                        generate_hash(definition),
+                    )
+            params = {}
+            if name:
+                params["name"] = name
             if namespace:
                 params["namespace"] = namespace
             if label_selectors:
                 params["label_selector"] = ",".join(label_selectors)
-            existing = self.client.get(resource, **params)
+            if "name" in params or "label_selector" in params:
+                existing = self.client.get(resource, **params)
         except (NotFoundError, MethodNotAllowedError):
             pass
         except ForbiddenError as e:
@@ -210,10 +216,7 @@ class K8sService:
             msg = "Failed to retrieve requested object: {0}".format(reason)
             raise CoreException(msg) from e
 
-        if existing:
-            results["result"] = existing.to_dict()
-
-        return results
+        return existing
 
     def find(
         self,
@@ -345,10 +348,12 @@ class K8sService:
         results["result"] = k8s_obj
 
         if wait and not self.module.check_mode:
-            definition["metadata"].update({"name": k8s_obj["metadata"]["name"]})
             waiter = get_waiter(self.client, resource, condition=wait_condition)
             success, results["result"], results["duration"] = waiter.wait(
-                timeout=wait_timeout, sleep=wait_sleep, name=name, namespace=namespace,
+                timeout=wait_timeout,
+                sleep=wait_sleep,
+                name=k8s_obj["metadata"]["name"],
+                namespace=namespace,
             )
 
         results["changed"] = True
@@ -358,7 +363,7 @@ class K8sService:
                 '"{0}" "{1}": Resource creation timed out'.format(
                     definition["kind"], origin_name
                 ),
-                **results
+                results,
             )
 
         return results
@@ -431,7 +436,7 @@ class K8sService:
                     '"{0}" "{1}": Resource apply timed out'.format(
                         definition["kind"], origin_name
                     ),
-                    **results
+                    results,
                 )
 
         return results
@@ -493,7 +498,7 @@ class K8sService:
                 '"{0}" "{1}": Resource replacement timed out'.format(
                     definition["kind"], origin_name
                 ),
-                **results
+                results,
             )
 
         return results
@@ -514,16 +519,25 @@ class K8sService:
             wait_condition = self.module.params["wait_condition"]
         results = {"changed": False, "result": {}}
 
-        if self.module.check_mode and not self.module.client.dry_run:
+        if self.module.check_mode and not self.client.dry_run:
             k8s_obj = dict_merge(existing.to_dict(), _encode_stringdata(definition))
         else:
+            exception = None
             for merge_type in self.module.params.get("merge_type") or [
                 "strategic-merge",
                 "merge",
             ]:
-                k8s_obj = self.patch_resource(
-                    resource, definition, name, namespace, merge_type=merge_type,
-                )
+                try:
+                    k8s_obj = self.patch_resource(
+                        resource, definition, name, namespace, merge_type=merge_type,
+                    )
+                    exception = None
+                except CoreException as e:
+                    exception = e
+                    continue
+                break
+            if exception:
+                raise exception
 
         success = True
         results["result"] = k8s_obj
@@ -545,7 +559,7 @@ class K8sService:
                 '"{0}" "{1}": Resource update timed out'.format(
                     definition["kind"], origin_name
                 ),
-                **results
+                results,
             )
 
         return results
@@ -581,6 +595,15 @@ class K8sService:
             if self.module.check_mode and not self.client.dry_run:
                 return results
             else:
+                if name:
+                    params["name"] = name
+
+                if namespace:
+                    params["namespace"] = namespace
+
+                if label_selectors:
+                    params["label_selector"] = ",".join(label_selectors)
+
                 if delete_options:
                     body = {
                         "apiVersion": "v1",
@@ -614,7 +637,7 @@ class K8sService:
                             '"{0}" "{1}": Resource deletion timed out'.format(
                                 definition["kind"], origin_name
                             ),
-                            **results
+                            results,
                         )
 
                 return results
diff --git a/plugins/module_utils/k8s/waiter.py b/plugins/module_utils/k8s/waiter.py
index 8ab60ada..d52491f4 100644
--- a/plugins/module_utils/k8s/waiter.py
+++ b/plugins/module_utils/k8s/waiter.py
@@ -157,7 +157,7 @@ class Waiter:
             try:
                 response = self.client.get(self.resource, **params)
             except NotFoundError:
-                pass
+                response = None
             if self.predicate(response):
                 break
         if response:
diff --git a/plugins/modules/k8s.py b/plugins/modules/k8s.py
index d7a42472..4b3c7462 100644
--- a/plugins/modules/k8s.py
+++ b/plugins/modules/k8s.py
@@ -391,6 +391,12 @@ from ansible_collections.kubernetes.core.plugins.module_utils.args_common import
     RESOURCE_ARG_SPEC,
     DELETE_OPTS_ARG_SPEC,
 )
+from ansible_collections.kubernetes.core.plugins.module_utils.k8s.core import (
+    AnsibleK8SModule,
+)
+from ansible_collections.kubernetes.core.plugins.module_utils.k8s.runner import (
+    run_module,
+)
 
 
 def validate_spec():
@@ -437,28 +443,6 @@ def argspec():
     return argument_spec
 
 
-def execute_module(module, k8s_ansible_mixin):
-    k8s_ansible_mixin.module = module
-    k8s_ansible_mixin.argspec = module.argument_spec
-    k8s_ansible_mixin.check_mode = k8s_ansible_mixin.module.check_mode
-    k8s_ansible_mixin.params = k8s_ansible_mixin.module.params
-    k8s_ansible_mixin.fail_json = k8s_ansible_mixin.module.fail_json
-    k8s_ansible_mixin.fail = k8s_ansible_mixin.module.fail_json
-    k8s_ansible_mixin.exit_json = k8s_ansible_mixin.module.exit_json
-    k8s_ansible_mixin.warn = k8s_ansible_mixin.module.warn
-    k8s_ansible_mixin.warnings = []
-
-    k8s_ansible_mixin.kind = k8s_ansible_mixin.params.get("kind")
-    k8s_ansible_mixin.api_version = k8s_ansible_mixin.params.get("api_version")
-    k8s_ansible_mixin.name = k8s_ansible_mixin.params.get("name")
-    k8s_ansible_mixin.generate_name = k8s_ansible_mixin.params.get("generate_name")
-    k8s_ansible_mixin.namespace = k8s_ansible_mixin.params.get("namespace")
-
-    k8s_ansible_mixin.check_library_version()
-    k8s_ansible_mixin.set_resource_definitions(module)
-    k8s_ansible_mixin.execute_module()
-
-
 def main():
     mutually_exclusive = [
         ("resource_definition", "src"),
@@ -467,19 +451,14 @@ def main():
         ("template", "src"),
         ("name", "generate_name"),
     ]
-    module = AnsibleModule(
+
+    module = AnsibleK8SModule(
+        module_class=AnsibleModule,
         argument_spec=argspec(),
         mutually_exclusive=mutually_exclusive,
         supports_check_mode=True,
     )
-    from ansible_collections.kubernetes.core.plugins.module_utils.common import (
-        K8sAnsibleMixin,
-        get_api_client,
-    )
-
-    k8s_ansible_mixin = K8sAnsibleMixin(module)
-    k8s_ansible_mixin.client = get_api_client(module=module)
-    execute_module(module, k8s_ansible_mixin)
+    run_module(module)
 
 
 if __name__ == "__main__":
diff --git a/tests/integration/targets/k8s_gc/tasks/main.yml b/tests/integration/targets/k8s_gc/tasks/main.yml
index 923ffc3f..a2f60c8a 100644
--- a/tests/integration/targets/k8s_gc/tasks/main.yml
+++ b/tests/integration/targets/k8s_gc/tasks/main.yml
@@ -5,7 +5,7 @@
     # This is a job definition that runs for 10 minutes and won't gracefully
     # shutdown. It allows us to test foreground vs background deletion.
     job_definition:
-      apiVersion: v1
+      apiVersion: batch/v1
       kind: Job
       metadata:
         name: "{{ gc_name }}"
diff --git a/tests/integration/targets/k8s_merge_type/tasks/main.yml b/tests/integration/targets/k8s_merge_type/tasks/main.yml
index 32e41d8b..3c544c76 100644
--- a/tests/integration/targets/k8s_merge_type/tasks/main.yml
+++ b/tests/integration/targets/k8s_merge_type/tasks/main.yml
@@ -97,6 +97,7 @@
   - name: patch service using json merge patch
     kubernetes.core.k8s:
       kind: Deployment
+      api_version: apps/v1
       namespace: "{{ k8s_patch_namespace }}"
       name: "{{ k8s_merge }}"
       merge_type:
diff --git a/tests/integration/targets/k8s_patched/tasks/main.yml b/tests/integration/targets/k8s_patched/tasks/main.yml
index 6fe514d3..8846d064 100644
--- a/tests/integration/targets/k8s_patched/tasks/main.yml
+++ b/tests/integration/targets/k8s_patched/tasks/main.yml
@@ -52,7 +52,7 @@
       assert:
         that:
           - patch_resource.changed
-          - patch_resource.result.results | selectattr('warning', 'defined') | list | length == 1
+          - patch_resource.result.results | selectattr('warnings', 'defined') | list | length == 1
 
     - name: Ensure namespace {{ patch_only_namespace[0] }} was patched correctly
       kubernetes.core.k8s_info:
diff --git a/tests/integration/targets/k8s_scale/tasks/main.yml b/tests/integration/targets/k8s_scale/tasks/main.yml
index d58a01bc..eb2107d3 100644
--- a/tests/integration/targets/k8s_scale/tasks/main.yml
+++ b/tests/integration/targets/k8s_scale/tasks/main.yml
@@ -150,7 +150,7 @@
     - name: Reapply the earlier deployment
       k8s:
         definition:
-          api_version: apps/v1
+          apiVersion: apps/v1
           kind: Deployment
           metadata:
             name: scale-deploy
diff --git a/tests/integration/targets/k8s_validate/tasks/main.yml b/tests/integration/targets/k8s_validate/tasks/main.yml
index b05e2a56..900e6f70 100644
--- a/tests/integration/targets/k8s_validate/tasks/main.yml
+++ b/tests/integration/targets/k8s_validate/tasks/main.yml
@@ -42,7 +42,7 @@
     - assert:
         that:
           - k8s_no_validate is failed
-          - "k8s_no_validate.msg == 'kubernetes-validate python library is required to validate resources'"
+          - "'Failed to import the required Python library (kubernetes-validate)' in k8s_no_validate.msg"
 
     - file:
         path: "{{ virtualenv }}"
diff --git a/tests/unit/module_utils/test_service.py b/tests/unit/module_utils/test_service.py
index 979f6d5c..8ef836d8 100644
--- a/tests/unit/module_utils/test_service.py
+++ b/tests/unit/module_utils/test_service.py
@@ -192,9 +192,8 @@ def test_service_retrieve_existing_resource(mock_pod_resource_instance):
     svc = K8sService(client, module)
     results = svc.retrieve(Mock(), pod_definition)
 
-    assert isinstance(results, dict)
-    assert results["changed"] is False
-    assert results["result"] == pod_definition
+    assert isinstance(results, ResourceInstance)
+    assert results.to_dict() == pod_definition
 
 
 def test_service_retrieve_no_existing_resource():
@@ -205,9 +204,7 @@ def test_service_retrieve_no_existing_resource():
     svc = K8sService(client, module)
     results = svc.retrieve(Mock(), pod_definition)
 
-    assert isinstance(results, dict)
-    assert results["changed"] is False
-    assert results["result"] == {}
+    assert results is None
 
 
 def test_create_project_request():