feat: mirrormaker exactly-once delivery

aiven · Nov 20, 2024 · 5e1f977 · 5e1f977
1 parent aaf3888
commit 5e1f977
Show file tree

Hide file tree

Showing 5 changed files with 24 additions and 5 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -21,6 +21,7 @@ nav_order: 1
   private networks
 - Add `aiven_thanos` datasource field `thanos_user_config.privatelink_access`: Allow access to selected service components
   through Privatelink
+- Add `exactly_once_delivery_enabled` field to `aiven_mirrormaker_replication_flow`: Whether to enable exactly-once message delivery for Mirrormaker replication flow
 
 ## [4.29.0] - 2024-11-14
 

diff --git 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
@@ -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
@@ -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
@@ -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{