diff --git a/compute/client_library/ingredients/disks/replication_disk_start.py b/compute/client_library/ingredients/disks/replication_disk_start.py
new file mode 100644
index 00000000000..69188dc2c76
--- /dev/null
+++ b/compute/client_library/ingredients/disks/replication_disk_start.py
@@ -0,0 +1,69 @@
+#  Copyright 2024 Google LLC
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+# This is an ingredient file. It is not meant to be run directly. Check the samples/snippets
+# folder for complete code samples that are ready to be used.
+# Disabling flake8 for the ingredients file, as it would fail F821 - undefined name check.
+# flake8: noqa
+
+from google.cloud import compute_v1
+
+
+# <INGREDIENT start_disk_replication>
+def start_disk_replication(
+    project_id: str,
+    primary_disk_location: str,
+    primary_disk_name: str,
+    secondary_disk_location: str,
+    secondary_disk_name: str,
+) -> bool:
+    """Starts the asynchronous replication of a primary disk to a secondary disk.
+    Args:
+        project_id (str): The ID of the Google Cloud project.
+        primary_disk_location (str): The location of the primary disk, either a zone or a region.
+        primary_disk_name (str): The name of the primary disk.
+        secondary_disk_location (str): The location of the secondary disk, either a zone or a region.
+        secondary_disk_name (str): The name of the secondary disk.
+    Returns:
+        bool: True if the replication was successfully started.
+    """
+    # Check if the primary disk location is a region or a zone.
+    if primary_disk_location[-1].isdigit():
+        region_client = compute_v1.RegionDisksClient()
+        request_resource = compute_v1.RegionDisksStartAsyncReplicationRequest(
+            async_secondary_disk=f"projects/{project_id}/regions/{secondary_disk_location}/disks/{secondary_disk_name}"
+        )
+        operation = region_client.start_async_replication(
+            project=project_id,
+            region=primary_disk_location,
+            disk=primary_disk_name,
+            region_disks_start_async_replication_request_resource=request_resource,
+        )
+    else:
+        client = compute_v1.DisksClient()
+        request_resource = compute_v1.DisksStartAsyncReplicationRequest(
+            async_secondary_disk=f"zones/{secondary_disk_location}/disks/{secondary_disk_name}"
+        )
+        operation = client.start_async_replication(
+            project=project_id,
+            zone=primary_disk_location,
+            disk=primary_disk_name,
+            disks_start_async_replication_request_resource=request_resource,
+        )
+    wait_for_extended_operation(operation, verbose_name="replication operation")
+    print(f"Replication for disk {primary_disk_name} started.")
+    return True
+
+
+# </INGREDIENT>
diff --git a/compute/client_library/ingredients/disks/replication_disk_stop.py b/compute/client_library/ingredients/disks/replication_disk_stop.py
new file mode 100644
index 00000000000..678370b2c16
--- /dev/null
+++ b/compute/client_library/ingredients/disks/replication_disk_stop.py
@@ -0,0 +1,53 @@
+#  Copyright 2024 Google LLC
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+# This is an ingredient file. It is not meant to be run directly. Check the samples/snippets
+# folder for complete code samples that are ready to be used.
+# Disabling flake8 for the ingredients file, as it would fail F821 - undefined name check.
+# flake8: noqa
+
+from google.cloud import compute_v1
+
+
+# <INGREDIENT stop_disk_replication>
+def stop_disk_replication(
+    project_id: str, primary_disk_location: str, primary_disk_name: str
+) -> bool:
+    """
+    Stops the asynchronous replication of a disk.
+    Args:
+        project_id (str): The ID of the Google Cloud project.
+        primary_disk_location (str): The location of the primary disk, either a zone or a region.
+        primary_disk_name (str): The name of the primary disk.
+    Returns:
+        bool: True if the replication was successfully stopped.
+    """
+    # Check if the primary disk is in a region or a zone
+    if primary_disk_location[-1].isdigit():
+        region_client = compute_v1.RegionDisksClient()
+        operation = region_client.stop_async_replication(
+            project=project_id, region=primary_disk_location, disk=primary_disk_name
+        )
+    else:
+        zone_client = compute_v1.DisksClient()
+        operation = zone_client.stop_async_replication(
+            project=project_id, zone=primary_disk_location, disk=primary_disk_name
+        )
+
+    wait_for_extended_operation(operation, verbose_name="replication operation")
+    print(f"Replication for disk {primary_disk_name} stopped.")
+    return True
+
+
+# </INGREDIENT>
diff --git a/compute/client_library/recipes/disks/replication_disk_start.py b/compute/client_library/recipes/disks/replication_disk_start.py
new file mode 100644
index 00000000000..f729bcec55f
--- /dev/null
+++ b/compute/client_library/recipes/disks/replication_disk_start.py
@@ -0,0 +1,23 @@
+#  Copyright 2024 Google LLC
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+# flake8: noqa
+
+# <REGION compute_disk_start_replication>
+# <IMPORTS/>
+
+# <INGREDIENT wait_for_extended_operation />
+
+# <INGREDIENT start_disk_replication />
+
+# </REGION compute_disk_start_replication>
diff --git a/compute/client_library/recipes/disks/replication_disk_stop.py b/compute/client_library/recipes/disks/replication_disk_stop.py
new file mode 100644
index 00000000000..b13e1a371eb
--- /dev/null
+++ b/compute/client_library/recipes/disks/replication_disk_stop.py
@@ -0,0 +1,23 @@
+#  Copyright 2024 Google LLC
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+# flake8: noqa
+
+# <REGION compute_disk_stop_replication>
+# <IMPORTS/>
+
+# <INGREDIENT wait_for_extended_operation />
+
+# <INGREDIENT stop_disk_replication />
+
+# </REGION compute_disk_stop_replication>
diff --git a/compute/client_library/snippets/disks/replication_disk_start.py b/compute/client_library/snippets/disks/replication_disk_start.py
new file mode 100644
index 00000000000..3a3f988b8dc
--- /dev/null
+++ b/compute/client_library/snippets/disks/replication_disk_start.py
@@ -0,0 +1,125 @@
+#  Copyright 2024 Google LLC
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+# flake8: noqa
+
+
+# This file is automatically generated. Please do not modify it directly.
+# Find the relevant recipe file in the samples/recipes or samples/ingredients
+# directory and apply your changes there.
+
+
+# [START compute_disk_start_replication]
+from __future__ import annotations
+
+import sys
+from typing import Any
+
+from google.api_core.extended_operation import ExtendedOperation
+from google.cloud import compute_v1
+
+
+def wait_for_extended_operation(
+    operation: ExtendedOperation, verbose_name: str = "operation", timeout: int = 300
+) -> Any:
+    """
+    Waits for the extended (long-running) operation to complete.
+
+    If the operation is successful, it will return its result.
+    If the operation ends with an error, an exception will be raised.
+    If there were any warnings during the execution of the operation
+    they will be printed to sys.stderr.
+
+    Args:
+        operation: a long-running operation you want to wait on.
+        verbose_name: (optional) a more verbose name of the operation,
+            used only during error and warning reporting.
+        timeout: how long (in seconds) to wait for operation to finish.
+            If None, wait indefinitely.
+
+    Returns:
+        Whatever the operation.result() returns.
+
+    Raises:
+        This method will raise the exception received from `operation.exception()`
+        or RuntimeError if there is no exception set, but there is an `error_code`
+        set for the `operation`.
+
+        In case of an operation taking longer than `timeout` seconds to complete,
+        a `concurrent.futures.TimeoutError` will be raised.
+    """
+    result = operation.result(timeout=timeout)
+
+    if operation.error_code:
+        print(
+            f"Error during {verbose_name}: [Code: {operation.error_code}]: {operation.error_message}",
+            file=sys.stderr,
+            flush=True,
+        )
+        print(f"Operation ID: {operation.name}", file=sys.stderr, flush=True)
+        raise operation.exception() or RuntimeError(operation.error_message)
+
+    if operation.warnings:
+        print(f"Warnings during {verbose_name}:\n", file=sys.stderr, flush=True)
+        for warning in operation.warnings:
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)
+
+    return result
+
+
+def start_disk_replication(
+    project_id: str,
+    primary_disk_location: str,
+    primary_disk_name: str,
+    secondary_disk_location: str,
+    secondary_disk_name: str,
+) -> bool:
+    """Starts the asynchronous replication of a primary disk to a secondary disk.
+    Args:
+        project_id (str): The ID of the Google Cloud project.
+        primary_disk_location (str): The location of the primary disk, either a zone or a region.
+        primary_disk_name (str): The name of the primary disk.
+        secondary_disk_location (str): The location of the secondary disk, either a zone or a region.
+        secondary_disk_name (str): The name of the secondary disk.
+    Returns:
+        bool: True if the replication was successfully started.
+    """
+    # Check if the primary disk location is a region or a zone.
+    if primary_disk_location[-1].isdigit():
+        region_client = compute_v1.RegionDisksClient()
+        request_resource = compute_v1.RegionDisksStartAsyncReplicationRequest(
+            async_secondary_disk=f"projects/{project_id}/regions/{secondary_disk_location}/disks/{secondary_disk_name}"
+        )
+        operation = region_client.start_async_replication(
+            project=project_id,
+            region=primary_disk_location,
+            disk=primary_disk_name,
+            region_disks_start_async_replication_request_resource=request_resource,
+        )
+    else:
+        client = compute_v1.DisksClient()
+        request_resource = compute_v1.DisksStartAsyncReplicationRequest(
+            async_secondary_disk=f"zones/{secondary_disk_location}/disks/{secondary_disk_name}"
+        )
+        operation = client.start_async_replication(
+            project=project_id,
+            zone=primary_disk_location,
+            disk=primary_disk_name,
+            disks_start_async_replication_request_resource=request_resource,
+        )
+    wait_for_extended_operation(operation, verbose_name="replication operation")
+    print(f"Replication for disk {primary_disk_name} started.")
+    return True
+
+
+# [END compute_disk_start_replication]
diff --git a/compute/client_library/snippets/disks/replication_disk_stop.py b/compute/client_library/snippets/disks/replication_disk_stop.py
new file mode 100644
index 00000000000..b2d89e4daed
--- /dev/null
+++ b/compute/client_library/snippets/disks/replication_disk_stop.py
@@ -0,0 +1,109 @@
+#  Copyright 2024 Google LLC
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+# flake8: noqa
+
+
+# This file is automatically generated. Please do not modify it directly.
+# Find the relevant recipe file in the samples/recipes or samples/ingredients
+# directory and apply your changes there.
+
+
+# [START compute_disk_stop_replication]
+from __future__ import annotations
+
+import sys
+from typing import Any
+
+from google.api_core.extended_operation import ExtendedOperation
+from google.cloud import compute_v1
+
+
+def wait_for_extended_operation(
+    operation: ExtendedOperation, verbose_name: str = "operation", timeout: int = 300
+) -> Any:
+    """
+    Waits for the extended (long-running) operation to complete.
+
+    If the operation is successful, it will return its result.
+    If the operation ends with an error, an exception will be raised.
+    If there were any warnings during the execution of the operation
+    they will be printed to sys.stderr.
+
+    Args:
+        operation: a long-running operation you want to wait on.
+        verbose_name: (optional) a more verbose name of the operation,
+            used only during error and warning reporting.
+        timeout: how long (in seconds) to wait for operation to finish.
+            If None, wait indefinitely.
+
+    Returns:
+        Whatever the operation.result() returns.
+
+    Raises:
+        This method will raise the exception received from `operation.exception()`
+        or RuntimeError if there is no exception set, but there is an `error_code`
+        set for the `operation`.
+
+        In case of an operation taking longer than `timeout` seconds to complete,
+        a `concurrent.futures.TimeoutError` will be raised.
+    """
+    result = operation.result(timeout=timeout)
+
+    if operation.error_code:
+        print(
+            f"Error during {verbose_name}: [Code: {operation.error_code}]: {operation.error_message}",
+            file=sys.stderr,
+            flush=True,
+        )
+        print(f"Operation ID: {operation.name}", file=sys.stderr, flush=True)
+        raise operation.exception() or RuntimeError(operation.error_message)
+
+    if operation.warnings:
+        print(f"Warnings during {verbose_name}:\n", file=sys.stderr, flush=True)
+        for warning in operation.warnings:
+            print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)
+
+    return result
+
+
+def stop_disk_replication(
+    project_id: str, primary_disk_location: str, primary_disk_name: str
+) -> bool:
+    """
+    Stops the asynchronous replication of a disk.
+    Args:
+        project_id (str): The ID of the Google Cloud project.
+        primary_disk_location (str): The location of the primary disk, either a zone or a region.
+        primary_disk_name (str): The name of the primary disk.
+    Returns:
+        bool: True if the replication was successfully stopped.
+    """
+    # Check if the primary disk is in a region or a zone
+    if primary_disk_location[-1].isdigit():
+        region_client = compute_v1.RegionDisksClient()
+        operation = region_client.stop_async_replication(
+            project=project_id, region=primary_disk_location, disk=primary_disk_name
+        )
+    else:
+        zone_client = compute_v1.DisksClient()
+        operation = zone_client.stop_async_replication(
+            project=project_id, zone=primary_disk_location, disk=primary_disk_name
+        )
+
+    wait_for_extended_operation(operation, verbose_name="replication operation")
+    print(f"Replication for disk {primary_disk_name} stopped.")
+    return True
+
+
+# [END compute_disk_stop_replication]
diff --git a/compute/client_library/snippets/tests/test_disks.py b/compute/client_library/snippets/tests/test_disks.py
index 11a639c0818..1e2b20f9eb0 100644
--- a/compute/client_library/snippets/tests/test_disks.py
+++ b/compute/client_library/snippets/tests/test_disks.py
@@ -35,6 +35,8 @@
 from ..disks.list import list_disks
 from ..disks.regional_create_from_source import create_regional_disk
 from ..disks.regional_delete import delete_regional_disk
