Kafka 的默认主体构建器类 支持根据对等证书的名称构建主体。自定义主体构建器应使用 SSL peer 证书的名称提供类型为 user 的主体。
以下示例显示了满足 AMQ Streams 的 OAuth 要求的自定义主体构建器。
自定义 OAuth 配置的主体构建器示例
public final class CustomKafkaPrincipalBuilder implements KafkaPrincipalBuilder {
public KafkaPrincipalBuilder() {}
@Override
public KafkaPrincipal build(AuthenticationContext context) {
if (context instanceof SslAuthenticationContext) {
SSLSession sslSession = ((SslAuthenticationContext) context).session();
try {
return new KafkaPrincipal(
KafkaPrincipal.USER_TYPE, sslSession.getPeerPrincipal().getName());
} catch (SSLPeerUnverifiedException e) {
throw new IllegalArgumentException("Cannot use an unverified peer for authentication", e);
}
}
// Create your own KafkaPrincipal here
...
}
}
public final class CustomKafkaPrincipalBuilder implements KafkaPrincipalBuilder {
public KafkaPrincipalBuilder() {}
@Override
public KafkaPrincipal build(AuthenticationContext context) {
if (context instanceof SslAuthenticationContext) {
SSLSession sslSession = ((SslAuthenticationContext) context).session();
try {
return new KafkaPrincipal(
KafkaPrincipal.USER_TYPE, sslSession.getPeerPrincipal().getName());
} catch (SSLPeerUnverifiedException e) {
throw new IllegalArgumentException("Cannot use an unverified peer for authentication", e);
}
}
// Create your own KafkaPrincipal here
...
}
}
Copy to ClipboardCopied!Toggle word wrapToggle overflow