KRaft controller OPA authorization

从 OpaAuthorizer.scala 开始，然后针对控制器请求，沿着 AuthHelper.scala 和 ControllerApis.scala 跟踪其中显示的授权调用。使用 OPA 插件复现 KRaft 控制器设置，并检查报告的缺失类。在 fetch 和注册请求期间，控制器授权能够正常工作且没有类加载错误，即表示完成。

In Kafka 4.0 and higher (KRaft-only mode), the authorization architecture for controllers is unified under the same pluggable interface as brokers (org.apache.kafka.server.authorizer.Authorizer), but the execution context and identity handling differ. The kafka-opa-plugin implements this interface
https://github.com/StyraOSS/opa-kafka-plugin/blob/v1.5.1/src/main/scala/org/openpolicyagent/kafka/OpaAuthorizer.scala#L34.

Simply trying the plugin in the controllers does not work (seems to be shading/scala?):

2026-01-15T09:51:15,093 ERROR [data-plane-kafka-request-handler-7] kafka.server.ControllerApis - [ControllerApis nodeId=2110489703] Unexpected error handling request RequestHeader(apiKey=FETCH, apiVersion=18, clientId=raft-client-1243966388, correlationId=1616, headerVersion=2) -- FetchRequestData(clusterId='test-kafka', replicaId=-1, replicaState=ReplicaState(replicaId=1243966388, replicaEpoch=-1), maxWaitMs=500, minBytes=0, maxBytes=8388608, isolationLevel=0, sessionId=0, sessionEpoch=-1, topics=[FetchTopic(topic='', topicId=AAAAAAAAAAAAAAAAAAAAAQ, partitions=[FetchPartition(partition=0, currentLeaderEpoch=0, fetchOffset=0, lastFetchedEpoch=0, logStartOffset=-1, partitionMaxBytes=0, replicaDirectoryId=pixhr-LaQ6K_ZtWDiJfcGQ, highWatermark=-1)])], forgottenTopicsData=[], rackId='') with context RequestContext(header=RequestHeader(apiKey=FETCH, apiVersion=18, clientId=raft-client-1243966388, correlationId=1616, headerVersion=2), connectionId='10.244.0.17:9093-10.244.0.16:35662-1-0', clientAddress=/10.244.0.16, principal=User:CN=generated certificate for pod, listenerName=ListenerName(CONTROLLER), securityProtocol=SSL, clientInformation=ClientInformation(softwareName=apache-kafka-java, softwareVersion=4.1.0-stackable0.0.0-dev), fromPrivilegedListener=false, principalSerde=Optional[org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder@590ff862])
com.google.common.util.concurrent.ExecutionError: java.lang.NoClassDefFoundError: com/fasterxml/jackson/module/scala/DefaultScalaModule$
	at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2049) ~[opa-authorizer-1.5.1-all.jar:?]
	at com.google.common.cache.LocalCache.get(LocalCache.java:3951) ~[opa-authorizer-1.5.1-all.jar:?]
	at com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4848) ~[opa-authorizer-1.5.1-all.jar:?]
	at org.openpolicyagent.kafka.OpaAuthorizer.allowAccess$1(OpaAuthorizer.scala:190) ~[opa-authorizer-1.5.1-all.jar:?]
	at org.openpolicyagent.kafka.OpaAuthorizer.doAuthorize(OpaAuthorizer.scala:201) ~[opa-authorizer-1.5.1-all.jar:?]
	at org.openpolicyagent.kafka.OpaAuthorizer.authorizeAction(OpaAuthorizer.scala:157) ~[opa-authorizer-1.5.1-all.jar:?]
	at org.openpolicyagent.kafka.OpaAuthorizer.$anonfun$authorize$1(OpaAuthorizer.scala:55) ~[opa-authorizer-1.5.1-all.jar:?]
	at scala.collection.StrictOptimizedIterableOps.map(StrictOptimizedIterableOps.scala:100) ~[scala-library-2.13.16.jar:?]
	at scala.collection.StrictOptimizedIterableOps.map$(StrictOptimizedIterableOps.scala:87) ~[scala-library-2.13.16.jar:?]
	at scala.collection.convert.JavaCollectionWrappers$JListWrapper.map(JavaCollectionWrappers.scala:138) ~[scala-library-2.13.16.jar:?]
	at org.openpolicyagent.kafka.OpaAuthorizer.authorize(OpaAuthorizer.scala:55) ~[opa-authorizer-1.5.1-all.jar:?]
	at kafka.server.AuthHelper.$anonfun$authorize$1(AuthHelper.scala:53) ~[kafka_2.13-4.1.0-stackable0.0.0-dev.jar:?]
	at kafka.server.AuthHelper.$anonfun$authorize$1$adapted(AuthHelper.scala:50) ~[kafka_2.13-4.1.0-stackable0.0.0-dev.jar:?]
	at scala.Option.forall(Option.scala:420) ~[scala-library-2.13.16.jar:?]
	at kafka.server.AuthHelper.authorize(AuthHelper.scala:50) ~[kafka_2.13-4.1.0-stackable0.0.0-dev.jar:?]
	at kafka.server.AuthHelper.authorizeClusterOperation(AuthHelper.scala:58) ~[kafka_2.13-4.1.0-stackable0.0.0-dev.jar:?]
	at kafka.server.ControllerApis.handleFetch(ControllerApis.scala:188) ~[kafka_2.13-4.1.0-stackable0.0.0-dev.jar:?]
	at kafka.server.ControllerApis.handle(ControllerApis.scala:96) ~[kafka_2.13-4.1.0-stackable0.0.0-dev.jar:?]
	at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:158) ~[kafka_2.13-4.1.0-stackable0.0.0-dev.jar:?]
	at java.base/java.lang.Thread.run(Unknown Source) [?:?]
Caused by: java.lang.NoClassDefFoundError: com/fasterxml/jackson/module/scala/DefaultScalaModule$
	at org.openpolicyagent.kafka.AllowCallable.call(OpaAuthorizer.scala:309) ~[opa-authorizer-1.5.1-all.jar:?]
	at org.openpolicyagent.kafka.AllowCallable.call(OpaAuthorizer.scala:306) ~[opa-authorizer-1.5.1-all.jar:?]
	at com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4853) ~[opa-authorizer-1.5.1-all.jar:?]
	at com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3529) ~[opa-authorizer-1.5.1-all.jar:?]
	at com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2278) ~[opa-authorizer-1.5.1-all.jar:?]
	at com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2155) ~[opa-authorizer-1.5.1-all.jar:?]
	at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2045) ~[opa-authorizer-1.5.1-all.jar:?]
	... 19 more
Caused by: java.lang.ClassNotFoundException: com.fasterxml.jackson.module.scala.DefaultScalaModule$
	at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(Unknown Source) ~[?:?]
	at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(Unknown Source) ~[?:?]
	at java.base/java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?]
	at org.openpolicyagent.kafka.AllowCallable.call(OpaAuthorizer.scala:309) ~[opa-authorizer-1.5.1-all.jar:?]
	at org.openpolicyagent.kafka.AllowCallable.call(OpaAuthorizer.scala:306) ~[opa-authorizer-1.5.1-all.jar:?]
	at com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4853) ~[opa-authorizer-1.5.1-all.jar:?]
	at com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3529) ~[opa-authorizer-1.5.1-all.jar:?]
	at com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2278) ~[opa-authorizer-1.5.1-all.jar:?]
	at com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2155) ~[opa-authorizer-1.5.1-all.jar:?]
	at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2045) ~[opa-authorizer-1.5.1-all.jar:?]
	... 19 more

Providing the jar https://mvnrepository.com/artifact/com.fasterxml.jackson.module/jackson-module-scala_2.13/2.19.3 leads to:

2026-02-24T15:33:36,971 INFO [controller-2110489703-to-controller-registration-channel-manager] kafka.server.ControllerRegistrationManager - [ControllerRegistrationManager id=2110489703 incarnation=S5OwPr-4SXqTwX3WYuDk1A] RegistrationResponseHandler: controller returned error UNKNOWN_SERVER_ERROR (The server experienced an unexpected error when processing the request.)
2026-02-24T15:33:49,592 INFO [controller-2110489703-registration-manager-event-handler] kafka.server.ControllerRegistrationManager - [ControllerRegistrationManager id=2110489703 incarnation=S5OwPr-4SXqTwX3WYuDk1A] sendControllerRegistration: attempting to send ControllerRegistrationRequestData(controllerId=2110489703, incarnationId=S5OwPr-4SXqTwX3WYuDk1A, zkMigrationReady=false, listeners=[Listener(name='CONTROLLER', host='test-kafka-controller-default-0.test-kafka-controller-default-headless.kuttl-test-charmed-coral.svc.cluster.local', port=9093, securityProtocol=1)], features=[Feature(name='group.version', minSupportedVersion=0, maxSupportedVersion=1), Feature(name='transaction.version', minSupportedVersion=0, maxSupportedVersion=2), Feature(name='eligible.leader.replicas.version', minSupportedVersion=0, maxSupportedVersion=1), Feature(name='kraft.version', minSupportedVersion=0, maxSupportedVersion=1), Feature(name='metadata.version', minSupportedVersion=7, maxSupportedVersion=27), Feature(name='share.version', minSupportedVersion=0, maxSupportedVersion=1)])
2026-02-24T15:33:49,595 ERROR [data-plane-kafka-request-handler-3] kafka.server.ControllerApis - [ControllerApis nodeId=2110489703] Unexpected error handling request RequestHeader(apiKey=CONTROLLER_REGISTRATION, apiVersion=0, clientId=2110489703, correlationId=9, headerVersion=2) -- ControllerRegistrationRequestData(controllerId=2110489703, incarnationId=S5OwPr-4SXqTwX3WYuDk1A, zkMigrationReady=false, listeners=[Listener(name='CONTROLLER', host='test-kafka-controller-default-0.test-kafka-controller-default-headless.kuttl-test-charmed-coral.svc.cluster.local', port=9093, securityProtocol=1)], features=[Feature(name='group.version', minSupportedVersion=0, maxSupportedVersion=1), Feature(name='transaction.version', minSupportedVersion=0, maxSupportedVersion=2), Feature(name='eligible.leader.replicas.version', minSupportedVersion=0, maxSupportedVersion=1), Feature(name='kraft.version', minSupportedVersion=0, maxSupportedVersion=1), Feature(name='metadata.version', minSupportedVersion=7, maxSupportedVersion=27), Feature(name='share.version', minSupportedVersion=0, maxSupportedVersion=1)]) with context RequestContext(header=RequestHeader(apiKey=CONTROLLER_REGISTRATION, apiVersion=0, clientId=2110489703, correlationId=9, headerVersion=2), connectionId='10.244.0.27:9093-10.244.0.27:37006-0-0', clientAddress=/10.244.0.27, principal=User:CN=generated certificate for pod, listenerName=ListenerName(CONTROLLER), securityProtocol=SSL, clientInformation=ClientInformation(softwareName=apache-kafka-java, softwareVersion=4.1.0-stackable0.0.0-dev), fromPrivilegedListener=false, principalSerde=Optional[org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder@24e367c5])
com.google.common.util.concurrent.ExecutionError: java.lang.NoClassDefFoundError: com/thoughtworks/paranamer/Paranamer
   at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2049) ~[opa-authorizer-1.5.1-all.jar:?]
   at com.google.common.cache.LocalCache.get(LocalCache.java:3951) ~[opa-authorizer-1.5.1-all.jar:?]
   at com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4848) ~[opa-authorizer-1.5.1-all.jar:?]
   at org.openpolicyagent.kafka.OpaAuthorizer.allowAccess$1(OpaAuthorizer.scala:190) ~[opa-authorizer-1.5.1-all.jar:?]
   at org.openpolicyagent.kafka.OpaAuthorizer.doAuthorize(OpaAuthorizer.scala:201) ~[opa-authorizer-1.5.1-all.jar:?]
   at org.openpolicyagent.kafka.OpaAuthorizer.authorizeAction(OpaAuthorizer.scala:157) ~[opa-authorizer-1.5.1-all.jar:?]
   at org.openpolicyagent.kafka.OpaAuthorizer.$anonfun$authorize$1(OpaAuthorizer.scala:55) ~[opa-authorizer-1.5.1-all.jar:?]
   at scala.collection.StrictOptimizedIterableOps.map(StrictOptimizedIterableOps.scala:100) ~[scala-library-2.13.16.jar:?]
   at scala.collection.StrictOptimizedIterableOps.map$(StrictOptimizedIterableOps.scala:87) ~[scala-library-2.13.16.jar:?]
   at scala.collection.convert.JavaCollectionWrappers$JListWrapper.map(JavaCollectionWrappers.scala:138) ~[scala-library-2.13.16.jar:?]
   at org.openpolicyagent.kafka.OpaAuthorizer.authorize(OpaAuthorizer.scala:55) ~[opa-authorizer-1.5.1-all.jar:?]
   at kafka.server.AuthHelper.$anonfun$authorize$1(AuthHelper.scala:53) ~[kafka_2.13-4.1.0-stackable0.0.0-dev.jar:?]
   at kafka.server.AuthHelper.$anonfun$authorize$1$adapted(AuthHelper.scala:50) ~[kafka_2.13-4.1.0-stackable0.0.0-dev.jar:?]
   at scala.Option.forall(Option.scala:420) ~[scala-library-2.13.16.jar:?]
   at kafka.server.AuthHelper.authorize(AuthHelper.scala:50) ~[kafka_2.13-4.1.0-stackable0.0.0-dev.jar:?]
   at kafka.server.AuthHelper.authorizeClusterOperation(AuthHelper.scala:58) ~[kafka_2.13-4.1.0-stackable0.0.0-dev.jar:?]
   at kafka.server.ControllerApis.handleControllerRegistration(ControllerApis.scala:863) ~[kafka_2.13-4.1.0-stackable0.0.0-dev.jar:?]
   at kafka.server.ControllerApis.handle(ControllerApis.scala:130) ~[kafka_2.13-4.1.0-stackable0.0.0-dev.jar:?]
   at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:158) ~[kafka_2.13-4.1.0-stackable0.0.0-dev.jar:?]
   at java.base/java.lang.Thread.run(Unknown Source) [?:?]
