Getting Error: trying to setup connection with GCP managed kafka service with Debezium using docker via compose.yaml?

Question

[org.apache.kafka.clients.admin.AdminClientConfig] debezium-connect-managed-new | 2024-12-20 18:01:24,017 INFO || Retrieved token with principal sub [org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredLoginCallbackHandler] debezium-connect-managed-new | 2024-12-20 18:01:24,026 INFO || Successfully logged in. [org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin] debezium-connect-managed-new | 2024-12-20 18:01:24,040 INFO || [Principal=:sub]: Expiring credential re-login thread started. [org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin] debezium-connect-managed-new | 2024-12-20 18:01:24,079 INFO || [Principal=sub]: Expiring credential valid from Fri Dec 20 18:01:23 UTC 2024 to Fri Dec 20 19:01:23 UTC 2024 [org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin] debezium-connect-managed-new | 2024-12-20 18:01:24,081 INFO || [Principal=:sub]: Expiring credential re-login sleeping until: Fri Dec 20 18:51:08 UTC 2024 [org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin] debezium-connect-managed-new | 2024-12-20 18:01:24,353 INFO || These configurations '[sasl.oauthbearer.token.endpoint.url, config.storage.topic, rest.advertised.host.name, group.id, status.storage.topic, rest.advertised.port, task.shutdown.graceful.timeout.ms, rest.host.name, plugin.path, offset.flush.timeout.ms, config.storage.replication.factor, offset.flush.interval.ms, rest.port, key.converter.schemas.enable, status.storage.replication.factor, value.converter.schemas.enable, offset.storage.replication.factor, offset.storage.topic, value.converter, key.converter]' were supplied but are not used yet. [org.apache.kafka.clients.admin.AdminClientConfig] debezium-connect-managed-new | 2024-12-20 18:01:24,357 INFO || Kafka version: 3.8.0 [org.apache.kafka.common.utils.AppInfoParser] debezium-connect-managed-new | 2024-12-20 18:01:24,358 INFO || Kafka commitId: 771b9576b00ecf5b [org.apache.kafka.common.utils.AppInfoParser] debezium-connect-managed-new | 2024-12-20 18:01:24,358 INFO || Kafka startTimeMs: 1734717684357 [org.apache.kafka.common.utils.AppInfoParser] debezium-connect-managed-new | 2024-12-20 18:01:26,038 INFO || [AdminClient clientId=adminclient-1] Failed authentication with bootstrap.managed-kafka-poc.asia-south2.managedkafka.managed-kafka-445110.cloud.goog/10.190.0.17 (channelId=-1) (Authentication failed during authentication due to invalid credentials with SASL mechanism OAUTHBEARER) [org.apache.kafka.common.network.Selector] debezium-connect-managed-new | 2024-12-20 18:01:26,043 INFO || [AdminClient clientId=adminclient-1] Node -1 disconnected. [org.apache.kafka.clients.NetworkClient] debezium-connect-managed-new | 2024-12-20 18:01:26,047 ERROR || [AdminClient clientId=adminclient-1] Connection to node -1 (bootstrap.managed-kafka-poc.asia-south2.managedkafka.managed-kafka-445110.cloud.goog/10.190.0.17:9092) failed authentication due to: Authentication failed during authentication due to invalid credentials with SASL mechanism OAUTHBEARER [org.apache.kafka.clients.NetworkClient] debezium-connect-managed-new | 2024-12-20 18:01:26,049 WARN || [AdminClient clientId=adminclient-1] Metadata update failed due to authentication error [org.apache.kafka.clients.admin.internals.AdminMetadataManager] debezium-connect-managed-new | org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed during authentication due to invalid credentials with SASL mechanism OAUTHBEARER debezium-connect-managed-new | 2024-12-20 18:01:26,057 INFO || App info kafka.admin.client for adminclient-1 unregistered [org.apache.kafka.common.utils.AppInfoParser] debezium-connect-managed-new | 2024-12-20 18:01:26,057 INFO || [AdminClient clientId=adminclient-1] Metadata update failed [org.apache.kafka.clients.admin.internals.AdminMetadataManager] debezium-connect-managed-new | org.apache.kafka.common.errors.TimeoutException: The AdminClient thread has exited. Call: fetchMetadata debezium-connect-managed-new | 2024-12-20 18:01:26,058 INFO || [AdminClient clientId=adminclient-1] Metadata update failed [org.apache.kafka.clients.admin.internals.AdminMetadataManager] debezium-connect-managed-new | org.apache.kafka.common.errors.TimeoutException: The AdminClient thread has exited. Call: fetchMetadata debezium-connect-managed-new | 2024-12-20 18:01:26,058 INFO || [AdminClient clientId=adminclient-1] Timed out 2 remaining operation(s) during close. [org.apache.kafka.clients.admin.KafkaAdminClient] debezium-connect-managed-new | 2024-12-20 18:01:26,065 INFO || [Principal=:sub]: Expiring credential re-login thread has been interrupted and will exit. [org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin] debezium-connect-managed-new | 2024-12-20 18:01:26,066 INFO || Metrics scheduler closed [org.apache.kafka.common.metrics.Metrics] debezium-connect-managed-new | 2024-12-20 18:01:26,067 INFO || Closing reporter org.apache.kafka.common.metrics.JmxReporter [org.apache.kafka.common.metrics.Metrics] debezium-connect-managed-new | 2024-12-20 18:01:26,067 INFO || Metrics reporters closed [org.apache.kafka.common.metrics.Metrics] debezium-connect-managed-new | 2024-12-20 18:01:26,068 ERROR || Stopping due to error [org.apache.kafka.connect.cli.AbstractConnectCli] debezium-connect-managed-new | org.apache.kafka.connect.errors.ConnectException: Failed to connect to and describe Kafka cluster. Check worker's broker connection and security properties.