+from ..disks.replication_disk_start import start_disk_replication
+from ..disks.replication_disk_stop import stop_disk_replication
 from ..disks.resize_disk import resize_disk
 from ..images.get import get_image_from_family
 from ..instances.create import create_instance, disk_from_image
@@ -443,3 +445,57 @@ def test_create_custom_secondary_disk(
     )
     assert disk.labels["secondary-disk-for-replication"] == "true"
     assert disk.labels["source-disk"] == test_empty_pd_balanced_disk.name
+
+
+def test_start_stop_region_replication(
+    autodelete_regional_blank_disk, autodelete_regional_disk_name
+):
+    create_secondary_region_disk(
+        autodelete_regional_blank_disk.name,
+        PROJECT,
+        REGION,
+        autodelete_regional_disk_name,
+        PROJECT,
+        REGION_SECONDARY,
+        DISK_SIZE,
+    )
+    assert start_disk_replication(
+        project_id=PROJECT,
+        primary_disk_location=REGION,
+        primary_disk_name=autodelete_regional_blank_disk.name,
+        secondary_disk_location=REGION_SECONDARY,
+        secondary_disk_name=autodelete_regional_disk_name,
+    )
+    assert stop_disk_replication(
+        project_id=PROJECT,
+        primary_disk_location=REGION,
+        primary_disk_name=autodelete_regional_blank_disk.name,
+    )
+    # Wait for the replication to stop
+    time.sleep(20)
+
+
+def test_start_stop_zone_replication(test_empty_pd_balanced_disk, autodelete_disk_name):
+    create_secondary_disk(
+        test_empty_pd_balanced_disk.name,
+        PROJECT,
+        ZONE_SECONDARY,
+        autodelete_disk_name,
+        PROJECT,
+        ZONE,
+        DISK_SIZE,
+    )
+    assert start_disk_replication(
+        project_id=PROJECT,
+        primary_disk_location=ZONE_SECONDARY,
+        primary_disk_name=test_empty_pd_balanced_disk.name,
+        secondary_disk_location=ZONE,
+        secondary_disk_name=autodelete_disk_name,
+    )
+    assert stop_disk_replication(
+        project_id=PROJECT,
+        primary_disk_location=ZONE_SECONDARY,
+        primary_disk_name=test_empty_pd_balanced_disk.name,
+    )
+    # Wait for the replication to stop
+    time.sleep(20)