diff --git a/common/queue/src/test/java/org/thingsboard/server/queue/kafka/TbKafkaSettingsConfluentOAuthHttpWarningTest.java b/common/queue/src/test/java/org/thingsboard/server/queue/kafka/TbKafkaSettingsConfluentOAuthHttpWarningTest.java new file mode 100644 index 0000000000..15219da536 --- /dev/null +++ b/common/queue/src/test/java/org/thingsboard/server/queue/kafka/TbKafkaSettingsConfluentOAuthHttpWarningTest.java @@ -0,0 +1,72 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.queue.kafka; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.TestPropertySource; + +import java.util.Properties; + +import static org.assertj.core.api.Assertions.assertThat; + +@SpringBootTest(classes = {TbKafkaSettings.class, KafkaAdmin.class}) +@TestPropertySource(properties = { + "queue.type=kafka", + "queue.kafka.bootstrap.servers=localhost:9092", + "queue.kafka.use_confluent_cloud=true", + "queue.kafka.confluent.sasl.mechanism=OAUTHBEARER", + "queue.kafka.confluent.security.protocol=SASL_PLAINTEXT", + "queue.kafka.confluent.oauth.client-id=my-client", + "queue.kafka.confluent.oauth.client-secret=my-secret", + "queue.kafka.confluent.oauth.endpoint-url=http://idp.example.com/oauth/token" +}) +class TbKafkaSettingsConfluentOAuthHttpWarningTest { + + @Autowired + TbKafkaSettings settings; + + @Test + void givenHttpEndpointUrl_whenToProps_thenProducesValidJaasConfigAndLogsWarning() { + Logger logger = (Logger) LoggerFactory.getLogger(TbKafkaSettings.class); + ListAppender appender = new ListAppender<>(); + appender.start(); + logger.addAppender(appender); + try { + Properties props = settings.toProps(); + + assertThat(props.getProperty("sasl.jaas.config")) + .contains("org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required") + .contains("clientId=\"my-client\"") + .contains("clientSecret=\"my-secret\""); + assertThat(props).containsEntry("sasl.oauthbearer.token.endpoint.url", + "http://idp.example.com/oauth/token"); + + assertThat(appender.list) + .filteredOn(e -> e.getLevel() == Level.WARN) + .anySatisfy(e -> assertThat(e.getFormattedMessage()).contains("not HTTPS")); + } finally { + logger.detachAppender(appender); + } + } + +} diff --git a/msa/js-executor/config/custom-environment-variables.yml b/msa/js-executor/config/custom-environment-variables.yml index 74c519b6fb..290ae4c219 100644 --- a/msa/js-executor/config/custom-environment-variables.yml +++ b/msa/js-executor/config/custom-environment-variables.yml @@ -54,17 +54,17 @@ kafka: mechanism: "TB_QUEUE_KAFKA_CONFLUENT_SASL_MECHANISM" username: "TB_QUEUE_KAFKA_CONFLUENT_USERNAME" password: "TB_QUEUE_KAFKA_CONFLUENT_PASSWORD" - # Optional: OAUTH Setting + # Optional: OAuth2 Setting oauth: - # Optional: Oauth Client ID + # Optional: OAuth2 Client ID client_id: "TB_QUEUE_KAFKA_CONFLUENT_OAUTH_CLIENT_ID" - # Optional: Oauth Client Secret + # Optional: OAuth2 Client Secret client_secret: "TB_QUEUE_KAFKA_CONFLUENT_OAUTH_CLIENT_SECRET" - # Optional: Oauth Endpoint URL to get the token from + # Optional: OAuth2 Endpoint URL to get the token from endpoint_url: "TB_QUEUE_KAFKA_CONFLUENT_OAUTH_ENDPOINT_URL" - # Optional: threshold to renew current token before it expires (ms) + # Optional: threshold to renew current token before it expires (ms). Token refresh is managed by this service; no equivalent setting exists on the JVM-based services. refresh_threshold: "TB_QUEUE_KAFKA_CONFLUENT_OAUTH_REFRESH_THRESHOLD" - # Optional: Oauth scope/audience requested for the token. Required by some IdPs for client-credentials (e.g. Azure AD's "api:///.default") + # Optional: OAuth2 scope/audience requested for the token. Required by some IdPs for client-credentials (e.g. Azure AD's "api:///.default") scope: "TB_QUEUE_KAFKA_CONFLUENT_OAUTH_SCOPE" logger: