diff --git a/changelogs/fragments/1904-route53_wait.yml b/changelogs/fragments/1904-route53_wait.yml
new file mode 100644
index 00000000000..f8f4568b43e
--- /dev/null
+++ b/changelogs/fragments/1904-route53_wait.yml
@@ -0,0 +1,2 @@
+trivial:
+  - "Add route53_wait module to community.aws.aws action group (https://github.com/ansible-collections/community.aws/pull/1904)."
diff --git a/meta/runtime.yml b/meta/runtime.yml
index 9f8cf5cd899..0c2578b02ed 100644
--- a/meta/runtime.yml
+++ b/meta/runtime.yml
@@ -191,6 +191,7 @@ action_groups:
   - redshift_cross_region_snapshots
   - redshift_info
   - redshift_subnet_group
+  - route53_wait
   - s3_bucket_notification
   - s3_bucket_info
   - s3_cors
diff --git a/plugins/modules/route53_wait.py b/plugins/modules/route53_wait.py
new file mode 100644
index 00000000000..45b199887fd
--- /dev/null
+++ b/plugins/modules/route53_wait.py
@@ -0,0 +1,185 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2023, Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+DOCUMENTATION = r"""
+---
+module: route53_wait
+version_added: 6.2.0
+short_description: wait for changes in Amazons Route 53 DNS service to propagate
+description:
+  - When using M(amazon.aws.route53) with I(wait=false), this module allows to wait for the
+    module's propagation to finish at a later point of time.
+options:
+  result:
+    aliases:
+      - results
+    description:
+      - The registered result of one or multiple M(amazon.aws.route53) invocations.
+    required: true
+    type: dict
+  wait_timeout:
+    description:
+      - How long to wait for the changes to be replicated, in seconds.
+      - This timeout will be used for every changed result in I(result).
+    default: 300
+    type: int
+  region:
+    description:
+      - This setting is ignored by the module. It is only present to make it possible to
+        have I(region) present in the module default group.
+    type: str
+author:
+  - Felix Fontein (@felixfontein)
+extends_documentation_fragment:
+  - amazon.aws.common.modules
+  - amazon.aws.boto3
+"""
+
+RETURN = r"""
+#
+"""
+
+EXAMPLES = r"""
+# Example when using a single route53 invocation:
+
+- name: Add new.foo.com as an A record with 3 IPs
+  amazon.aws.route53:
+    state: present
+    zone: foo.com
+    record: new.foo.com
+    type: A
+    ttl: 7200
+    value:
+      - 1.1.1.1
+      - 2.2.2.2
+      - 3.3.3.3
+  register: module_result
+
+# do something else
+
+- name: Wait for the changes of the above route53 invocation to propagate
+  community.aws.route53_wait:
+    result: "{{ module_result }}"
+
+#########################################################################
+# Example when using a loop over amazon.aws.route53:
+
+- name: Add various A records
+  amazon.aws.route53:
+    state: present
+    zone: foo.com
+    record: "{{ item.record }}"
+    type: A
+    ttl: 300
+    value: "{{ item.value }}"
+  loop:
+    - record: new.foo.com
+      value: 1.1.1.1
+    - record: foo.foo.com
+      value: 2.2.2.2
+    - record: bar.foo.com
+      value:
+        - 3.3.3.3
+        - 4.4.4.4
+  register: module_results
+
+# do something else
+
+- name: Wait for the changes of the above three route53 invocations to propagate
+  community.aws.route53_wait:
+    results: "{{ module_results }}"
+"""
+
+try:
+    import botocore
+except ImportError:
+    pass  # Handled by AnsibleAWSModule
+
+from ansible.module_utils._text import to_native
+
+from ansible_collections.amazon.aws.plugins.module_utils.waiters import get_waiter
+
+from ansible_collections.community.aws.plugins.module_utils.modules import AnsibleCommunityAWSModule as AnsibleAWSModule
+
+WAIT_RETRY = 5  # how many seconds to wait between propagation status polls
+
+
+def detect_task_results(results):
+    if "results" in results:
+        # This must be the registered result of a loop of route53 tasks
+        for key in ("changed", "msg", "skipped"):
+            if key not in results:
+                raise ValueError(f"missing {key} key")
+        if not isinstance(results["results"], list):
+            raise ValueError("results is present, but not a list")
+        for index, result in enumerate(results["results"]):
+            if not isinstance(result, dict):
+                raise ValueError(f"result {index + 1} is not a dictionary")
+            for key in ("changed", "failed", "ansible_loop_var", "invocation"):
+                if key not in result:
+                    raise ValueError(f"missing {key} key for result {index + 1}")
+            yield f" for result #{index + 1}", result
+        return
+    # This must be a single route53 task
+    for key in ("changed", "failed"):
+        if key not in results:
+            raise ValueError(f"missing {key} key")
+    yield "", results
+
+
+def main():
+    argument_spec = dict(
+        result=dict(type="dict", required=True, aliases=["results"]),
+        wait_timeout=dict(type="int", default=300),
+        region=dict(type="str"),  # ignored
+    )
+
+    module = AnsibleAWSModule(argument_spec=argument_spec, supports_check_mode=True)
+
+    result_in = module.params["result"]
+    wait_timeout_in = module.params.get("wait_timeout")
+
+    changed_results = []
+    try:
+        for id, result in detect_task_results(result_in):
+            if result.get("wait_id"):
+                changed_results.append((id, result["wait_id"]))
+    except ValueError as exc:
+        module.fail_json(
+            msg=f"The value passed as result does not seem to be a registered route53 result: {to_native(exc)}"
+        )
+
+    # connect to the route53 endpoint
+    try:
+        route53 = module.client("route53")
+    except botocore.exceptions.HTTPClientError as e:
+        module.fail_json_aws(e, msg="Failed to connect to AWS")
+
+    for what, wait_id in changed_results:
+        try:
+            waiter = get_waiter(route53, "resource_record_sets_changed")
+            waiter.wait(
+                Id=wait_id,
+                WaiterConfig=dict(
+                    Delay=WAIT_RETRY,
+                    MaxAttempts=wait_timeout_in // WAIT_RETRY,
+                ),
+            )
+        except botocore.exceptions.WaiterError as e:
+            module.fail_json_aws(e, msg=f"Timeout waiting for resource records changes{what} to be applied")
+        except (
+            botocore.exceptions.BotoCoreError,
+            botocore.exceptions.ClientError,
+        ) as e:  # pylint: disable=duplicate-except
+            module.fail_json_aws(e, msg="Failed to update records")
+        except Exception as e:
+            module.fail_json(msg=f"Unhandled exception. ({to_native(e)})")
+
+    module.exit_json(changed=False)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/integration/targets/route53_wait/aliases b/tests/integration/targets/route53_wait/aliases
new file mode 100644
index 00000000000..4ef4b2067d0
--- /dev/null
+++ b/tests/integration/targets/route53_wait/aliases
@@ -0,0 +1 @@
+cloud/aws
diff --git a/tests/integration/targets/route53_wait/tasks/main.yml b/tests/integration/targets/route53_wait/tasks/main.yml
new file mode 100644
index 00000000000..f9df05f5c00
--- /dev/null
+++ b/tests/integration/targets/route53_wait/tasks/main.yml
@@ -0,0 +1,245 @@
+---
+# tasks file for route53_wait integration tests
+
+- set_fact:
+    zone_one: '{{ resource_prefix | replace("-", "") }}.one.ansible.test.'
+- debug:
+    msg: Set zone {{ zone_one }}
+
+- name: Test basics (new zone, A and AAAA records)
+  module_defaults:
+    group/aws:
+      aws_access_key: '{{ aws_access_key }}'
+      aws_secret_key: '{{ aws_secret_key }}'
+      security_token: '{{ security_token | default(omit) }}'
+      region: '{{ aws_region }}'
+    amazon.aws.route53:
+      # Route53 is explicitly a global service
+      region:
+  block:
+  - name: create VPC
+    ec2_vpc_net:
+      cidr_block: 192.0.2.0/24
+      name: '{{ resource_prefix }}_vpc'
+      state: present
+    register: vpc
+
+  - name: Create a zone
+    route53_zone:
+      zone: '{{ zone_one }}'
+      comment: Created in Ansible test {{ resource_prefix }}
+      tags:
+        TestTag: '{{ resource_prefix }}.z1'
+    register: z1
+
+  - name: Create A record (check mode)
+    route53:
+      state: present
+      hosted_zone_id: '{{ z1.zone_id }}'
+      record: test.{{ zone_one }}
+      overwrite: true
+      type: A
+      value: 192.0.2.1
+      wait: false
+    register: result
+    check_mode: true
+  - assert:
+      that:
+      - result is not failed
+      - result is changed
+      - "'wait_id' in result"
+      - result.wait_id is none
+
+  - name: Wait for A record to propagate (should do nothing)
+    route53_wait:
+      result: '{{ result }}'
+
+  - name: Create A record
+    route53:
+      state: present
+      hosted_zone_id: '{{ z1.zone_id }}'
+      record: test.{{ zone_one }}
+      overwrite: true
+      type: A
+      value: 192.0.2.1
+      wait: false
+    register: result
+  - assert:
+      that:
+      - result is not failed
+      - result is changed
+      - "'wait_id' in result"
+      - result.wait_id is string
+
+  - name: Wait for A record to propagate
+    route53_wait:
+      result: '{{ result }}'
+
+  - name: Create A record (idempotent)
+    route53:
+      state: present
+      hosted_zone_id: '{{ z1.zone_id }}'
+      record: test.{{ zone_one }}
+      overwrite: true
+      type: A
+      value: 192.0.2.1
+      wait: false
+    register: result
+  - assert:
+      that:
+      - result is not failed
+      - result is not changed
+      - "'wait_id' not in result"
+
+  - name: Wait for A record to propagate (should do nothing)
+    route53_wait:
+      result: '{{ result }}'
+
+  - name: Create A records
+    route53:
+      state: present
+      hosted_zone_id: '{{ z1.zone_id }}'
+      record: '{{ item.record }}'
+      overwrite: true
+      type: A
+      value: '{{ item.value }}'
+      wait: false
+    loop:
+      - record: test-1.{{ zone_one }}
+        value: 192.0.2.1
+      - record: test-2.{{ zone_one }}
+        value: 192.0.2.2
+      - record: test-3.{{ zone_one }}
+        value: 192.0.2.3
+    register: results
+  - assert:
+      that:
+      - results is not failed
+      - results is changed
+      - results.results | length == 3
+      - results.results[0] is changed
+      - results.results[1] is changed
+      - results.results[2] is changed
+
+  - name: Wait for A records to propagate
+    route53_wait:
+      results: '{{ results }}'
+
+  - name: Create A records (idempotent)
+    route53:
+      state: present
+      hosted_zone_id: '{{ z1.zone_id }}'
+      record: '{{ item.record }}'
+      overwrite: true
+      type: A
+      value: '{{ item.value }}'
+      wait: false
+    loop:
+      - record: test-1.{{ zone_one }}
+        value: 192.0.2.1
+      - record: test-2.{{ zone_one }}
+        value: 192.0.2.2
+      - record: test-3.{{ zone_one }}
+        value: 192.0.2.3
+    register: results
+  - assert:
+      that:
+      - results is not failed
+      - results is not changed
+      - results.results | length == 3
+      - results.results[0] is not changed
+      - results.results[1] is not changed
+      - results.results[2] is not changed
+
+  - name: Wait for A records to propagate (should do nothing)
+    route53_wait:
+      results: '{{ results }}'
+
+  - name: Update some A records
+    route53:
+      state: present
+      hosted_zone_id: '{{ z1.zone_id }}'
+      record: '{{ item.record }}'
+      overwrite: true
+      type: A
+      value: '{{ item.value }}'
+      wait: false
+    loop:
+      - record: test-1.{{ zone_one }}
+        value: 192.0.2.1
+      - record: test-2.{{ zone_one }}
+        value: 192.0.2.4
+      - record: test-3.{{ zone_one }}
+        value: 192.0.2.3
+    register: results
+  - assert:
+      that:
+      - results is not failed
+      - results is changed
+      - results.results | length == 3
+      - results.results[0] is not changed
+      - results.results[1] is changed
+      - results.results[2] is not changed
+
+  - name: Wait for A records to propagate
+    route53_wait:
+      results: '{{ results }}'
+
+#Cleanup------------------------------------------------------
+
+  always:
+
+  - route53_info:
+      query: record_sets
+      hosted_zone_id: '{{ z1.zone_id }}'
+    register: z1_records
+
+  - name: Loop over A/AAAA/CNAME records and delete them
+    route53:
+      state: absent
+      zone: '{{ zone_one }}'
+      record: '{{ item.Name }}'
+      type: '{{ item.Type }}'
+      value: '{{ item.ResourceRecords | map(attribute="Value") | join(",") }}'
+      weight: '{{ item.Weight | default(omit) }}'
+      identifier: '{{ item.SetIdentifier }}'
+      region: '{{ omit }}'
+    ignore_errors: true
+    loop: '{{ z1_records.ResourceRecordSets | selectattr("Type", "in", ["A", "AAAA",
+      "CNAME", "CAA"]) | list }}'
+    when:
+    - '"ResourceRecords" in item'
+    - '"SetIdentifier" in item'
+
+  - name: Loop over A/AAAA/CNAME records and delete them
+    route53:
+      state: absent
+      zone: '{{ zone_one }}'
+      record: '{{ item.Name }}'
+      type: '{{ item.Type }}'
+      value: '{{ item.ResourceRecords | map(attribute="Value") | join(",") }}'
+    ignore_errors: true
+    loop: '{{ z1_records.ResourceRecordSets | selectattr("Type", "in", ["A", "AAAA",
+      "CNAME", "CAA"]) | list }}'
+    when:
+    - '"ResourceRecords" in item'
+
+  - name: Delete test zone one {{ zone_one }}
+    route53_zone:
+      state: absent
+      zone: '{{ zone_one }}'
+    register: delete_one
+    ignore_errors: true
+    retries: 10
+    until: delete_one is not failed
+
+  - name: destroy VPC
+    ec2_vpc_net:
+      cidr_block: 192.0.2.0/24
+      name: '{{ resource_prefix }}_vpc'
+      state: absent
+    register: remove_vpc
+    retries: 10
+    delay: 5
+    until: remove_vpc is success
+    ignore_errors: true
diff --git a/tests/unit/plugins/modules/test_route53_wait.py b/tests/unit/plugins/modules/test_route53_wait.py
new file mode 100644
index 00000000000..ac9cffb8fc5
--- /dev/null
+++ b/tests/unit/plugins/modules/test_route53_wait.py
@@ -0,0 +1,243 @@
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2023, Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+import pytest
+
+from ansible_collections.community.aws.plugins.modules.route53_wait import (
+    detect_task_results,
+)
+
+
+_SINGLE_RESULT_SUCCESS = {
+    "changed": True,
+    "diff": {},
+    "failed": False,
+    "wait_id": None,
+}
+
+_SINGLE_RESULT_FAILED = {
+    "changed": False,
+    "failed": True,
+    "msg": "value of type must be one of: A, AAAA, CAA, CNAME, MX, NS, PTR, SOA, SPF, SRV, TXT, got: bar",
+}
+
+_MULTI_RESULT_SUCCESS = {
+    "ansible_loop_var": "item",
+    "changed": True,
+    "diff": {},
+    "failed": False,
+    "invocation": {
+        "module_args": {
+            "access_key": "asdf",
+            "alias": None,
+            "alias_evaluate_target_health": False,
+            "alias_hosted_zone_id": None,
+            "aws_access_key": "asdf",
+            "aws_ca_bundle": None,
+            "aws_config": None,
+            "aws_secret_key": "VALUE_SPECIFIED_IN_NO_LOG_PARAMETER",
+            "debug_botocore_endpoint_logs": False,
+            "endpoint_url": None,
+            "failover": None,
+            "geo_location": None,
+            "health_check": None,
+            "hosted_zone_id": None,
+            "identifier": None,
+            "overwrite": True,
+            "private_zone": False,
+            "profile": None,
+            "record": "foo.example.org",
+            "region": None,
+            "retry_interval": 500,
+            "secret_key": "VALUE_SPECIFIED_IN_NO_LOG_PARAMETER",
+            "session_token": None,
+            "state": "present",
+            "ttl": 300,
+            "type": "TXT",
+            "validate_certs": True,
+            "value": ["foo"],
+            "vpc_id": None,
+            "wait": False,
+            "wait_timeout": 300,
+            "weight": None,
+            "zone": "example.org",
+        },
+    },
+    "item": {"record": "foo.example.org", "value": "foo"},
+    "wait_id": None,
+}
+
+_MULTI_RESULT_FAILED = {
+    "ansible_loop_var": "item",
+    "changed": False,
+    "failed": True,
+    "invocation": {
+        "module_args": {
+            "access_key": "asdf",
+            "alias": None,
+            "alias_evaluate_target_health": False,
+            "alias_hosted_zone_id": None,
+            "aws_access_key": "asdf",
+            "aws_ca_bundle": None,
+            "aws_config": None,
+            "aws_secret_key": "VALUE_SPECIFIED_IN_NO_LOG_PARAMETER",
+            "debug_botocore_endpoint_logs": False,
+            "endpoint_url": None,
+            "failover": None,
+            "geo_location": None,
+            "health_check": None,
+            "hosted_zone_id": None,
+            "identifier": None,
+            "overwrite": True,
+            "private_zone": False,
+            "profile": None,
+            "record": "foo.example.org",
+            "region": None,
+            "retry_interval": 500,
+            "secret_key": "VALUE_SPECIFIED_IN_NO_LOG_PARAMETER",
+            "session_token": None,
+            "state": "present",
+            "ttl": 300,
+            "type": "bar",
+            "validate_certs": True,
+            "value": ["foo"],
+            "vpc_id": None,
+            "wait": False,
+            "wait_timeout": 300,
+            "weight": None,
+            "zone": "example.org",
+        },
+    },
+    "item": {"record": "foo.example.org", "value": "foo"},
+    "msg": "value of type must be one of: A, AAAA, CAA, CNAME, MX, NS, PTR, SOA, SPF, SRV, TXT, got: bar",
+}
+
+
+DETECT_TASK_RESULTS_DATA = [
+    [
+        _SINGLE_RESULT_SUCCESS,
+        [
+            (
+                "",
+                _SINGLE_RESULT_SUCCESS,
+            ),
+        ],
+    ],
+    [
+        {
+            "changed": True,
+            "msg": "All items completed",
+            "results": [
+                _MULTI_RESULT_SUCCESS,
+            ],
+            "skipped": False,
+        },
+        [
+            (
+                " for result #1",
+                _MULTI_RESULT_SUCCESS,
+            ),
+        ],
+    ],
+    [
+        _SINGLE_RESULT_FAILED,
+        [
+            (
+                "",
+                _SINGLE_RESULT_FAILED,
+            ),
+        ],
+    ],
+    [
+        {
+            "changed": False,
+            "failed": True,
+            "msg": "One or more items failed",
+            "results": [
+                _MULTI_RESULT_FAILED,
+            ],
+            "skipped": False,
+        },
+        [
+            (
+                " for result #1",
+                _MULTI_RESULT_FAILED,
+            ),
+        ],
+    ],
+]
+
+
+@pytest.mark.parametrize(
+    "input, expected",
+    DETECT_TASK_RESULTS_DATA,
+)
+def test_detect_task_results(input, expected):
+    assert list(detect_task_results(input)) == expected
+
+
+DETECT_TASK_RESULTS_FAIL_DATA = [
+    [
+        {},
+        "missing changed key",
+        [],
+    ],
+    [
+        {"changed": True},
+        "missing failed key",
+        [],
+    ],
+    [
+        {"results": None},
+        "missing changed key",
+        [],
+    ],
+    [
+        {"results": None, "changed": True, "msg": "foo"},
+        "missing skipped key",
+        [],
+    ],
+    [
+        {"results": None, "changed": True, "msg": "foo", "skipped": False},
+        "results is present, but not a list",
+        [],
+    ],
+    [
+        {"results": [None], "changed": True, "msg": "foo", "skipped": False},
+        "result 1 is not a dictionary",
+        [],
+    ],
+    [
+        {"results": [{}], "changed": True, "msg": "foo", "skipped": False},
+        "missing changed key for result 1",
+        [],
+    ],
+    [
+        {
+            "results": [{"changed": True, "failed": False, "ansible_loop_var": "item", "invocation": {}}, {}],
+            "changed": True,
+            "msg": "foo",
+            "skipped": False,
+        },
+        "missing changed key for result 2",
+        [(" for result #1", {"changed": True, "failed": False, "ansible_loop_var": "item", "invocation": {}})],
+    ],
+]
+
+
+@pytest.mark.parametrize(
+    "input, expected_exc, expected_result",
+    DETECT_TASK_RESULTS_FAIL_DATA,
+)
+def test_detect_task_fail_results(input, expected_exc, expected_result):
+    result = []
+    with pytest.raises(ValueError) as exc:
+        for res in detect_task_results(input):
+            result.append(res)
+
+    print(exc.value.args[0])
+    assert expected_exc == exc.value.args[0]
+    print(result)
+    assert expected_result == result