Caused by: java.lang.NoClassDefFoundError: com/thoughtworks/paranamer/Paranamer
   at com.fasterxml.jackson.module.scala.introspect.BeanIntrospector$.getCtorParams(BeanIntrospector.scala:245) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at com.fasterxml.jackson.module.scala.introspect.BeanIntrospector$.$anonfun$apply$2(BeanIntrospector.scala:53) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at scala.collection.StrictOptimizedIterableOps.flatMap(StrictOptimizedIterableOps.scala:118) ~[scala-library-2.13.16.jar:?]
   at scala.collection.StrictOptimizedIterableOps.flatMap$(StrictOptimizedIterableOps.scala:105) ~[scala-library-2.13.16.jar:?]
   at scala.collection.immutable.Vector.flatMap(Vector.scala:116) ~[scala-library-2.13.16.jar:?]
   at com.fasterxml.jackson.module.scala.introspect.BeanIntrospector$.findConstructorParam$1(BeanIntrospector.scala:53) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at com.fasterxml.jackson.module.scala.introspect.BeanIntrospector$.$anonfun$apply$22(BeanIntrospector.scala:194) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at scala.collection.ArrayOps$.map$extension(ArrayOps.scala:936) ~[scala-library-2.13.16.jar:?]
   at com.fasterxml.jackson.module.scala.introspect.BeanIntrospector$.$anonfun$apply$16(BeanIntrospector.scala:187) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at scala.collection.immutable.List.flatMap(List.scala:294) ~[scala-library-2.13.16.jar:?]
   at scala.collection.immutable.List.flatMap(List.scala:79) ~[scala-library-2.13.16.jar:?]
   at com.fasterxml.jackson.module.scala.introspect.BeanIntrospector$.apply(BeanIntrospector.scala:186) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at com.fasterxml.jackson.module.scala.introspect.ScalaAnnotationIntrospector$._descriptorFor(ScalaAnnotationIntrospectorModule.scala:196) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at com.fasterxml.jackson.module.scala.introspect.ScalaAnnotationIntrospector$.fieldName(ScalaAnnotationIntrospectorModule.scala:207) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at com.fasterxml.jackson.module.scala.introspect.ScalaAnnotationIntrospector$.findImplicitPropertyName(ScalaAnnotationIntrospectorModule.scala:41) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at com.fasterxml.jackson.databind.introspect.AnnotationIntrospectorPair.findImplicitPropertyName(AnnotationIntrospectorPair.java:462) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.introspect.POJOPropertiesCollector._addFields(POJOPropertiesCollector.java:562) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.introspect.POJOPropertiesCollector.collectAll(POJOPropertiesCollector.java:440) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.introspect.POJOPropertiesCollector.getJsonValueAccessor(POJOPropertiesCollector.java:270) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.introspect.BasicBeanDescription.findJsonValueAccessor(BasicBeanDescription.java:248) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.ser.BasicSerializerFactory.findSerializerByAnnotations(BasicSerializerFactory.java:394) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.ser.BeanSerializerFactory._createSerializer2(BeanSerializerFactory.java:222) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.ser.BeanSerializerFactory.createSerializer(BeanSerializerFactory.java:171) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.SerializerProvider._createUntypedSerializer(SerializerProvider.java:1554) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.SerializerProvider._createAndCacheUntypedSerializer(SerializerProvider.java:1502) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.SerializerProvider.findValueSerializer(SerializerProvider.java:586) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.SerializerProvider.findTypedValueSerializer(SerializerProvider.java:869) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.ser.DefaultSerializerProvider.serializeValue(DefaultSerializerProvider.java:331) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.ObjectMapper._writeValueAndClose(ObjectMapper.java:4859) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.ObjectMapper.writeValueAsString(ObjectMapper.java:4079) ~[jackson-databind-2.19.0.jar:2.19.0]
   at org.openpolicyagent.kafka.AllowCallable.call(OpaAuthorizer.scala:309) ~[opa-authorizer-1.5.1-all.jar:?]
   at org.openpolicyagent.kafka.AllowCallable.call(OpaAuthorizer.scala:306) ~[opa-authorizer-1.5.1-all.jar:?]
   at com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4853) ~[opa-authorizer-1.5.1-all.jar:?]
   at com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3529) ~[opa-authorizer-1.5.1-all.jar:?]
   at com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2278) ~[opa-authorizer-1.5.1-all.jar:?]
   at com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2155) ~[opa-authorizer-1.5.1-all.jar:?]
   at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2045) ~[opa-authorizer-1.5.1-all.jar:?]
   ... 19 more
