KRaft controller OPA authorization

Beginne mit OpaAuthorizer.scala und verfolge anschließend die dort gezeigten Autorisierungsaufrufe für Controller-Anfragen durch AuthHelper.scala und ControllerApis.scala. Reproduziere das KRaft-Controller-Setup mit dem OPA-Plugin und untersuche die gemeldeten fehlenden Klassen. Als erledigt gilt die Aufgabe, wenn die Controller-Autorisierung während Fetch- und Registrierungsanfragen ohne Fehler beim Laden von Klassen funktioniert.

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.