diff --git a/docs/data-sources/mirrormaker_replication_flow.md b/docs/data-sources/mirrormaker_replication_flow.md
index 13efd4408..2397b0b51 100644
--- a/docs/data-sources/mirrormaker_replication_flow.md
+++ b/docs/data-sources/mirrormaker_replication_flow.md
@@ -45,3 +45,4 @@ data "aiven_mirrormaker_replication_flow" "f1" {
 - `sync_group_offsets_interval_seconds` (Number) Frequency of consumer group offset sync. The default value is `1`.
 - `topics` (List of String) List of topics and/or regular expressions to replicate
 - `topics_blacklist` (List of String) List of topics and/or regular expressions to not replicate.
+- `exactly_once_delivery_enabled` (Boolean) Whether to enable exactly-once message delivery. We recommend you set this to `enabled` for new replications. The default value is `false`.
diff --git a/docs/resources/mirrormaker_replication_flow.md b/docs/resources/mirrormaker_replication_flow.md
index 71bf00fa5..1daa301dc 100644
--- a/docs/resources/mirrormaker_replication_flow.md
+++ b/docs/resources/mirrormaker_replication_flow.md
@@ -65,6 +65,7 @@ resource "aiven_mirrormaker_replication_flow" "f1" {
 - `timeouts` (Block, Optional) (see [below for nested schema](#nestedblock--timeouts))
 - `topics` (List of String) List of topics and/or regular expressions to replicate
 - `topics_blacklist` (List of String) List of topics and/or regular expressions to not replicate.
+- `exactly_once_delivery_enabled` (Boolean) Whether to enable exactly-once message delivery. We recommend you set this to `enabled` for new replications. The default value is `false`.
 
 ### Read-Only
 
diff --git a/internal/sdkprovider/service/kafka/mirrormaker_replication_flow.go b/internal/sdkprovider/service/kafka/mirrormaker_replication_flow.go
index cb85f66c9..0d3f02d2a 100644
--- a/internal/sdkprovider/service/kafka/mirrormaker_replication_flow.go
+++ b/internal/sdkprovider/service/kafka/mirrormaker_replication_flow.go
@@ -122,6 +122,14 @@ var aivenMirrorMakerReplicationFlowSchema = map[string]*schema.Schema{
 		ValidateFunc: validation.IntAtLeast(1),
 		Description:  "Replication factor, `>= 1`.",
 	},
+	"exactly_once_delivery_enabled": {
+		Type:     schema.TypeBool,
+		Optional: true,
+		Default:  false,
+		Description: userconfig.Desc(
+			"Whether to enable exactly-once message delivery. We recommend you set this to `enabled` for new replications.",
+		).DefaultValue(false).Build(),
+	},
 }
 
 func ResourceMirrorMakerReplicationFlow() *schema.Resource {
diff --git a/internal/sdkprovider/service/kafka/mirrormaker_replication_flow_test.go b/internal/sdkprovider/service/kafka/mirrormaker_replication_flow_test.go
index b3e183406..a715a3f05 100644
--- a/internal/sdkprovider/service/kafka/mirrormaker_replication_flow_test.go
+++ b/internal/sdkprovider/service/kafka/mirrormaker_replication_flow_test.go
@@ -43,10 +43,11 @@ func TestAccAivenMirrorMakerReplicationFlow_basic(t *testing.T) {
 					resource.TestCheckResourceAttr(resourceName, "topics_blacklist.1", ".*\\.replica"),
 					resource.TestCheckResourceAttr(resourceName, "topics_blacklist.2", "__.*"),
 					resource.TestCheckResourceAttr(resourceName, "config_properties_exclude.#", "0"),
+					resource.TestCheckResourceAttr(resourceName, "exactly_once_delivery_enabled", "true"),
 				),
 			},
 			{
-				Config: testAccMirrorMakerReplicationFlowResource(rName, `	
+				Config: testAccMirrorMakerReplicationFlowResource(rName, `
 				config_properties_exclude = [
 					"follower\\.replication\\.throttled\\.replicas",
 					"leader\\.replication\\.throttled\\.replicas",
@@ -226,6 +227,7 @@ resource "aiven_mirrormaker_replication_flow" "foo" {
   emit_heartbeats_enabled             = true
   emit_backward_heartbeats_enabled    = true
   offset_syncs_topic_location         = "source"
+  exactly_once_delivery_enabled       = true
 
   topics = [
     ".*",
@@ -241,10 +243,11 @@ resource "aiven_mirrormaker_replication_flow" "foo" {
 }
 
 data "aiven_mirrormaker_replication_flow" "flow" {
-  project        = data.aiven_project.foo.project
-  service_name   = aiven_kafka_mirrormaker.mm.service_name
-  source_cluster = aiven_mirrormaker_replication_flow.foo.source_cluster
-  target_cluster = aiven_mirrormaker_replication_flow.foo.target_cluster
+  project                       = data.aiven_project.foo.project
+  service_name                  = aiven_kafka_mirrormaker.mm.service_name
+  source_cluster                = aiven_mirrormaker_replication_flow.foo.source_cluster
+  target_cluster                = aiven_mirrormaker_replication_flow.foo.target_cluster
+  exactly_once_delivery_enabled = true
 
   depends_on = [aiven_mirrormaker_replication_flow.foo]
 }`, os.Getenv("AIVEN_PROJECT_NAME"), name, configExclude)
@@ -275,6 +278,10 @@ func testAccCheckAivenMirrorMakerReplicationFlowAttributes(n string) resource.Te
 			return fmt.Errorf("expected to get a correct enable from Aiven")
 		}
 
+		if a["exactly_once_delivery_enabled"] != "true" {
+			return fmt.Errorf("expected to get a correct exactly_once_delivery_enabled from Aiven")
+		}
+
 		return nil
 	}
 }
@@ -293,6 +300,7 @@ resource "aiven_mirrormaker_replication_flow" "foo" {
   emit_heartbeats_enabled             = true
   emit_backward_heartbeats_enabled    = true
   offset_syncs_topic_location         = "lol_offset"
+  exactly_once_delivery_enabled       = true
 }
 `
 	resource.ParallelTest(t, resource.TestCase{