Caused by: java.lang.ClassNotFoundException: com.thoughtworks.paranamer.Paranamer
   at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(Unknown Source) ~[?:?]
   at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(Unknown Source) ~[?:?]
   at java.base/java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?]
   at com.fasterxml.jackson.module.scala.introspect.BeanIntrospector$.getCtorParams(BeanIntrospector.scala:245) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at com.fasterxml.jackson.module.scala.introspect.BeanIntrospector$.$anonfun$apply$2(BeanIntrospector.scala:53) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at scala.collection.StrictOptimizedIterableOps.flatMap(StrictOptimizedIterableOps.scala:118) ~[scala-library-2.13.16.jar:?]
   at scala.collection.StrictOptimizedIterableOps.flatMap$(StrictOptimizedIterableOps.scala:105) ~[scala-library-2.13.16.jar:?]
   at scala.collection.immutable.Vector.flatMap(Vector.scala:116) ~[scala-library-2.13.16.jar:?]
   at com.fasterxml.jackson.module.scala.introspect.BeanIntrospector$.findConstructorParam$1(BeanIntrospector.scala:53) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at com.fasterxml.jackson.module.scala.introspect.BeanIntrospector$.$anonfun$apply$22(BeanIntrospector.scala:194) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at scala.collection.ArrayOps$.map$extension(ArrayOps.scala:936) ~[scala-library-2.13.16.jar:?]
   at com.fasterxml.jackson.module.scala.introspect.BeanIntrospector$.$anonfun$apply$16(BeanIntrospector.scala:187) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at scala.collection.immutable.List.flatMap(List.scala:294) ~[scala-library-2.13.16.jar:?]
   at scala.collection.immutable.List.flatMap(List.scala:79) ~[scala-library-2.13.16.jar:?]
   at com.fasterxml.jackson.module.scala.introspect.BeanIntrospector$.apply(BeanIntrospector.scala:186) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at com.fasterxml.jackson.module.scala.introspect.ScalaAnnotationIntrospector$._descriptorFor(ScalaAnnotationIntrospectorModule.scala:196) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at com.fasterxml.jackson.module.scala.introspect.ScalaAnnotationIntrospector$.fieldName(ScalaAnnotationIntrospectorModule.scala:207) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at com.fasterxml.jackson.module.scala.introspect.ScalaAnnotationIntrospector$.findImplicitPropertyName(ScalaAnnotationIntrospectorModule.scala:41) ~[jackson-module-scala_2.13-2.19.3.jar:2.19.3]
   at com.fasterxml.jackson.databind.introspect.AnnotationIntrospectorPair.findImplicitPropertyName(AnnotationIntrospectorPair.java:462) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.introspect.POJOPropertiesCollector._addFields(POJOPropertiesCollector.java:562) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.introspect.POJOPropertiesCollector.collectAll(POJOPropertiesCollector.java:440) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.introspect.POJOPropertiesCollector.getJsonValueAccessor(POJOPropertiesCollector.java:270) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.introspect.BasicBeanDescription.findJsonValueAccessor(BasicBeanDescription.java:248) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.ser.BasicSerializerFactory.findSerializerByAnnotations(BasicSerializerFactory.java:394) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.ser.BeanSerializerFactory._createSerializer2(BeanSerializerFactory.java:222) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.ser.BeanSerializerFactory.createSerializer(BeanSerializerFactory.java:171) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.SerializerProvider._createUntypedSerializer(SerializerProvider.java:1554) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.SerializerProvider._createAndCacheUntypedSerializer(SerializerProvider.java:1502) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.SerializerProvider.findValueSerializer(SerializerProvider.java:586) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.SerializerProvider.findTypedValueSerializer(SerializerProvider.java:869) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.ser.DefaultSerializerProvider.serializeValue(DefaultSerializerProvider.java:331) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.ObjectMapper._writeValueAndClose(ObjectMapper.java:4859) ~[jackson-databind-2.19.0.jar:2.19.0]
   at com.fasterxml.jackson.databind.ObjectMapper.writeValueAsString(ObjectMapper.java:4079) ~[jackson-databind-2.19.0.jar:2.19.0]
   at org.openpolicyagent.kafka.AllowCallable.call(OpaAuthorizer.scala:309) ~[opa-authorizer-1.5.1-all.jar:?]
   at org.openpolicyagent.kafka.AllowCallable.call(OpaAuthorizer.scala:306) ~[opa-authorizer-1.5.1-all.jar:?]
   at com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4853) ~[opa-authorizer-1.5.1-all.jar:?]
   at com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3529) ~[opa-authorizer-1.5.1-all.jar:?]
   at com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2278) ~[opa-authorizer-1.5.1-all.jar:?]
   at com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2155) ~[opa-authorizer-1.5.1-all.jar:?]
   at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2045) ~[opa-authorizer-1.5.1-all.jar:?]
   ... 19 more

Strimzi uses the kafka-opa-plugin in 1.5.1 https://github.com/strimzi/strimzi-kafka-operator/blob/9e9525225700a51fee721ae7a5c9d970c75fd0e3/docker-images/artifacts/kafka-thirdparty-libs/4.0.x/pom.xml#L23.

When reading https://kafka.apache.org/42/security/authorization-and-acls/#kraft-principal-forwarding:

In KRaft clusters, admin requests such as CreateTopics and DeleteTopics are sent to the broker listeners by the client. The broker then forwards the request to the active controller through the first listener configured in controller.listener.names. Authorization of these requests is done on the controller node. This is achieved by way of an Envelope request which packages both the underlying request from the client as well as the client principal. When the controller receives the forwarded Envelope request from the broker, it first authorizes the Envelope request using the authenticated broker principal. Then it authorizes the underlying request using the forwarded principal.

It is questionable what operations we would need to do on the controller and if the broker "proxy" does not suffice?
Additionally controllers a queried constantly, authorizing and requesting OPA constantly might introduce additional latency.