How to configure regarding schema registry with SASL_SSL/OAUTHBEARER ?

[nyeongna]

Hi folks! I’m trying to use kafbat-ui to browse Google Managed Schema Registry.

Google’s Schema Registry REST API only accepts OAuth (Bearer token), so Basic Auth doesn’t work. I’m looking for the correct way (if any) to configure kafbat-ui to authenticate with OAuth when calling the Schema Registry endpoints.

Environment

• kafbat-ui version: 1.3.0 (image)
• Deployment: GKE (Helm)
• Kafka cluster: Google Cloud Managed Kafka

What I tried

• Broker auth via SASL/OAUTHBEARER using
  com.google.cloud.hosted.kafka.auth.GcpLoginCallbackHandler — this works for broker connections.

• For Schema Registry, I attempted env vars like:

  - name: KAFKA_CLUSTERS_0_SCHEMAREGISTRY
    value: https://managedkafka.googleapis.com/v1/projects/<project>/locations/us-east1/schemaRegistries/<registry>
  - name: KAFKA_CLUSTERS_0_SCHEMAREGISTRY_AUTH_TYPE
    value: OAUTHBEARER
  - name: KAFKA_CLUSTERS_0_SCHEMAREGISTRY_PROPERTIES_SASL_LOGIN_CALLBACK_HANDLER_CLASS
    value: com.google.cloud.hosted.kafka.auth.GcpLoginCallbackHandler
  - name: KAFKA_CLUSTERS_0_SCHEMAREGISTRY_PROPERTIES_SASL_JAAS_CONFIG
    value: org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;

  But Schema Registry requests from kafbat-ui still fail with 401 Unauthorized.

Error

ava:322)
│     Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
│ Error has been observed at the following site(s):
│     *__checkpoint ⇢ 401 UNAUTHORIZED from GET https://managedkafka.googleapis.com/v1/projects/<project>/locations/us-east1/schemaRegistries/<registry>
│     *__checkpoint ⇢ Handler io.kafbat.ui.controller.SchemasController#getSchemas(String, Integer, Integer, String, ServerWebExchange) [DispatcherHan
│     *__checkpoint ⇢ io.kafbat.ui.config.CorsGlobalConfiguration$$Lambda/0x00007c25176ee0c8 [DefaultWebFilterChain]
│     *__checkpoint ⇢ io.kafbat.ui.config.ReadOnlyModeFilter [DefaultWebFilterChain]
│     *__checkpoint ⇢ io.kafbat.ui.config.CustomWebFilter [DefaultWebFilterChain]
│     *__checkpoint ⇢ AuthorizationWebFilter [DefaultWebFilterChain]
│     *__checkpoint ⇢ ExceptionTranslationWebFilter [DefaultWebFilterChain]
│     *__checkpoint ⇢ LogoutWebFilter [DefaultWebFilterChain]
│     *__checkpoint ⇢ ServerRequestCacheWebFilter [DefaultWebFilterChain]
│     *__checkpoint ⇢ SecurityContextServerWebExchangeWebFilter [DefaultWebFilterChain]
│     *__checkpoint ⇢ ReactorContextWebFilter [DefaultWebFilterChain]
│     *__checkpoint ⇢ HttpHeaderWriterWebFilter [DefaultWebFilterChain]
│     *__checkpoint ⇢ ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]
│     *__checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]
│     *__checkpoint ⇢ HTTP GET "/api/clusters/socar-data-dev-kafka-cluster/schemas?page=1&perPage=25" [ExceptionHandlingWebHandler]
│ Original Stack Trace:
│         at org.springframework.web.reactive.function.client.WebClientResponseException.create(WebClientResponseException.java:322)
│         at org.springframework.web.reactive.function.client.DefaultClientResponse.lambda$createException$1(DefaultClientResponse.java:214)
│         at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:106)
│         at reactor.core.publisher.FluxOnErrorReturn$ReturnSubscriber.onNext(FluxOnErrorReturn.java:162)
│         at reactor.core.publisher.FluxDefaultIfEmpty$DefaultIfEmptySubscriber.onNext(FluxDefaultIfEmpty.java:122)
│         at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:129)
│         at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:107)
│         at reactor.core.publisher.FluxMapFuseable$MapFuseableConditionalSubscriber.onNext(FluxMapFuseable.java:299)
│         at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber.onNext(FluxFilterFuseable.java:337)
│         at reactor.core.publisher.Operators$BaseFluxToMonoOperator.completePossiblyEmpty(Operators.java:2096)
│         at reactor.core.publisher.MonoCollect$CollectSubscriber.onComplete(MonoCollect.java:145)
│         at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:144)
│         at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:260)
│         at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:144)
│         at reactor.netty.channel.FluxReceive.onInboundComplete(FluxReceive.java:413)
│         at reactor.netty.channel.ChannelOperations.onInboundComplete(ChannelOperations.java:455)
│         at reactor.netty.channel.ChannelOperations.terminate(ChannelOperations.java:509)
│         at reactor.netty.http.client.HttpClientOperations.onInboundNext(HttpClientOperations.java:821)
│         at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:115)
│         at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
│         at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
│         at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
│         at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
│         at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:346)
│         at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:318)
│         at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
│         at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
│         at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
│         at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
│         at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1519)
│         at io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1377)
│         at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1428)
│         at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:530)
│         at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:469)
│         at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:290)
│         at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
│         at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
│         at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
│         at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1357)
│         at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
│         at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
│         at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:868)
│         at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799)
│         at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:501)
│         at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:399)
│         at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
│         at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
│         at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
│         at java.base/java.lang.Thread.run(Thread.java:1583)

Questions

1. Does kafbat-ui currently support attaching an OAuth Bearer token to Schema Registry REST calls?
2. If yes, what is the exact configuration to enable it?
3. If not, is the recommended approach to place a reverse proxy that injects the Bearer token, or is there another supported pattern?

Any guidance or examples would be greatly appreciated. Thanks!