@Override
        public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
            DELEGATE.handle(callbacks);
            // now change any returned token to have a different principal name
            for (Callback callback : callbacks) {
                if (callback instanceof OAuthBearerTokenCallback) {
                    OAuthBearerTokenCallback oauthBearerTokenCallback = (OAuthBearerTokenCallback) callback;
                    OAuthBearerToken token = oauthBearerTokenCallback.token();
                    if (token != null) {
                        String changedPrincipalNameToUse = token.principalName()
                                + ++numInvocations;
                        String headerJson = "{" + claimOrHeaderJsonText("alg", "none") + "}";
                        /*
                         * Use a short lifetime so the background refresh thread replaces it before we
                         * re-authenticate
                         */
                        String lifetimeSecondsValueToUse = "1";
                        String claimsJson;
                        try {
                            claimsJson = String.format("{%s,%s,%s}",
                                    expClaimText(Long.parseLong(lifetimeSecondsValueToUse)),
                                    claimOrHeaderJsonText("iat", time.milliseconds() / 1000.0),
                                    claimOrHeaderJsonText("sub", changedPrincipalNameToUse));
                        } catch (NumberFormatException e) {
                            throw new OAuthBearerConfigException(e.getMessage());
                        }
                        try {
                            Encoder urlEncoderNoPadding = Base64.getUrlEncoder().withoutPadding();
                            OAuthBearerUnsecuredJws jws = new OAuthBearerUnsecuredJws(String.format("%s.%s.",
                                    urlEncoderNoPadding.encodeToString(headerJson.getBytes(StandardCharsets.UTF_8)),
                                    urlEncoderNoPadding
                                            .encodeToString(claimsJson.getBytes(StandardCharsets.UTF_8))),
                                    "sub", "scope");
                            oauthBearerTokenCallback.token(jws);
                        } catch (OAuthBearerIllegalTokenException e) {
                            // occurs if the principal claim doesn't exist or has an empty value
                            throw new OAuthBearerConfigException(e.getMessage(), e);
                        }
                    }
                }
            }
        }
Tests that Kafka ApiVersionsRequests are handled by the SASL server authenticator
prior to SASL handshake flow and that subsequent authentication succeeds
when transport layer is PLAINTEXT/SSL. This test uses a non-SASL client that simulates
SASL authentication after ApiVersionsRequest.
<p>
Test sequence (using <tt>securityProtocol=PLAINTEXT</tt> as an example):
<ol>
  <li>Starts a SASL_PLAINTEXT test server that simply echoes back client requests after authentication.</li>
  <li>A (non-SASL) PLAINTEXT test client connects to the SASL server port. Client is now unauthenticated.<./li>
  <li>The unauthenticated non-SASL client sends an ApiVersionsRequest and validates the response.
      A valid response indicates that {@link SaslServerAuthenticator} of the test server responded to
      the ApiVersionsRequest even though the client is not yet authenticated.</li>
  <li>The unauthenticated non-SASL client sends a SaslHandshakeRequest and validates the response. A valid response
      indicates that {@link SaslServerAuthenticator} of the test server responded to the SaslHandshakeRequest
      after processing ApiVersionsRequest.</li>
  <li>The unauthenticated non-SASL client sends the SASL/PLAIN packet containing username/password to authenticate
      itself. The client is now authenticated by the server. At this point this test client is at the
      same state as a regular SASL_PLAINTEXT client that is <tt>ready</tt>.</li>
  <li>The authenticated client sends random data to the server and checks that the data is echoed
      back by the test server (ie, not Kafka request-response) to ensure that the client now
      behaves exactly as a regular SASL_PLAINTEXT client that has completed authentication.</li>
</ol>
clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java