Add data infra consumers for prod account-identity

[matiit]

No description provided.

[uw-infra]

Terraform plan output for

Cluster: prod-aws
Module: pubsub/kafka-shared-msk-account-identity
Path: prod-aws/kafka-shared-msk/account-identity
Commit ID: 4b1c493a57a55b86c545f9edb5c96bf1c137f408

✅ Run Status: Ok, Run Summary: Plan: 60 to add, 0 to change, 0 to destroy.

Terraform used the selected providers to generate the following execution
plan. Resource actions are indicated with the following symbols:
  + create

Terraform will perform the following actions:

  # module.account_identity_di_addr_lookup.kafka_acl.group_acl["account-identity.di-kafka-source-address-lookup-analytics"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_addr_lookup"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-address-lookup-analytics"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_addr_lookup.kafka_acl.topic_acl["account-identity.address.lookup.analytics.v1"] will be created
  + resource "kafka_acl" "topic_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_addr_lookup"
      + id                           = (known after apply)
      + resource_name                = "account-identity.address.lookup.analytics.v1"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Topic"
    }

  # module.account_identity_di_addr_lookup.kafka_quota.quota will be created
  + resource "kafka_quota" "quota" {
      + config      = {
          + "consumer_byte_rate" = 5242880
          + "producer_byte_rate" = 5242880
          + "request_percentage" = 100
        }
      + entity_name = "CN=account-platform/di_addr_lookup"
      + entity_type = "user"
      + id          = (known after apply)
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.account-identity.di-kafka-source-supply-address-match-cancelled"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.account-identity.di-kafka-source-supply-address-match-cancelled"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-corr-address-debt-payment-requested"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-corr-address-debt-payment-requested"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-address-bad-debt-cancelled"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-address-bad-debt-cancelled"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-address-bad-debt-check-passed"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-address-bad-debt-check-passed"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-address-bad-debt-check-skipped"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-address-bad-debt-check-skipped"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-address-bad-debt-created"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-address-bad-debt-created"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-address-bad-debt-proof-requested"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-address-bad-debt-proof-requested"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-address-bad-debt-resolved"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-address-bad-debt-resolved"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-address-match-cancelled"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-address-match-cancelled"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-address-match-check-passed"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-address-match-check-passed"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-address-match-check-skipped"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-address-match-check-skipped"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-address-match-created"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-address-match-created"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-address-match-deposit-requested"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-address-match-deposit-requested"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-address-match-resolved"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-address-match-resolved"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-land-registry-exception-cancelled"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-land-registry-exception-cancelled"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-land-registry-exception-check-attempt-failed"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-land-registry-exception-check-attempt-failed"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-land-registry-exception-check-marked-as-failed"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-land-registry-exception-check-marked-as-failed"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-land-registry-exception-created"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-land-registry-exception-created"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-land-registry-exception-deferred"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-land-registry-exception-deferred"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-land-registry-exception-letter-sent"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-land-registry-exception-letter-sent"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-land-registry-exception-resolved"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-land-registry-exception-resolved"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-supply-address-bad-debt-cancelled"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-address-bad-debt-cancelled"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-supply-address-bad-debt-check-passed"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-address-bad-debt-check-passed"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-supply-address-bad-debt-check-skipped"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-address-bad-debt-check-skipped"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-supply-address-bad-debt-created"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-address-bad-debt-created"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-supply-address-bad-debt-payment-requested"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-address-bad-debt-payment-requested"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-supply-address-bad-debt-proof-requested"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-address-bad-debt-proof-requested"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-supply-address-bad-debt-resolved"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-address-bad-debt-resolved"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-supply-address-match-cancelled"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-address-match-cancelled"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-supply-address-match-check-passed"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-address-match-check-passed"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-supply-address-match-check-skipped"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-address-match-check-skipped"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-supply-address-match-created"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-address-match-created"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-supply-address-match-handed-over"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-address-match-handed-over"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.group_acl["account-identity.di-kafka-source-supply-address-match-resolved"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-address-match-resolved"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_exceptions.kafka_acl.topic_acl["account-identity.account.exceptions.v1"] will be created
  + resource "kafka_acl" "topic_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_exceptions"
      + id                           = (known after apply)
      + resource_name                = "account-identity.account.exceptions.v1"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Topic"
    }

  # module.account_identity_di_exceptions.kafka_quota.quota will be created
  + resource "kafka_quota" "quota" {
      + config      = {
          + "consumer_byte_rate" = 5242880
          + "producer_byte_rate" = 5242880
          + "request_percentage" = 100
        }
      + entity_name = "CN=account-platform/di_exceptions"
      + entity_type = "user"
      + id          = (known after apply)
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-account-dob-one"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-account-dob-one"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-account-dob-two"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-account-dob-two"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-account-label"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-account-label"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-account-occupancy-type"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-account-occupancy-type"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-address"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-address"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-email"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-email"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-landline"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-landline"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-mobile"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-mobile"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-correspondence-name"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-correspondence-name"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-legacy-account"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-legacy-account"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-legacy-account-braze"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-legacy-account-braze"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-residential-account-created"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-residential-account-created"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-supply-address"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-address"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-supply-email"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-email"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-supply-landline"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-landline"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-supply-mobile"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-mobile"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.group_acl["account-identity.di-kafka-source-supply-name"] will be created
  + resource "kafka_acl" "group_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.di-kafka-source-supply-name"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Group"
    }

  # module.account_identity_di_legacy_account.kafka_acl.topic_acl["account-identity.internal.legacy.account.events"] will be created
  + resource "kafka_acl" "topic_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.internal.legacy.account.events"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Topic"
    }

  # module.account_identity_di_legacy_account.kafka_acl.topic_acl["account-identity.legacy.account.braze.events.compacted"] will be created
  + resource "kafka_acl" "topic_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.legacy.account.braze.events.compacted"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Topic"
    }

  # module.account_identity_di_legacy_account.kafka_acl.topic_acl["account-identity.legacy.account.events"] will be created
  + resource "kafka_acl" "topic_acl" {
      + acl_host                     = "*"
      + acl_operation                = "Read"
      + acl_permission_type          = "Allow"
      + acl_principal                = "User:CN=account-platform/di_legacy_account"
      + id                           = (known after apply)
      + resource_name                = "account-identity.legacy.account.events"
      + resource_pattern_type_filter = "Literal"
      + resource_type                = "Topic"
    }

  # module.account_identity_di_legacy_account.kafka_quota.quota will be created
  + resource "kafka_quota" "quota" {
      + config      = {
          + "consumer_byte_rate" = 5242880
          + "producer_byte_rate" = 5242880
          + "request_percentage" = 100
        }
      + entity_name = "CN=account-platform/di_legacy_account"
      + entity_type = "user"
      + id          = (known after apply)
    }

Plan: 60 to add, 0 to change, 0 to destroy.

To manually trigger plan again please post @terraform-applier plan prod-aws/kafka-shared-msk/account-identity as comment.