feat: mirrormaker exactly-once delivery

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
 

docs/data-sources/mirrormaker_replication_flow.md

@@ -37,6 +37,7 @@ data "aiven_mirrormaker_replication_flow" "f1" {
 - `emit_backward_heartbeats_enabled` (Boolean) Whether to emit heartbeats to the direction opposite to the flow, i.e. to the source cluster. The default value is `false`.
 - `emit_heartbeats_enabled` (Boolean) Whether to emit heartbeats to the target cluster. The default value is `false`.
 - `enable` (Boolean) Enable of disable replication flows for a service.
+- `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`.
 - `id` (String) The ID of this resource.
 - `offset_syncs_topic_location` (String) Offset syncs topic location. The possible values are `source` and `target`.
 - `replication_factor` (Number) Replication factor, `>= 1`.

docs/resources/mirrormaker_replication_flow.md

@@ -59,6 +59,7 @@ resource "aiven_mirrormaker_replication_flow" "f1" {
 - `config_properties_exclude` (Set of String) List of topic configuration properties and/or regular expressions to not replicate. The properties that are not replicated by default are: `follower.replication.throttled.replicas`, `leader.replication.throttled.replicas`, `message.timestamp.difference.max.ms`, `message.timestamp.type`, `unclean.leader.election.enable`, and `min.insync.replicas`. Setting this overrides the defaults. For example, to enable replication for 'min.insync.replicas' and 'unclean.leader.election.enable' set this to: ["follower\\\\.replication\\\\.throttled\\\\.replicas", "leader\\\\.replication\\\\.throttled\\\\.replicas", "message\\\\.timestamp\\\\.difference\\\\.max\\\\.ms",  "message\\\\.timestamp\\\\.type"]
 - `emit_backward_heartbeats_enabled` (Boolean) Whether to emit heartbeats to the direction opposite to the flow, i.e. to the source cluster. The default value is `false`.
 - `emit_heartbeats_enabled` (Boolean) Whether to emit heartbeats to the target cluster. The default value is `false`.
+- `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`.
 - `replication_factor` (Number) Replication factor, `>= 1`.
 - `sync_group_offsets_enabled` (Boolean) Sync consumer group offsets. The default value is `false`.
 - `sync_group_offsets_interval_seconds` (Number) Frequency of consumer group offset sync. The default value is `1`.

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 {

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{