Unable to connect the cluster using confluent python Adminclient .

[L040639]

I am trying to get some topic info using confluent python admin client , i am getting below error. Do i need set anything to enable at the broker side ???

INFO:producer.main:Admin client created successfully
INFO:producer.main:Calling list_topics()...
ERROR:producer.main:Error fetching topic metadata: KafkaError{code=_TRANSPORT,val=-195,str="Failed to get metadata: Local: Broker transport failure"}
Traceback (most recent call last):
File "/home/app/app/producer/main.py", line 179, in fetch_topic_metadata
cluster_metadata = admin_client.list_topics(topic=topic, timeout=10)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/app/venv/lib/python3.12/site-packages/confluent_kafka/admin/init.py", line 639, in list_topics
return super(AdminClient, self).list_topics(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
cimpl.KafkaException: KafkaError{code=_TRANSPORT,val=-195,str="Failed to get metadata: Local: Broker transport failure"}

ef fetch_topic_metadata():
    """Fetch metadata for the configured Kafka topic"""
    
    bootstrap_servers = os.getenv('KAFKA_BOOTSTRAP_SERVERS')
    
    # Create Admin client with OAuth bearer token
    auth = TokenClientCredentials(ADMIN_CLIENT_ID, ADMIN_CLIENT_SECRET, SCOPE, TENANT_ID)
    
    admin_client = AdminClient({
        'bootstrap.servers': bootstrap_servers,
        'security.protocol': 'SASL_SSL',
        'sasl.mechanism': 'OAUTHBEARER',
        'oauth_cb': auth,
        'ssl.ca.location': '/home/jboss/cluster/ca.crt'
    })
    
    logger.info("Admin client created successfully")

    topic = os.getenv("KAFKA_TOPIC")

    try:
        # Fetch cluster metadata for the specific topic
        start_time = time.time()
        
        logger.info("Calling list_topics()...")
        # Fix the typo: 'imeout' -> 'timeout'
        cluster_metadata = admin_client.list_topics(topic=topic, timeout=10)
        

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka-cluster
  namespace: <XXX>
spec:
  kafka:
    authorization:
      type: opa
      url: https://<XXX>/v1/data/kafka/allow
      allowOnError: false
      enableMetrics: true
      tlsTrustedCertificates:
        - secretName: lk-ca-root-bundle
          certificate: ca.crt
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
      inter.broker.protocol.version: '3.9'
      connections.max.idle.ms: 1800000
    storage:
      type: persistent-claim
      size: 500Gi
      class: gp3
    listeners:
      - authentication:
          checkAccessTokenType: false
          clientId: <XX>
          clientSecret:
            key: secret
            secretName: entra-id-client
          jwksEndpointUri: '<XX>'
          maxSecondsWithoutReauthentication: 3600
          enableMetrics: true
          type: oauth
          validIssuerUri: '<XXX>'
          userNameClaim: azp
        configuration:
          brokerCertChainAndKey:
            certificate: tls.crt
            key: tls.key
            secretName: kafka-streams-prd-cert-secret
        name: oauth
        port: 9093
        tls: true
        type: internal
      - authentication:
          checkAccessTokenType: false
          clientId: <XXX>
          clientSecret:
            key: secret
            secretName: entra-id-client
          jwksEndpointUri: '<XXX>'
          maxSecondsWithoutReauthentication: 3600
          enableMetrics: true
          type: oauth
          validIssuerUri: '<XXX>'
          userNameClaim: azp
        configuration:
          brokerCertChainAndKey:
            certificate: tls.crt
            key: tls.key
            secretName: kafka-streams-prd-cert-secret
          bootstrap:
            alternativeNames:
            - <XX>
            annotations:
              external-dns.alpha.kubernetes.io/hostname: <XX>.
              external-dns.alpha.kubernetes.io/access: "private"
          brokers:
          - broker: 0
            advertisedHost: <XXX>
            annotations:
              external-dns.alpha.kubernetes.io/hostname: <XXXX>.
              external-dns.alpha.kubernetes.io/access: "private"
          - broker: 1
            advertisedHost: <XXX>
            annotations:
              external-dns.alpha.kubernetes.io/hostname: <XXX>.
              external-dns.alpha.kubernetes.io/access: "private"
          - broker: 2
            advertisedHost: <XXX>
            annotations:
              external-dns.alpha.kubernetes.io/hostname: <XXX>.
              external-dns.alpha.kubernetes.io/access: "private"
        name: external
        port: 9094
        tls: true
        type: loadbalancer
    version: 3.9.0
    replicas: 3
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          key: kafka-metrics-config.yml
          name: kafka-streams-prd-metrics
    template:
      externalBootstrapService:
        metadata:
          annotations:
            service.beta.kubernetes.io/aws-load-balancer-type: nlb
            service.beta.kubernetes.io/aws-load-balancer-scheme: internal
            service.beta.kubernetes.io/aws-load-balancer-nlb-target-type: instance
            service.beta.kubernetes.io/aws-load-balancer-internal: "true"
            service.beta.kubernetes.io/aws-load-balancer-connection-draining-enabled: "true"
            service.beta.kubernetes.io/aws-load-balancer-connection-draining-timeout: "60"
            service.beta.kubernetes.io/aws-load-balancer-listener-attributes.TCP-9094: tcp.idle_timeout.seconds=5000
      perPodService:
        metadata:
          annotations:
            service.beta.kubernetes.io/aws-load-balancer-type: nlb
            service.beta.kubernetes.io/aws-load-balancer-nlb-target-type: instance
            service.beta.kubernetes.io/aws-load-balancer-scheme: internal
            service.beta.kubernetes.io/aws-load-balancer-internal: "true"
            service.beta.kubernetes.io/aws-load-balancer-connection-draining-enabled: "true"
            service.beta.kubernetes.io/aws-load-balancer-connection-draining-timeout: "60"
            service.beta.kubernetes.io/aws-load-balancer-listener-attributes.TCP-9094: tcp.idle_timeout.seconds=5000
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
                - matchExpressions:
                    - key: <XX>   ##dedicated nodegroup for kafka
                      operator: Exists
          podAntiAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              - labelSelector:
                  matchExpressions:
                    - key: strimzi.io/cluster
                      operator: In
                      values:
                        - kafka-cluster
                    - key: strimzi.io/component-type
                      operator: In
                      values:
                        - kafka
                topologyKey: kubernetes.io/hostname
        tolerations:
          - key: "<XX>"
            operator: "Exists"
            effect: "NoSchedule"
        topologySpreadConstraints:
          - maxSkew: 1
            topologyKey: topology.kubernetes.io/zone
            whenUnsatisfiable: ScheduleAnyway
            labelSelector:
              matchLabels:
                strimzi.io/cluster: kafka-cluster
                strimzi.io/component-type: kafka
  entityOperator:
    topicOperator: {}
    userOperator: {}
    template:
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
                - matchExpressions:
                    - key: <XXX>
                      operator: Exists
        tolerations:
          - key: "<XX>"
            operator: "Exists"
            effect: "NoSchedule"
  zookeeper:
    storage:
      type: persistent-claim
      class: gp3
      size: 50Gi
    replicas: 3
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: kafka-streams-prd-metrics
          key: zookeeper-metrics-config.yml
    template:
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
                - matchExpressions:
                    - key: <XX>  ##dedicated nodegroup for kafka
                      operator: Exists
          podAntiAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              - labelSelector:
                  matchExpressions:
                    - key: strimzi.io/cluster
                      operator: In
                      values:
                        - kafka-cluster
                    - key: strimzi.io/component-type
                      operator: In
                      values:
                        - zookeeper
                topologyKey: kubernetes.io/hostname
        tolerations:
          - key: "<XX>"
            operator: "Exists"
            effect: "NoSchedule"
        topologySpreadConstraints:
          - maxSkew: 1
            topologyKey: topology.kubernetes.io/zone
            whenUnsatisfiable: ScheduleAnyway
            labelSelector:
              matchLabels:
                strimzi.io/cluster: kafka-cluster
                strimzi.io/component-type: zookeeper
  kafkaExporter:
    topicRegex: ".*"
    groupRegex: ".*"
    template:
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
                - matchExpressions:
                    - key: node-role.kubernetes.io/worker
                      operator: Exists
        tolerations:
          - key: "<XXX>"
            operator: "Exists"
            effect: "NoSchedule"