diff --git a/changelogs/fragments/65854-docker_container-wait-for-removal.yml b/changelogs/fragments/65854-docker_container-wait-for-removal.yml
new file mode 100644
index 00000000000..5c992a5edd3
--- /dev/null
+++ b/changelogs/fragments/65854-docker_container-wait-for-removal.yml
@@ -0,0 +1,2 @@
+bugfixes:
+- "docker_container - wait for removal of container if docker API returns early (https://github.com/ansible/ansible/issues/65811)."
diff --git a/lib/ansible/module_utils/docker/common.py b/lib/ansible/module_utils/docker/common.py
index 5b747952ec6..03307250d69 100644
--- a/lib/ansible/module_utils/docker/common.py
+++ b/lib/ansible/module_utils/docker/common.py
@@ -517,6 +517,17 @@ class AnsibleDockerClient(Client):
                         msg = 'Cannot %s with your configuration.' % (usg, )
                     self.fail(msg)
 
+    def get_container_by_id(self, container_id):
+        try:
+            self.log("Inspecting container Id %s" % container_id)
+            result = self.inspect_container(container=container_id)
+            self.log("Completed container inspection")
+            return result
+        except NotFound as dummy:
+            return None
+        except Exception as exc:
+            self.fail("Error inspecting container: %s" % exc)
+
     def get_container(self, name=None):
         '''
         Lookup a container and return the inspection results.
@@ -546,17 +557,10 @@ class AnsibleDockerClient(Client):
         except Exception as exc:
             self.fail("Error retrieving container list: %s" % exc)
 
-        if result is not None:
-            try:
-                self.log("Inspecting container Id %s" % result['Id'])
-                result = self.inspect_container(container=result['Id'])
-                self.log("Completed container inspection")
-            except NotFound as dummy:
-                return None
-            except Exception as exc:
-                self.fail("Error inspecting container: %s" % exc)
+        if result is None:
+            return None
 
-        return result
+        return self.get_container_by_id(result['Id'])
 
     def get_network(self, name=None, network_id=None):
         '''
diff --git a/lib/ansible/modules/cloud/docker/docker_container.py b/lib/ansible/modules/cloud/docker/docker_container.py
index b8a83f50dde..695d62f70da 100644
--- a/lib/ansible/modules/cloud/docker/docker_container.py
+++ b/lib/ansible/modules/cloud/docker/docker_container.py
@@ -1064,6 +1064,7 @@ import re
 import shlex
 import traceback
 from distutils.version import LooseVersion
+from time import sleep
 
 from ansible.module_utils.common.text.formatters import human_to_bytes
 from ansible.module_utils.docker.common import (
@@ -1953,6 +1954,12 @@ class Container(DockerBaseClass):
     def exists(self):
         return True if self.container else False
 
+    @property
+    def removing(self):
+        if self.container and self.container.get('State'):
+            return self.container['State'].get('Status') == 'removing'
+        return False
+
     @property
     def running(self):
         if self.container and self.container.get('State'):
@@ -2554,6 +2561,31 @@ class ContainerManager(DockerBaseClass):
             self.results['ansible_facts'] = {'docker_container': self.facts}
             self.results['container'] = self.facts
 
+    def wait_for_state(self, container_id, complete_states=None, wait_states=None, accept_removal=False):
+        delay = 1.0
+        while True:
+            # Inspect container
+            result = self.client.get_container_by_id(container_id)
+            if result is None:
+                if accept_removal:
+                    return
+                msg = 'Encontered vanished container while waiting for container {0}'
+                self.fail(msg.format(container_id))
+            # Check container state
+            state = result.get('State', {}).get('Status')
+            if complete_states is not None and state in complete_states:
+                return
+            if wait_states is not None and state not in wait_states:
+                msg = 'Encontered unexpected state "{1}" while waiting for container {0}'
+                self.fail(msg.format(container_id, state))
+            # Wait
+            sleep(delay)
+            # Exponential backoff, but never wait longer than 10 seconds
+            # (1.1**24 < 10, 1.1**25 > 10, so it will take 25 iterations
+            #  until the maximal 10 seconds delay is reached. By then, the
+            #  code will have slept for ~1.5 minutes.)
+            delay = min(delay * 1.1, 10)
+
     def present(self, state):
         container = self._get_container(self.parameters.name)
         was_running = container.running
@@ -2567,12 +2599,18 @@ class ContainerManager(DockerBaseClass):
         # image ID.
         image = self._get_image()
         self.log(image, pretty_print=True)
-        if not container.exists:
+        if not container.exists or container.removing:
             # New container
-            self.log('No container found')
+            if container.removing:
+                self.log('Found container in removal phase')
+            else:
+                self.log('No container found')
             if not self.parameters.image:
                 self.fail('Cannot create container when image is not specified!')
             self.diff_tracker.add('exists', parameter=True, active=False)
+            if container.removing and not self.check_mode:
+                # Wait for container to be removed before trying to create it
+                self.wait_for_state(container.Id, wait_states=['removing'], accept_removal=True)
             new_container = self.container_create(self.parameters.image, self.parameters.create_parameters)
             if new_container:
                 container = new_container
@@ -2598,6 +2636,8 @@ class ContainerManager(DockerBaseClass):
                 if container.running:
                     self.container_stop(container.Id)
                 self.container_remove(container.Id)
+                if not self.check_mode:
+                    self.wait_for_state(container.Id, wait_states=['removing'], accept_removal=True)
                 new_container = self.container_create(image_to_use, self.parameters.create_parameters)
                 if new_container:
                     container = new_container