CustomPrincipalBuilder implements KafkaPrincipalBuilder { @Override public CustomPrincipal build(AuthenticationContext authenticationContext) throws KafkaException{ try { CustomPrincipal customPrincipal; if (authenticationContext instanceof SaslAuthenticationContext) { SaslAuthenticationContext context = (SaslAuthenticationContext) authenticationContext; OAuthBearerTokenJwt token = (OAuthBearerTokenJwt) context.server() .getNegotiatedProperty("OAUTHBEARER.token"); customPrincipal = new CustomPrincipal("User", token.principalName()); customPrincipal.setOauthBearerTokenJwt(token); return customPrincipal; } else { throw new KafkaException("Failed to build CustomPrincipal. SaslAuthenticationContext is required."); } } catch (Exception ex) { throw new KafkaException("Failed to build CustomPrincipal due to: ", ex); } }  @Override CustomPrincipal build(AuthenticationContext authenticationContext);  }
@Test public void build() { OAuthBearerTokenJwt jwt = new OAuthBearerTokenJwt("token", 1, 1, "User"); Mockito.when(saslAuthenticationContext.server()).thenReturn(saslServer); Mockito.when(saslServer.getNegotiatedProperty("OAUTHBEARER.token")).thenReturn(jwt); CustomPrincipal customPrincipal = customPrincipalBuilder.build(saslAuthenticationContext); assertTrue(customPrincipal != null); }
@Test(expected = KafkaException.class) public void buildThrowException() { OAuthBearerTokenJwt jwt = new OAuthBearerTokenJwt("token", 1, 1, "User"); Mockito.when(saslAuthenticationContext.server()).thenReturn(null); CustomPrincipal customPrincipal = customPrincipalBuilder.build(saslAuthenticationContext); }
@Test(expected = KafkaException.class) public void buildBadType() { CustomPrincipal customPrincipal = customPrincipalBuilder.build(sslAuthenticationContext); }