Skip to content

KRaft controller OPA authorization #941

Open
Open
KRaft controller OPA authorization#941
Assignees
razvan

Description

@maltesander
maltesander
opened on Feb 24, 2026

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.

Activity

  1. sbernauer commented on Oct 14, 2026

    @sbernauer
    sbernauer
    on Oct 14, 2026
    Member

    I didn't fully read all details, but if Kafka controllers talk to OPA we should add a affinity, see #1035 :)

  2. razvan
    added this to Stackable End-to-End Coordination and removed this from Stackable End-to-End Coordinationon Oct 28, 2026

  3. sbernauer commented on Oct 29, 2026

    @sbernauer
    sbernauer
    on Oct 29, 2026
    Member

    If we don't figure this out for SDP 26.11, we should re-visit if the can mark 4.x as stable (vs keeping it experimental)

  4. razvan commented on Oct 29, 2026

    @razvan
    razvan
    on Oct 29, 2026
    Member

    Investigation

    1. The OPA plugin build we ship is broken on Apache Kafka 4.x on brokers too, not just controllers. v1.5.1 expects Kafka to provide jackson-module-scala, but Kafka removed it in 4.0 (KAFKA-18773). Upstream fixed it on master (9c79c6d, 5f45709): it builds against Kafka 4.0 and bundles jackson-module-scala + paranamer into the shaded jar. There's no release yet. Our kuttl opa test only runs 3.9.2 because the issue was known.
    2. Authorizing on the brokers isn't enough in KRaft. Brokers forward CreateTopics, DeleteTopics, CreatePartitions, CreateAcls, AlterPartitionReassignments, etc. to the controller without authorizing them (KafkaApis.forwardToController). Authorization happens only in ControllerApis, and with no authorizer configured it allows everything. Right now any authenticated client can create or delete topics on a KRaft cluster, whatever the OPA policy says.

    Next steps

    • docker-images: build the plugin from upstream master, or backport the two commits as patchable patches on 1.5.1. Make sure the bundled jackson-module-scala matches Kafka's jackson-databind version.
    • operator: configure authorizer.class.name / opa.authorizer.url (+ TLS) on controllers as well. Add the internal node principal to super.users on both roles. The plugin skips OPA for super users, so controller-to-controller/broker-to-controller traffic won't touch OPA, which also covers the latency concern. Forwarded requests are still checked against the original client principal.
    • Rego: check the policy handles the cluster-level operations that reach the controller.
    • tests: add a Kafka 4.x KRaft variant to the opa test, with a negative test (an unauthorized user can't create topics).
    • Add a affinity from the controllers to OPA Pods, see KRaft controller OPA authorization #941 (comment)
  5. razvan
    self-assigned this
    on Oct 29, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Labels

No labels
No labels

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions