...

Before KRaft, custom KafkaPrincipalBuilder implementations only needed to build principals. However, KRaft requires brokers to forward requests involving these principals to controllers. Without corresponding KafkaPrincipalSerde implementations, brokers cannot serialize/deserialize them, causing failures.

Currently, the  the API doesn't enforce the implementation of KafkaPrincipalSerde alongside KafkaPrincipalBuilder. Users would have already encountered serialization/deserialization issues during controller communication. This KIP aims to rectify this by making it a compile-time requirementerror, allowing developers to identify and fix the issue when implementing their custom KafkaPrincipalBuilder classes.

Public Interfaces

org.apache.kafka.common.security.auth.KafkaPrincipalBuilder

...

This is a change to a public API, and therefore has the potential to break existing code. However, the risk is considered acceptable because existing custom KafkaPrincipalBuilder implementations that are already implemented KafkaPrincipalSerde. must implement KafkaPrincipalSerde to work with KRaft. (This is already noted in the official doc)

Thus this change is proposed for a minor or feature release (specifically, 4.1 as suggested), allowing users sufficient time to adapt the changeand it will not affect any user.

Test Plan

Given that existing KafkaPrincipalBuilder implementations used with KRaft already implement KafkaPrincipalSerde, this change should not result in new compile-time errors, and existing test cases are expected to pass.

Rejected Alternatives

An alternative solution is to introduce a new interface that extends both KafkaPrincipalBuilder and KafkaPrincipalSerde. This approach avoids directly modifying the existing KafkaPrincipalBuilder interface and offers a more conservative evolution of the API.

public interface KafkaPrincipalBuilderWithSerde extends KafkaPrincipalBuilder, KafkaPrincipalSerde {
    // Inherits build(AuthenticationContext context), serialize(KafkaPrincipal),
    // and deserialize(byte[]) methods.
}

However, given that any existing KafkaPrincipalBuilder implementation within a KRaft environment must have already implemented KafkaPrincipalSerde, introducing a new interface is unnecessary.N/A