diff --git a/changelogs/fragments/49-k8s-loop-flattening-and-continue_on_error.yaml b/changelogs/fragments/49-k8s-loop-flattening-and-continue_on_error.yaml
new file mode 100644
index 0000000000..73c8d3dd33
--- /dev/null
+++ b/changelogs/fragments/49-k8s-loop-flattening-and-continue_on_error.yaml
@@ -0,0 +1,4 @@
+---
+minor_changes:
+  - k8s - Handle list of definition for option `template` (https://github.com/ansible-collections/kubernetes.core/pull/49).
+  - k8s - `continue_on_error` option added (whether to continue on creation/deletion errors) (https://github.com/ansible-collections/kubernetes.core/pull/49).
diff --git a/molecule/default/tasks/template.yml b/molecule/default/tasks/template.yml
index 3e18b3bbcf..cb5b91ac81 100644
--- a/molecule/default/tasks/template.yml
+++ b/molecule/default/tasks/template.yml
@@ -38,9 +38,9 @@
             replicas: 1
             selector:
               matchLabels:
-                app: "{{ k8s_pod_name }}"
+                app: "{{ k8s_pod_name_one }}"
       vars:
-        k8s_pod_name: pod
+        k8s_pod_name_one: pod
         k8s_pod_namespace: "{{ template_namespace }}"
       register: r
       ignore_errors: yes
@@ -57,7 +57,7 @@
         src: "../templates/pod_template_one.j2"
         template: "pod_template_one.j2"
       vars:
-        k8s_pod_name: pod
+        k8s_pod_name_one: pod
         k8s_pod_namespace: "{{ template_namespace }}"
       register: r
       ignore_errors: yes
@@ -73,7 +73,7 @@
         template: "pod_template_one.j2"
         wait: yes
       vars:
-        k8s_pod_name: pod-1
+        k8s_pod_name_one: pod-1
         k8s_pod_namespace: "{{ template_namespace }}"
       register: r
 
@@ -88,7 +88,7 @@
           - default
         wait: yes
       vars:
-        k8s_pod_name: pod-2
+        k8s_pod_name_one: pod-2
         k8s_pod_namespace: "{{ template_namespace }}"
       register: r
       ignore_errors: True
@@ -97,7 +97,6 @@
       assert:
         that:
           - r is failed
-          - "'Error while reading template file' in r.msg"
 
     - name: Create pod using template (path parameter)
       kubernetes.core.k8s:
@@ -105,7 +104,7 @@
           path: "pod_template_one.j2"
         wait: yes
       vars:
-        k8s_pod_name: pod-3
+        k8s_pod_name_one: pod-3
         k8s_pod_namespace: "{{ template_namespace }}"
       register: r
 
@@ -122,7 +121,7 @@
           variable_end_string: ']]'
         wait: yes
       vars:
-        k8s_pod_name: pod-4
+        k8s_pod_name_two: pod-4
         k8s_pod_namespace: "[[ template_namespace ]]"
         ansible_python_interpreter: "[[ ansible_playbook_python ]]"
       register: r
@@ -138,8 +137,8 @@
           path: "pod_template_three.j2"
         wait: yes
       vars:
-        k8s_pod_name_one: pod-5
-        k8s_pod_name_two: pod-6
+        k8s_pod_name_three_one: pod-5
+        k8s_pod_name_three_two: pod-6
         k8s_pod_namespace: "{{ template_namespace }}"
       register: r
 
@@ -148,8 +147,100 @@
         that:
           - r is successful
 
+    - name: Create pods using list of template
+      kubernetes.core.k8s:
+        template:
+          - pod_template_one.j2
+          - path: "pod_template_two.j2"
+            variable_start_string: '[['
+            variable_end_string: ']]'
+          - path: "pod_template_three.j2"
+        wait: yes
+      vars:
+        k8s_pod_name_one: pod-7
+        k8s_pod_name_two: pod-8
+        k8s_pod_name_three_one: pod-9
+        k8s_pod_name_three_two: pod-10
+        k8s_pod_namespace: "template-test"
+      register: r
+
+    - name: Assert that pod creation succeeded using template
+      assert:
+        that:
+          - r is successful
+
+    # continue_on_error
+    - name: define variable for test
+      set_fact:
+        k8s_pod_name_one: pod-11
+        k8s_pod_bad_name: pod-12
+        k8s_pod_namespace: "{{ template_namespace }}"
+        k8s_pod_bad_namespace: "dummy-namespace-012345"
+
+    - name: delete pod if it exists
+      kubernetes.core.k8s:
+        template: pod_template_one.j2
+        wait: true
+        state: absent
+
+    - name: create pod on bad namespace ( continue_on_error set to default(false) )
+      kubernetes.core.k8s:
+        template:
+          - pod_with_bad_namespace.j2
+          - pod_template_one.j2
+      register: resource
+      ignore_errors: true
+
+    - name: validate that creation failed
+      assert:
+        that:
+          - resource is failed
+          - '"Failed to create object" in resource.msg'
+
+    - name: assert pod has not been created
+      kubernetes.core.k8s_info:
+        kind: "{{ item.kind }}"
+        namespace: "{{ item.namespace }}"
+        name: "{{ item.name }}"
+      with_items:
+        - kind: pod
+          namespace: "{{ k8s_pod_bad_namespace }}"
+          name: "{{ k8s_pod_bad_name }}"
+        - kind: pod
+          namespace: "{{ k8s_pod_name_one }}"
+          name: "{{ k8s_pod_namespace }}"
+      register: resource
+
+    - name: check that resources creation failed
+      assert:
+        that:
+          - '{{ resource.results[0].resources | length == 0 }}'
+          - '{{ resource.results[1].resources | length == 0 }}'
+
+    - name: create pod without namespace (continue_on_error = true)
+      kubernetes.core.k8s:
+        template:
+          - pod_with_bad_namespace.j2
+          - pod_template_one.j2
+        continue_on_error: true
+        wait: true
+      register: resource
+      ignore_errors: true
+
+    - name: validate that creation succeeded
+      assert:
+        that:
+          - resource is successful
+
+    - name: validate resource creation succeeded for some and failed for others
+      assert:
+        that:
+          - resource is successful
+          - resource.result.results | selectattr('changed') | length == 1
+          - resource.result.results | selectattr('error', 'defined') | length == 1
+
     - name: Remove Pod (Cleanup)
-      k8s:
+      kubernetes.core.k8s:
         api_version: v1
         kind: Pod
         name: "pod-{{ item }}"
@@ -157,11 +248,11 @@
         state: absent
         wait: yes
       ignore_errors: yes
-      loop: "{{ range(1, 7) | list }}"
+      loop: "{{ range(1, 12) | list }}"
 
   always:
     - name: Remove namespace (Cleanup)
-      k8s:
+      kubernetes.core.k8s:
         kind: Namespace
         name: "{{ template_namespace }}"
         state: absent
diff --git a/molecule/default/templates/pod_template_one.j2 b/molecule/default/templates/pod_template_one.j2
index bafb7d9f6b..66970977d6 100644
--- a/molecule/default/templates/pod_template_one.j2
+++ b/molecule/default/templates/pod_template_one.j2
@@ -2,8 +2,8 @@ apiVersion: v1
 kind: Pod
 metadata:
   labels:
-    app: "{{ k8s_pod_name }}"
-  name: '{{ k8s_pod_name }}'
+    app: "{{ k8s_pod_name_one }}"
+  name: '{{ k8s_pod_name_one }}'
   namespace: '{{ k8s_pod_namespace }}'
 spec:
   containers:
@@ -13,4 +13,4 @@ spec:
     - while true; do echo $(date); sleep 10; done
     image: python:3.7-alpine
     imagePullPolicy: Always
-    name: '{{ k8s_pod_name }}'
+    name: '{{ k8s_pod_name_one }}'
diff --git a/molecule/default/templates/pod_template_three.j2 b/molecule/default/templates/pod_template_three.j2
index 06e4686e75..6d6592f4d1 100644
--- a/molecule/default/templates/pod_template_three.j2
+++ b/molecule/default/templates/pod_template_three.j2
@@ -3,8 +3,8 @@ apiVersion: v1
 kind: Pod
 metadata:
   labels:
-    app: "{{ k8s_pod_name_one }}"
-  name: '{{ k8s_pod_name_one }}'
+    app: "{{ k8s_pod_name_three_one }}"
+  name: '{{ k8s_pod_name_three_one }}'
   namespace: '{{ k8s_pod_namespace }}'
 spec:
   containers:
@@ -14,15 +14,15 @@ spec:
     - while true; do echo $(date); sleep 10; done
     image: python:3.7-alpine
     imagePullPolicy: Always
-    name: '{{ k8s_pod_name_one }}'
+    name: '{{ k8s_pod_name_three_one }}'
 
 ---
 apiVersion: v1
 kind: Pod
 metadata:
   labels:
-    app: "{{ k8s_pod_name_two }}"
-  name: '{{ k8s_pod_name_two }}'
+    app: "{{ k8s_pod_name_three_two }}"
+  name: '{{ k8s_pod_name_three_two }}'
   namespace: '{{ k8s_pod_namespace }}'
 spec:
   containers:
@@ -32,4 +32,4 @@ spec:
     - while true; do echo $(date); sleep 10; done
     image: python:3.7-alpine
     imagePullPolicy: Always
-    name: '{{ k8s_pod_name_two }}'
+    name: '{{ k8s_pod_name_three_two }}'
diff --git a/molecule/default/templates/pod_template_two.j2 b/molecule/default/templates/pod_template_two.j2
index cef89bf156..76820d493c 100644
--- a/molecule/default/templates/pod_template_two.j2
+++ b/molecule/default/templates/pod_template_two.j2
@@ -2,8 +2,8 @@ apiVersion: v1
 kind: Pod
 metadata:
   labels:
-    app: '[[ k8s_pod_name ]]'
-  name: '[[ k8s_pod_name ]]'
+    app: '[[ k8s_pod_name_two ]]'
+  name: '[[ k8s_pod_name_two ]]'
   namespace: '[[ k8s_pod_namespace ]]'
 spec:
   containers:
@@ -13,4 +13,4 @@ spec:
     - while true; do echo $(date); sleep 10; done
     image: python:3.7-alpine
     imagePullPolicy: Always
-    name: '[[ k8s_pod_name ]]'
+    name: '[[ k8s_pod_name_two ]]'
diff --git a/molecule/default/templates/pod_with_bad_namespace.j2 b/molecule/default/templates/pod_with_bad_namespace.j2
new file mode 100644
index 0000000000..ce0ea80add
--- /dev/null
+++ b/molecule/default/templates/pod_with_bad_namespace.j2
@@ -0,0 +1,16 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  labels:
+    app: "{{ k8s_pod_bad_name }}"
+  name: '{{ k8s_pod_bad_name }}'
+  namespace: '{{ k8s_pod_bad_namespace }}'
+spec:
+  containers:
+  - args:
+    - /bin/sh
+    - -c
+    - while true; do echo $(date); sleep 10; done
+    image: python:3.7-alpine
+    imagePullPolicy: Always
+    name: '{{ k8s_pod_bad_name }}'
diff --git a/plugins/action/k8s_info.py b/plugins/action/k8s_info.py
index 41ccf265bf..f36cdb129c 100644
--- a/plugins/action/k8s_info.py
+++ b/plugins/action/k8s_info.py
@@ -14,7 +14,7 @@
 from ansible.config.manager import ensure_type
 from ansible.errors import AnsibleError, AnsibleFileNotFound, AnsibleAction, AnsibleActionFail
 from ansible.module_utils.parsing.convert_bool import boolean
-from ansible.module_utils.six import string_types
+from ansible.module_utils.six import string_types, iteritems
 from ansible.module_utils._text import to_text, to_bytes, to_native
 from ansible.plugins.action import ActionBase
 
@@ -64,25 +64,25 @@ def get_template_data(self, template_path):
         finally:
             self._loader.cleanup_tmp_file(b_tmp_source)
 
-    def load_template(self, template, new_module_args, task_vars):
-        # template is only supported by k8s module.
-        if self._task.action not in ('k8s', 'kubernetes.core.k8s', 'community.okd.k8s'):
-            raise AnsibleActionFail("'template' is only supported parameter for 'k8s' module.")
+    def get_template_args(self, template):
+        template_param = {
+            "newline_sequence": self.DEFAULT_NEWLINE_SEQUENCE,
+            "variable_start_string": None,
+            "variable_end_string": None,
+            "block_start_string": None,
+            "block_end_string": None,
+            "trim_blocks": True,
+            "lstrip_blocks": False
+        }
         if isinstance(template, string_types):
             # treat this as raw_params
-            template_path = template
-            newline_sequence = self.DEFAULT_NEWLINE_SEQUENCE
-            variable_start_string = None
-            variable_end_string = None
-            block_start_string = None
-            block_end_string = None
-            trim_blocks = True
-            lstrip_blocks = False
+            template_param['path'] = template
         elif isinstance(template, dict):
             template_args = template
             template_path = template_args.get('path', None)
-            if not template:
+            if not template_path:
                 raise AnsibleActionFail("Please specify path for template.")
+            template_param['path'] = template_path
 
             # Options type validation strings
             for s_type in ('newline_sequence', 'variable_start_string', 'variable_end_string', 'block_start_string',
@@ -92,22 +92,28 @@ def load_template(self, template, new_module_args, task_vars):
                     if value is not None and not isinstance(value, string_types):
                         raise AnsibleActionFail("%s is expected to be a string, but got %s instead" % (s_type, type(value)))
             try:
-                trim_blocks = boolean(template_args.get('trim_blocks', True), strict=False)
-                lstrip_blocks = boolean(template_args.get('lstrip_blocks', False), strict=False)
+                template_param.update({
+                    "trim_blocks": boolean(template_args.get('trim_blocks', True), strict=False),
+                    "lstrip_blocks": boolean(template_args.get('lstrip_blocks', False), strict=False)
+                })
             except TypeError as e:
                 raise AnsibleActionFail(to_native(e))
 
-            newline_sequence = template_args.get('newline_sequence', self.DEFAULT_NEWLINE_SEQUENCE)
-            variable_start_string = template_args.get('variable_start_string', None)
-            variable_end_string = template_args.get('variable_end_string', None)
-            block_start_string = template_args.get('block_start_string', None)
-            block_end_string = template_args.get('block_end_string', None)
+            template_param.update({
+                "newline_sequence": template_args.get('newline_sequence', self.DEFAULT_NEWLINE_SEQUENCE),
+                "variable_start_string": template_args.get('variable_start_string', None),
+                "variable_end_string": template_args.get('variable_end_string', None),
+                "block_start_string": template_args.get('block_start_string', None),
+                "block_end_string": template_args.get('block_end_string', None)
+            })
         else:
             raise AnsibleActionFail("Error while reading template file - "
                                     "a string or dict for template expected, but got %s instead" % type(template))
+        return template_param
 
+    def import_jinja2_lstrip(self, templates):
         # Option `lstrip_blocks' was added in Jinja2 version 2.7.
-        if lstrip_blocks:
+        if any([tmp['lstrip_blocks'] for tmp in templates]):
             try:
                 import jinja2.defaults
             except ImportError:
@@ -118,39 +124,60 @@ def load_template(self, template, new_module_args, task_vars):
             except AttributeError:
                 raise AnsibleError("Option `lstrip_blocks' is only available in Jinja2 versions >=2.7")
 
+    def load_template(self, template, new_module_args, task_vars):
+        # template is only supported by k8s module.
+        if self._task.action not in ('k8s', 'kubernetes.core.k8s', 'community.okd.k8s'):
+            raise AnsibleActionFail("'template' is only supported parameter for 'k8s' module.")
+
+        template_params = []
+        if isinstance(template, string_types) or isinstance(template, dict):
+            template_params.append(self.get_template_args(template))
+        elif isinstance(template, list):
+            for element in template:
+                template_params.append(self.get_template_args(element))
+        else:
+            raise AnsibleActionFail("Error while reading template file - "
+                                    "a string or dict for template expected, but got %s instead" % type(template))
+
+        self.import_jinja2_lstrip(template_params)
+
         wrong_sequences = ["\\n", "\\r", "\\r\\n"]
         allowed_sequences = ["\n", "\r", "\r\n"]
 
-        # We need to convert unescaped sequences to proper escaped sequences for Jinja2
-        if newline_sequence in wrong_sequences:
-            newline_sequence = allowed_sequences[wrong_sequences.index(newline_sequence)]
-        elif newline_sequence not in allowed_sequences:
-            raise AnsibleActionFail("newline_sequence needs to be one of: \n, \r or \r\n")
-
-        # template the source data locally & get ready to transfer
-        with self.get_template_data(template_path) as template_data:
-            # add ansible 'template' vars
-            temp_vars = task_vars.copy()
-            old_vars = self._templar.available_variables
-
-            self._templar.environment.newline_sequence = newline_sequence
-            if block_start_string is not None:
-                self._templar.environment.block_start_string = block_start_string
-            if block_end_string is not None:
-                self._templar.environment.block_end_string = block_end_string
-            if variable_start_string is not None:
-                self._templar.environment.variable_start_string = variable_start_string
-            if variable_end_string is not None:
-                self._templar.environment.variable_end_string = variable_end_string
-            self._templar.environment.trim_blocks = trim_blocks
-            self._templar.environment.lstrip_blocks = lstrip_blocks
-            self._templar.available_variables = temp_vars
-            resultant = self._templar.do_template(template_data, preserve_trailing_newlines=True, escape_backslashes=False)
-            self._templar.available_variables = old_vars
-            resource_definition = self._task.args.get('definition', None)
-            if not resource_definition:
-                new_module_args.pop('template')
-            new_module_args['definition'] = resultant
+        result_template = []
+        old_vars = self._templar.available_variables
+
+        default_environment = {}
+        for key in ("newline_sequence", "variable_start_string", "variable_end_string",
+                    "block_start_string", "block_end_string", "trim_blocks", "lstrip_blocks"):
+            if hasattr(self._templar.environment, key):
+                default_environment[key] = getattr(self._templar.environment, key)
+        for template_item in template_params:
+            # We need to convert unescaped sequences to proper escaped sequences for Jinja2
+            newline_sequence = template_item['newline_sequence']
+            if newline_sequence in wrong_sequences:
+                template_item['newline_sequence'] = allowed_sequences[wrong_sequences.index(newline_sequence)]
+            elif newline_sequence not in allowed_sequences:
+                raise AnsibleActionFail("newline_sequence needs to be one of: \n, \r or \r\n")
+
+            # template the source data locally & get ready to transfer
+            with self.get_template_data(template_item['path']) as template_data:
+                # add ansible 'template' vars
+                temp_vars = copy.deepcopy(task_vars)
+                for key, value in iteritems(template_item):
+                    if hasattr(self._templar.environment, key):
+                        if value is not None:
+                            setattr(self._templar.environment, key, value)
+                        else:
+                            setattr(self._templar.environment, key, default_environment.get(key))
+                self._templar.available_variables = temp_vars
+                result = self._templar.do_template(template_data, preserve_trailing_newlines=True, escape_backslashes=False)
+                result_template.append(result)
+        self._templar.available_variables = old_vars
+        resource_definition = self._task.args.get('definition', None)
+        if not resource_definition:
+            new_module_args.pop('template')
+        new_module_args['definition'] = result_template
 
     def run(self, tmp=None, task_vars=None):
         ''' handler for k8s options '''
diff --git a/plugins/module_utils/common.py b/plugins/module_utils/common.py
index b872fb3325..4dc8d6bdb3 100644
--- a/plugins/module_utils/common.py
+++ b/plugins/module_utils/common.py
@@ -429,7 +429,6 @@ def set_resource_definitions(self, module):
         resource_definition = module.params.get('resource_definition')
 
         self.resource_definitions = []
-
         if resource_definition:
             if isinstance(resource_definition, string_types):
                 try:
@@ -437,7 +436,14 @@ def set_resource_definitions(self, module):
                 except (IOError, yaml.YAMLError) as exc:
                     self.fail(msg="Error loading resource_definition: {0}".format(exc))
             elif isinstance(resource_definition, list):
-                self.resource_definitions = resource_definition
+                for resource in resource_definition:
+                    if isinstance(resource, string_types):
+                        yaml_data = yaml.safe_load_all(resource)
+                        for item in yaml_data:
+                            if item is not None:
+                                self.resource_definitions.append(item)
+                    else:
+                        self.resource_definitions.append(resource)
             else:
                 self.resource_definitions = [resource_definition]
 
@@ -560,15 +566,20 @@ def perform_action(self, resource, definition):
         state = self.params.get('state', None)
         force = self.params.get('force', False)
         name = definition['metadata'].get('name')
+        origin_name = definition['metadata'].get('name')
         namespace = definition['metadata'].get('namespace')
         existing = None
         wait = self.params.get('wait')
         wait_sleep = self.params.get('wait_sleep')
         wait_timeout = self.params.get('wait_timeout')
         wait_condition = None
+        continue_on_error = self.params.get('continue_on_error')
         if self.params.get('wait_condition') and self.params['wait_condition'].get('type'):
             wait_condition = self.params['wait_condition']
 
+        def build_error_msg(kind, name, msg):
+            return "%s %s: %s" % (kind, name, msg)
+
         self.remove_aliases()
 
         try:
@@ -590,14 +601,26 @@ def perform_action(self, resource, definition):
         except ForbiddenError as exc:
             if definition['kind'] in ['Project', 'ProjectRequest'] and state != 'absent':
                 return self.create_project_request(definition)
-            self.fail_json(msg='Failed to retrieve requested object: {0}'.format(exc.body),
-                           error=exc.status, status=exc.status, reason=exc.reason)
+            msg = 'Failed to retrieve requested object: {0}'.format(exc.body)
+            if continue_on_error:
+                result['error'] = dict(msg=build_error_msg(definition['kind'], origin_name, msg), error=exc.status, status=exc.status, reason=exc.reason)
+                return result
+            else:
+                self.fail_json(msg=build_error_msg(definition['kind'], origin_name, msg), error=exc.status, status=exc.status, reason=exc.reason)
         except DynamicApiError as exc:
-            self.fail_json(msg='Failed to retrieve requested object: {0}'.format(exc.body),
-                           error=exc.status, status=exc.status, reason=exc.reason)
+            msg = 'Failed to retrieve requested object: {0}'.format(exc.body)
+            if continue_on_error:
+                result['error'] = dict(msg=build_error_msg(definition['kind'], origin_name, msg), error=exc.status, status=exc.status, reason=exc.reason)
+                return result
+            else:
+                self.fail_json(msg=build_error_msg(definition['kind'], origin_name, msg), error=exc.status, status=exc.status, reason=exc.reason)
         except ValueError as value_exc:
-            self.fail_json(msg='Failed to retrieve requested object: {0}'.format(to_native(value_exc)),
-                           error='', status='', reason='')
+            msg = 'Failed to retrieve requested object: {0}'.format(to_native(value_exc))
+            if continue_on_error:
+                result['error'] = dict(msg=build_error_msg(definition['kind'], origin_name, msg), error='', status='', reason='')
+                return result
+            else:
+                self.fail_json(msg=build_error_msg(definition['kind'], origin_name, msg), error='', status='', reason='')
 
         if state == 'absent':
             result['method'] = "delete"
@@ -619,13 +642,23 @@ def perform_action(self, resource, definition):
                         k8s_obj = resource.delete(**params)
                         result['result'] = k8s_obj.to_dict()
                     except DynamicApiError as exc:
-                        self.fail_json(msg="Failed to delete object: {0}".format(exc.body),
-                                       error=exc.status, status=exc.status, reason=exc.reason)
+                        msg = "Failed to delete object: {0}".format(exc.body)
+                        if continue_on_error:
+                            result['error'] = dict(msg=build_error_msg(definition['kind'], origin_name, msg),
+                                                   error=exc.status, status=exc.status, reason=exc.reason)
+                            return result
+                        else:
+                            self.fail_json(msg=build_error_msg(definition['kind'], origin_name, msg), error=exc.status, status=exc.status, reason=exc.reason)
                     if wait:
                         success, resource, duration = self.wait(resource, definition, wait_sleep, wait_timeout, 'absent')
                         result['duration'] = duration
                         if not success:
-                            self.fail_json(msg="Resource deletion timed out", **result)
+                            msg = "Resource deletion timed out"
+                            if continue_on_error:
+                                result['error'] = dict(msg=build_error_msg(definition['kind'], origin_name, msg), **result)
+                                return result
+                            else:
+                                self.fail_json(msg=build_error_msg(definition['kind'], origin_name, msg), **result)
                 return result
         else:
             if self.apply:
@@ -642,7 +675,12 @@ def perform_action(self, resource, definition):
                         msg = "Failed to apply object: {0}".format(exc.body)
                         if self.warnings:
                             msg += "\n" + "\n    ".join(self.warnings)
-                        self.fail_json(msg=msg, error=exc.status, status=exc.status, reason=exc.reason)
+                        if continue_on_error:
+                            result['error'] = dict(msg=build_error_msg(definition['kind'],
+                                                   origin_name, msg), error=exc.status, status=exc.status, reason=exc.reason)
+                            return result
+                        else:
+                            self.fail_json(msg=build_error_msg(definition['kind'], origin_name, msg), error=exc.status, status=exc.status, reason=exc.reason)
                 success = True
                 result['result'] = k8s_obj
                 if wait and not self.check_mode:
@@ -656,7 +694,12 @@ def perform_action(self, resource, definition):
                 result['diff'] = diffs
                 result['method'] = 'apply'
                 if not success:
-                    self.fail_json(msg="Resource apply timed out", **result)
+                    msg = "Resource apply timed out"
+                    if continue_on_error:
+                        result['error'] = dict(msg=build_error_msg(definition['kind'], origin_name, msg), **result)
+                        return result
+                    else:
+                        self.fail_json(msg=build_error_msg(definition['kind'], origin_name, msg), **result)
                 return result
 
             if not existing:
@@ -676,12 +719,21 @@ def perform_action(self, resource, definition):
                         msg = "Failed to create object: {0}".format(exc.body)
                         if self.warnings:
                             msg += "\n" + "\n    ".join(self.warnings)
-                        self.fail_json(msg=msg, error=exc.status, status=exc.status, reason=exc.reason)
+                        if continue_on_error:
+                            result['error'] = dict(msg=build_error_msg(definition['kind'], origin_name, msg),
+                                                   error=exc.status, status=exc.status, reason=exc.reason)
+                            return result
+                        else:
+                            self.fail_json(msg=build_error_msg(definition['kind'], origin_name, msg), error=exc.status, status=exc.status, reason=exc.reason)
                     except Exception as exc:
                         msg = "Failed to create object: {0}".format(exc)
                         if self.warnings:
                             msg += "\n" + "\n    ".join(self.warnings)
-                        self.fail_json(msg=msg, error='', status='', reason='')
+                        if continue_on_error:
+                            result['error'] = dict(msg=build_error_msg(definition['kind'], origin_name, msg), error='', status='', reason='')
+                            return result
+                        else:
+                            self.fail_json(msg=msg, error='', status='', reason='')
                 success = True
                 result['result'] = k8s_obj
                 if wait and not self.check_mode:
@@ -689,7 +741,12 @@ def perform_action(self, resource, definition):
                 result['changed'] = True
                 result['method'] = 'create'
                 if not success:
-                    self.fail_json(msg="Resource creation timed out", **result)
+                    msg = "Resource creation timed out"
+                    if continue_on_error:
+                        result['error'] = dict(msg=build_error_msg(definition['kind'], origin_name, msg), **result)
+                        return result
+                    else:
+                        self.fail_json(msg=msg, **result)
                 return result
 
             match = False
@@ -705,7 +762,12 @@ def perform_action(self, resource, definition):
                         msg = "Failed to replace object: {0}".format(exc.body)
                         if self.warnings:
                             msg += "\n" + "\n    ".join(self.warnings)
-                        self.fail_json(msg=msg, error=exc.status, status=exc.status, reason=exc.reason)
+                        if continue_on_error:
+                            result['error'] = dict(msg=build_error_msg(definition['kind'], origin_name, msg),
+                                                   error=exc.status, status=exc.status, reason=exc.reason)
+                            return result
+                        else:
+                            self.fail_json(msg=msg, error=exc.status, status=exc.status, reason=exc.reason)
                 match, diffs = self.diff_objects(existing.to_dict(), k8s_obj)
                 success = True
                 result['result'] = k8s_obj
@@ -716,7 +778,12 @@ def perform_action(self, resource, definition):
                 result['method'] = 'replace'
                 result['diff'] = diffs
                 if not success:
-                    self.fail_json(msg="Resource replacement timed out", **result)
+                    msg = "Resource replacement timed out"
+                    if continue_on_error:
+                        result['error'] = dict(msg=build_error_msg(definition['kind'], origin_name, msg), **result)
+                        return result
+                    else:
+                        self.fail_json(msg=msg, **result)
                 return result
 
             # Differences exist between the existing obj and requested params
@@ -733,7 +800,12 @@ def perform_action(self, resource, definition):
                         if not error:
                             break
                 if error:
-                    self.fail_json(**error)
+                    if continue_on_error:
+                        result['error'] = error
+                        result['error']['msg'] = build_error_msg(definition['kind'], origin_name, result['error'].get('msg'))
+                        return result
+                    else:
+                        self.fail_json(**error)
 
             success = True
             result['result'] = k8s_obj
@@ -745,7 +817,12 @@ def perform_action(self, resource, definition):
             result['diff'] = diffs
 
             if not success:
-                self.fail_json(msg="Resource update timed out", **result)
+                msg = "Resource update timed out"
+                if continue_on_error:
+                    result['error'] = dict(msg=build_error_msg(definition['kind'], origin_name, msg), **result)
+                    return result
+                else:
+                    self.fail_json(msg=msg, **result)
             return result
 
     def patch_resource(self, resource, definition, existing, name, namespace, merge_type=None):
diff --git a/plugins/modules/k8s.py b/plugins/modules/k8s.py
index 24883d0485..0672aa9301 100644
--- a/plugins/modules/k8s.py
+++ b/plugins/modules/k8s.py
@@ -2,6 +2,7 @@
 # -*- coding: utf-8 -*-
 
 # (c) 2018, Chris Houseknecht <@chouseknecht>
+# (c) 2021, Aubin Bikouo <@abikouo>
 # GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
 
 from __future__ import absolute_import, division, print_function
@@ -123,6 +124,13 @@
       When set to C(yes) leading spaces and tabs are stripped from the start of a line to a block.
       This functionality requires Jinja 2.7 or newer. Default value is false.'
     type: raw
+  continue_on_error:
+    description:
+    - Whether to continue on creation/deletion errors when multiple resources are defined.
+    - This has no effect on the validation step which is controlled by the C(validate.fail_on_error) parameter.
+    type: bool
+    default: False
+    version_added: 2.0.0
 
 requirements:
   - "python >= 2.7"
@@ -276,6 +284,10 @@
        returned: when C(wait) is true
        type: int
        sample: 48
+     error:
+       description: error while trying to create/delete the object.
+       returned: error
+       type: complex
 '''
 
 import copy
@@ -305,6 +317,7 @@ def argspec():
     argument_spec['apply'] = dict(type='bool', default=False)
     argument_spec['template'] = dict(type='raw', default=None)
     argument_spec['delete_options'] = dict(type='dict', default=None, options=copy.deepcopy(DELETE_OPTS_ARG_SPEC))
+    argument_spec['continue_on_error'] = dict(type='bool', default=False)
     return argument_spec