my docker-compose.yaml file:

ervices: debezium-connect-managed-new: image: debezium/connect:3.0.0.Final container_name: debezium-connect-managed-new environment: BOOTSTRAP_SERVERS: bootstrap.managed-kafka-poc.asia-south2.managedkafka.managed-kafka-445110.cloud.goog:9092 GROUP_ID: debezium-group CONFIG_STORAGE_TOPIC: my_connect_configs OFFSET_STORAGE_TOPIC: my_connect_offsets STATUS_STORAGE_TOPIC: my_connect_statuses CONNECT_SECURITY_PROTOCOL: SASL_SSL CONNECT_SASL_MECHANISM: OAUTHBEARER CONNECT_SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL: "https://oauth2.googleapis.com/token" CONNECT_SASL_JAAS_CONFIG: 'org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="" clientSecret="cliendSecret" oauth.refresh.token="token" oauth.grant.type="refresh_token" sasl.login.callback.handler.class="org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler" sasl.oauthbearer.token.endpoint.url.file="/etc/kafka/kafka_client_jaas.txt" unsecuredLoginStringClaim_sub="sub" unsecuredLoginStringClaim_scope="scope";' volumes: - ./kafka_client_jaas_token.txt:/etc/kafka/kafka_client_jaas.txt ports: - "8084:8083" networks: - dev_net restart: unless-stopped

networks: dev_net: driver: bridge

This file /kafka_client_jaas_token.txt: has JWT token which i am generating via this curl

curl --location 'https://oauth2.googleapis.com/token'
--header 'Content-Type: application/x-www-form-urlencoded'
--data-urlencode 'client_id=764086051850-6qr4p6gpi6hn506pt8ejuq83di341hur.apps.googleusercontent.com'
--data-urlencode 'client_secret=d-FL95Q19q7MQmFpd7hHD0Ty'
--data-urlencode 'refresh_token=1//0glwSxrRN_NvzCgYIARAAGBASNwF-L9IrDuwgfFRqJLMtBLcmeguSvNwJbqXsq-QRYh3YFnhgAvT3M5NGoTHOqbuAjqbaq8YaqTc'
--data-urlencode 'grant_type=refresh_token'

I tried method given by google documentations but this doesn't work out https://cloud.google.com/managed-service-for-apache-kafka/docs/authentication-kafka#gcloud

I tried with authentication method :

1. PLAIN
2. OAUTHBEARER