Type                 Name                                                Plan        Info
     pulumi:pulumi:Stack  modal-main                                                      1 error; 13 debugs
 ~   └─ kafka:index:Acl   kafka-cluster-<removed>                             refresh     1 error

Diagnostics:
  kafka:index:Acl (<removed>):
    error: Preview failed: refreshing urn:pulumi:main::<removed>: 1 error occurred:
        * kafka server: The client is not authorized to send this request type

  pulumi:pulumi:Stack (modal-main):
    debug: [TRACE] configuring provider with brokers @ &[<removed>.<http://amazonaws.com:9096|amazonaws.com:9096> <removed>.<http://amazonaws.com:9096|amazonaws.com:9096> <removed>.<http://amazonaws.com:9096|amazonaws.com:9096>]
    debug: [TRACE] Config @ {0xc0000105a0 120   ***** ***** 2.7.0 true false <removed> ***** scram-sha512    false }
    debug: [INFO] Reading ACL
    debug: [INFO] Reading ACL User:<removed>|*|All|Allow|Topic|*|Literal
    debug: [TRACE] configuring bootstrap_servers {0xc0000105a0 120   ***** ***** 2.7.0 true false <removed> ***** scram-sha512    false }
    debug: [INFO] configuring kafka client with {0xc0000105a0 120   ***** ***** 2.7.0 true false <removed> ***** scram-sha512    false }
    debug: [WARN] no CA file set skipping
    debug: [DEBUG] Got 11 topics from Kafka
    debug: [TRACE] lazy client init %!s(<nil>); config, {0xc0000105a0 120   ***** ***** 2.7.0 true false <removed> ***** scram-sha512    false }
    debug: [INFO] Listing all ACLS
    debug: [TRACE] Asking Kafka for all the resources
    debug: [TRACE] Describe Acl Requst &{1 {0 2 <nil> 1 <nil> <nil> 1 1}}
    debug: [TRACE] ThrottleTime: 0
    error: preview failed

class KafkaCluster(pulumi.ComponentResource):
    def __init__(self, name: str, env: str, admin_user: KafkaUser, other_users: list[KafkaUser], *, opts=None) -> None:
        ...

        kafka_provider = kafka.Provider(
            f"{name}-kafka-provider",
            bootstrap_servers=bootsrap_servers_private.apply(lambda x: x.split(",")),
            sasl_username=admin_user.username,
            sasl_password=admin_user.password.result.apply(lambda x: x),
            sasl_mechanism="scram-sha512",
            opts=pulumi.ResourceOptions(
                parent=self,
            ),
        )

        ...
        for resource_type in ["Topic", "Group", "TransactionalID"]:
            ...
            for user in all_users:
                _ = kafka.Acl(
                    f"{name}-kafka-acl-{user.username}-{resource_type}",
                    acl_host="*",
                    acl_operation="All",
                    acl_permission_type="Allow",
                    acl_principal=f"User:{user.username}",
                    acl_resource_type=resource_type,
                    acl_resource_name="*",
                    resource_pattern_type_filter="Literal",
                    opts=pulumi.ResourceOptions(parent=self, provider=kafka_provider),
                )