Chapter 17. Enabling OAuth 2.0 token-based access

1. The Kafka client requests an access token from the authorization server using a client ID and credentials, and optionally a refresh token. Alternatively, the client may authenticate using a username and a password.
2. The authorization server generates a new access token.
3. The Kafka client authenticates with the Kafka broker using the SASL OAUTHBEARER mechanism to pass the access token.
4. The Kafka broker validates the access token by calling a token introspection endpoint on the authorization server using its own client ID and secret.
5. A Kafka client session is established if the token is valid.

1. The Kafka client authenticates with the authorization server from the token endpoint, using a client ID and credentials, and optionally a refresh token. Alternatively, the client may authenticate using a username and a password.
2. The authorization server generates a new access token.
3. The Kafka client authenticates with the Kafka broker using the SASL OAUTHBEARER mechanism to pass the access token.
4. The Kafka broker validates the access token locally using a JWT token signature check, and local token introspection.

1. The Kafka client authenticates with the Kafka broker using the SASL OAUTHBEARER mechanism to pass the long-lived access token.
2. The Kafka broker validates the access token by calling a token introspection endpoint on the authorization server, using its own client ID and secret.
3. A Kafka client session is established if the token is valid.

1. The Kafka client authenticates with the Kafka broker using the SASL OAUTHBEARER mechanism to pass the long-lived access token.
2. The Kafka broker validates the access token locally using a JWT token signature check and local token introspection.

1. The Kafka client passes a clientId as a username and a secret as a password.
2. The Kafka broker uses a token endpoint to pass the clientId and secret to the authorization server.
3. The authorization server returns a fresh access token or an error if the client credentials are not valid.

4. The Kafka broker validates the token in one of the following ways:

   1. If a token introspection endpoint is specified, the Kafka broker validates the access token by calling the endpoint on the authorization server. A session is established if the token validation is successful.
   2. If local token introspection is used, a request is not made to the authorization server. The Kafka broker validates the access token locally using a JWT token signature check.

1. The Kafka client passes a username and password. The password provides the value of an access token that was obtained manually and configured before running the client.

2. The password is passed with or without an $accessToken: string prefix depending on whether or not the Kafka broker listener is configured with a token endpoint for authentication.

   1. If the token endpoint is configured, the password should be prefixed by $accessToken: to let the broker know that the password parameter contains an access token rather than a client secret. The Kafka broker interprets the username as the account username.
   2. If the token endpoint is not configured on the Kafka broker listener (enforcing a no-client-credentials mode), the password should provide the access token without the prefix. The Kafka broker interprets the username as the account username. In this mode, the client doesn’t use a client ID and secret, and the password parameter is always interpreted as a raw access token.

3. The Kafka broker validates the token in one of the following ways:

   1. If a token introspection endpoint is specified, the Kafka broker validates the access token by calling the endpoint on the authorization server. A session is established if token validation is successful.
   2. If local token introspection is used, there is no request made to the authorization server. Kafka broker validates the access token locally using a JWT token signature check.

Procedure

1. Specify a listener in the Kafka resource with an oauth authentication type.

   Example listener configuration with OAuth 2.0 authentication

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     kafka:
       # ...
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: oauth
         - name: external3
           port: 9094
           type: loadbalancer
           tls: true
           authentication:
             type: oauth
         #...

2. Configure the OAuth listener depending on the authorization server and validation type:

   • Configuring fast local JWT token validation
   • Configuring token validation using an introspection endpoint
   • Including additional configuration options
3. Apply the changes to the Kafka configuration.

4. Check the update in the logs or by watching the pod state transitions:

   oc logs -f ${POD_NAME} -c ${CONTAINER_NAME}
   oc get pod -w

   The rolling update configures the brokers to use OAuth 2.0 authentication.

Procedure

1. Add the client library with OAuth 2.0 support to the pom.xml file for the Kafka client:

   <dependency>
    <groupId>io.strimzi</groupId>
    <artifactId>kafka-oauth-client</artifactId>
    <version>0.15.0.redhat-00012</version>
   </dependency>

2. Configure the client depending on the OAuth 2.0 authentication method:

   • Example client credentials configuration using the client secret
   • Example password grants configuration
   • Example access token configuration
   • Example refresh token configuration

   For example, specify the properties for the authentication method in a client.properties file.

3. Input the client properties for OAUTH 2.0 authentication into the Java client code.

   Example showing input of client properties

   Properties props = new Properties();
   try (FileReader reader = new FileReader("client.properties", StandardCharsets.UTF_8)) {
     props.load(reader);
   }

4. Verify that the Kafka client can access the Kafka brokers.

Procedure

1. Create a client secret and mount it to the component as an environment variable.

   For example, here we are creating a client Secret for the Kafka Bridge:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Secret
   metadata:
    name: my-bridge-oauth
   type: Opaque
   data:
    clientSecret: MGQ1OTRmMzYtZTllZS00MDY2LWI5OGEtMTM5MzM2NjdlZjQw 1

   1

   The clientSecret key must be in base64 format.

2. Create or edit the resource for the Kafka component so that OAuth 2.0 authentication is configured for the authentication property.

   For OAuth 2.0 authentication, you can use the following options:

   • Client ID and secret
   • Client ID and client assertion
   • Client ID and refresh token
   • Access token
   • Username and password
   • TLS

   For example, here OAuth 2.0 is assigned to the Kafka Bridge client using a client ID and secret, and TLS:

   Example OAuth 2.0 authentication configuration using the client secret

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaBridge
   metadata:
     name: my-bridge
   spec:
     # ...
     authentication:
       type: oauth 1
       tokenEndpointUri: https://<auth_server_address>/<path_to_token_endpoint> 2
       clientId: kafka-bridge
       clientSecret:
         secretName: my-bridge-oauth
         key: clientSecret
       tlsTrustedCertificates: 3
         - secretName: oauth-server-cert
           pattern: "*.crt"

   1

   Authentication type set to oauth.

   2

   URI of the token endpoint for authentication.

   3

   Certificates stored in X.509 format within the specified secrets for TLS connection to the authorization server.

   In this example, OAuth 2.0 is assigned to the Kafka Bridge client using a client ID and the location of a client assertion file, with TLS to connect to the authorization server:

   Example OAuth 2.0 authentication configuration using client assertion

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaBridge
   metadata:
     name: my-bridge
   spec:
     # ...
     authentication:
       type: oauth
       tokenEndpointUri: https://<auth_server_address>/<path_to_token_endpoint>
       clientId: kafka-bridge
       clientAssertionLocation: /var/run/secrets/sso/assertion 1
       tlsTrustedCertificates:
         - secretName: oauth-server-cert
           pattern: "*.crt"

   1

   Filesystem path to the client assertion file used for authenticating the client. This file is typically added to the deployed pod by an external operator service. Alternatively, use clientAssertion to refer to a secret containing the client assertion value.

   Here, OAuth 2.0 is assigned to the Kafka Bridge client using a service account token:

   Example OAuth 2.0 authentication configuration using the service account token

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaBridge
   metadata:
     name: my-bridge
   spec:
     # ...
     authentication:
       type: oauth
       accessTokenLocation: /var/run/secrets/kubernetes.io/serviceaccount/token 1

   1

   Path to the service account token file location.

   Depending on how you apply OAuth 2.0 authentication, and the type of authorization server, there are additional configuration options you can use:

   Additional configuration options

   # ...
   spec:
     # ...
     authentication:
       # ...
       disableTlsHostnameVerification: true 1
       accessTokenIsJwt: false 2
       scope: any 3
       audience: kafka 4
       connectTimeoutSeconds: 60 5
       readTimeoutSeconds: 60 6
       httpRetries: 2 7
       httpRetryPauseMs: 300 8
       includeAcceptHeader: false 9

   1

   (Optional) Disable TLS hostname verification. Default is false.

   2

   If you are using opaque tokens, you can apply accessTokenIsJwt: false so that access tokens are not treated as JWT tokens.

   3

   (Optional) The scope for requesting the token from the token endpoint. An authorization server may require a client to specify the scope. In this case it is any.

   4

   (Optional) The audience for requesting the token from the token endpoint. An authorization server may require a client to specify the audience. In this case it is kafka.

   5

   (Optional) The connect timeout in seconds when connecting to the authorization server. The default value is 60.

   6

   (Optional) The read timeout in seconds when connecting to the authorization server. The default value is 60.

   7

   (Optional) The maximum number of times to retry a failed HTTP request to the authorization server. The default value is 0, meaning that no retries are performed. To use this option effectively, consider reducing the timeout times for the connectTimeoutSeconds and readTimeoutSeconds options. However, note that retries may prevent the current worker thread from being available to other requests, and if too many requests stall, it could make the Kafka broker unresponsive.

   8

   (Optional) The time to wait before attempting another retry of a failed HTTP request to the authorization server. By default, this time is set to zero, meaning that no pause is applied. This is because many issues that cause failed requests are per-request network glitches or proxy issues that can be resolved quickly. However, if your authorization server is under stress or experiencing high traffic, you may want to set this option to a value of 100 ms or more to reduce the load on the server and increase the likelihood of successful retries.

   9

   (Optional) Some authorization servers have issues with client sending Accept: application/json header. By setting includeAcceptHeader: false the header will not be sent. Default is true.

3. Apply the changes to the resource configuration of the component.

4. Check the update in the logs or by watching the pod state transitions:

   oc logs -f ${POD_NAME} -c ${CONTAINER_NAME}
   oc get pod -w

   The rolling updates configure the component for interaction with Kafka brokers using OAuth 2.0 authentication.

Procedure

1. Install the Red Hat build of Keycloak server using the Red Hat build of Keycloak Operator as described in Getting Started in the Red Hat build of Keycloak documentation.
2. Wait until the Red Hat build of Keycloak instance is running.

3. Get the external hostname to be able to access the Admin Console.

   NS=sso
   oc get ingress keycloak -n $NS

   In this example, we assume the Red Hat build of Keycloak server is running in the sso namespace.

4. Get the password for the admin user.

   oc get -n $NS pod keycloak-0 -o yaml | less

   The password is stored as a secret, so get the configuration YAML file for the Red Hat build of Keycloak instance to identify the name of the secret (secretKeyRef.name).

5. Use the name of the secret to obtain the clear text password.

   SECRET_NAME=credential-keycloak
   oc get -n $NS secret $SECRET_NAME -o yaml | grep PASSWORD | awk '{print $2}' | base64 -D

   In this example, we assume the name of the secret is credential-keycloak.

6. Log in to the Admin Console with the username admin and the password you obtained.

   Use https://HOSTNAME to access the Kubernetes Ingress.

   You can now upload the example realm to Red Hat build of Keycloak using the Admin Console.

7. Click Add Realm to import the example realm.

8. Add the examples/security/keycloak-authorization/kafka-authz-realm.json file, and then click Create.

   You now have kafka-authz as your current realm in the Admin Console.

   The default view displays the Master realm.

9. In the Red Hat build of Keycloak Admin Console, go to Clients > kafka > Authorization > Settings and check that Decision Strategy is set to Affirmative.

   An affirmative policy means that at least one policy must be satisfied for a client to access the Kafka cluster.

10. In the Red Hat build of Keycloak Admin Console, go to Groups, Users, Roles and Clients to view the realm configuration.

    Groups

    Groups organize users and set permissions. Groups can be linked to an LDAP identity provider and used to compartmentalize users, such as by region or department. Users can be added to groups through a custom LDAP interface to manage permissions for Kafka resources. For more information on using KeyCloak’s LDAP provider, see the guide to Keycloak Server Administration.

    Users

    Users define individual users. In this example, alice (in the ClusterManager group) and bob (in ClusterManager-my-cluster) are created. Users can also be stored in an LDAP identity provider.

    Roles

    Roles assign specific permissions to users or clients. Roles function like tags to give users certain rights. While roles cannot be stored in LDAP, you can add Red Hat build of Keycloak roles to groups to combine both roles and permissions.

    Clients

    Clients define configurations for Kafka interactions.

    • The kafka client handles OAuth 2.0 token validation for brokers and contains authorization policies (which require authorization to be enabled).
    • The kafka-cli client is used by command line tools to obtain access or refresh tokens.
    • team-a-client and team-b-client represent services with partial access to specific Kafka topics.

11. In the Red Hat build of Keycloak Admin Console, go to Authorization > Permissions to see the granted permissions that use the resources and policies defined for the realm.

    For example, the kafka client has the following permissions:

    Dev Team A can write to topics that start with x_ on any cluster
    Dev Team B can read from topics that start with x_ on any cluster
    Dev Team B can update consumer group offsets that start with x_ on any cluster
    ClusterManager of my-cluster Group has full access to cluster config on my-cluster
    ClusterManager of my-cluster Group has full access to consumer groups on my-cluster
    ClusterManager of my-cluster Group has full access to topics on my-cluster

    Dev Team A

    The Dev Team A realm role can write to topics that start with x_ on any cluster. This combines a resource called Topic:x_*, Describe and Write scopes, and the Dev Team A policy. The Dev Team A policy matches all users that have a realm role called Dev Team A.

    Dev Team B

    The Dev Team B realm role can read from topics that start with x_ on any cluster. This combines Topic:x_*, Group:x_* resources, Describe and Read scopes, and the Dev Team B policy. The Dev Team B policy matches all users that have a realm role called Dev Team B. Matching users and clients have the ability to read from topics, and update the consumed offsets for topics and consumer groups that have names starting with x_.

Procedure

1. Use the hostname of the Red Hat build of Keycloak instance you deployed to prepare a truststore certificate for Kafka brokers to communicate with the Red Hat build of Keycloak server.

   SSO_HOST=SSO-HOSTNAME
   SSO_HOST_PORT=$SSO_HOST:443
   STOREPASS=storepass
   
   echo "Q" | openssl s_client -showcerts -connect $SSO_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/sso.pem

   The certificate is required as Kubernetes Ingress is used to make a secure (HTTPS) connection.

   Usually there is not one single certificate, but a certificate chain. You only have to provide the top-most issuer CA, which is listed last in the /tmp/sso.pem file. You can extract it manually or using the following commands:

   Example command to extract the top CA certificate in a certificate chain

   split -p "-----BEGIN CERTIFICATE-----" sso.pem sso-
   for f in $(ls sso-*); do mv $f $f.pem; done
   cp $(ls sso-* | sort -r | head -n 1) sso-ca.crt

   Note

   A trusted CA certificate is normally obtained from a trusted source, and not by using the openssl command.

2. Deploy the certificate to OpenShift as a secret.

   oc create secret generic oauth-server-cert --from-file=/tmp/sso-ca.crt -n $NS

3. Set the hostname as an environment variable

   SSO_HOST=SSO-HOSTNAME

4. Create and deploy the example Kafka cluster.

   cat examples/security/keycloak-authorization/kafka-ephemeral-oauth-single-keycloak-authz.yaml | sed -E 's#\${SSO_HOST}'"#$SSO_HOST#" | oc create -n $NS -f -

Procedure

1. Run a new interactive pod container using the Kafka image to connect to a running Kafka broker.

   NS=sso
   oc run -ti --restart=Never --image=registry.redhat.io/amq-streams/kafka-39-rhel9:2.9.0 kafka-cli -n $NS -- /bin/sh

   Note

   If oc times out waiting on the image download, subsequent attempts may result in an AlreadyExists error.

2. Attach to the pod container.

   oc attach -ti kafka-cli -n $NS

3. Use the hostname of the Red Hat build of Keycloak instance to prepare a certificate for client connection using TLS.

   SSO_HOST=SSO-HOSTNAME
   SSO_HOST_PORT=$SSO_HOST:443
   STOREPASS=storepass
   
   echo "Q" | openssl s_client -showcerts -connect $SSO_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/sso.pem

   Usually there is not one single certificate, but a certificate chain. You only have to provide the top-most issuer CA, which is listed last in the /tmp/sso.pem file. You can extract it manually or using the following command:

   Example command to extract the top CA certificate in a certificate chain

   split -p "-----BEGIN CERTIFICATE-----" sso.pem sso-
   for f in $(ls sso-*); do mv $f $f.pem; done
   cp $(ls sso-* | sort -r | head -n 1) sso-ca.crt

   Note

   A trusted CA certificate is normally obtained from a trusted source, and not by using the openssl command.

4. Create a truststore for TLS connection to the Kafka brokers.

   keytool -keystore /tmp/truststore.p12 -storetype pkcs12 -alias sso -storepass $STOREPASS -import -file /tmp/sso-ca.crt -noprompt

5. Use the Kafka bootstrap address as the hostname of the Kafka broker and the tls listener port (9093) to prepare a certificate for the Kafka broker.

   KAFKA_HOST_PORT=my-cluster-kafka-bootstrap:9093
   STOREPASS=storepass
   
   echo "Q" | openssl s_client -showcerts -connect $KAFKA_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/my-cluster-kafka.pem

   The obtained .pem file is usually not one single certificate, but a certificate chain. You only have to provide the top-most issuer CA, which is listed last in the /tmp/my-cluster-kafka.pem file. You can extract it manually or using the following command:

   Example command to extract the top CA certificate in a certificate chain

   split -p "-----BEGIN CERTIFICATE-----" /tmp/my-cluster-kafka.pem kafka-
   for f in $(ls kafka-*); do mv $f $f.pem; done
   cp $(ls kafka-* | sort -r | head -n 1) my-cluster-kafka-ca.crt

   Note

   A trusted CA certificate is normally obtained from a trusted source, and not by using the openssl command. For this example we assume the client is running in a pod in the same namespace where the Kafka cluster was deployed. If the client is accessing the Kafka cluster from outside the OpenShift cluster, you would have to first determine the bootstrap address. In that case you can also get the cluster certificate directly from the OpenShift secret, and there is no need for openssl. For more information, see Chapter 15, Setting up client access to a Kafka cluster.

6. Add the certificate for the Kafka broker to the truststore.

   keytool -keystore /tmp/truststore.p12 -storetype pkcs12 -alias my-cluster-kafka -storepass $STOREPASS -import -file /tmp/my-cluster-kafka-ca.crt -noprompt

   Keep the session open to check authorized access.

Setting up client and admin user configuration

1. Prepare a Kafka configuration file with authentication properties for the team-a-client client.

   SSO_HOST=SSO-HOSTNAME
   
   cat > /tmp/team-a-client.properties << EOF
   security.protocol=SASL_SSL
   ssl.truststore.location=/tmp/truststore.p12
   ssl.truststore.password=$STOREPASS
   ssl.truststore.type=PKCS12
   sasl.mechanism=OAUTHBEARER
   sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.client.id="team-a-client" \
     oauth.client.secret="team-a-client-secret" \
     oauth.ssl.truststore.location="/tmp/truststore.p12" \
     oauth.ssl.truststore.password="$STOREPASS" \
     oauth.ssl.truststore.type="PKCS12" \
     oauth.token.endpoint.uri="https://$SSO_HOST/realms/kafka-authz/protocol/openid-connect/token" ;
   sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
   EOF

   The SASL OAUTHBEARER mechanism is used. This mechanism requires a client ID and client secret, which means the client first connects to the Red Hat build of Keycloak server to obtain an access token. The client then connects to the Kafka broker and uses the access token to authenticate.

2. Prepare a Kafka configuration file with authentication properties for the team-b-client client.

   cat > /tmp/team-b-client.properties << EOF
   security.protocol=SASL_SSL
   ssl.truststore.location=/tmp/truststore.p12
   ssl.truststore.password=$STOREPASS
   ssl.truststore.type=PKCS12
   sasl.mechanism=OAUTHBEARER
   sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.client.id="team-b-client" \
     oauth.client.secret="team-b-client-secret" \
     oauth.ssl.truststore.location="/tmp/truststore.p12" \
     oauth.ssl.truststore.password="$STOREPASS" \
     oauth.ssl.truststore.type="PKCS12" \
     oauth.token.endpoint.uri="https://$SSO_HOST/realms/kafka-authz/protocol/openid-connect/token" ;
   sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
   EOF

3. Authenticate admin user alice by using curl and performing a password grant authentication to obtain a refresh token.

   USERNAME=alice
   PASSWORD=alice-password
   
   GRANT_RESPONSE=$(curl -X POST "https://$SSO_HOST/realms/kafka-authz/protocol/openid-connect/token" -H 'Content-Type: application/x-www-form-urlencoded' -d "grant_type=password&username=$USERNAME&password=$PASSWORD&client_id=kafka-cli&scope=offline_access" -s -k)
   
   REFRESH_TOKEN=$(echo $GRANT_RESPONSE | awk -F "refresh_token\":\"" '{printf $2}' | awk -F "\"" '{printf $1}')

   The refresh token is an offline token that is long-lived and does not expire.

4. Prepare a Kafka configuration file with authentication properties for the admin user alice.

   cat > /tmp/alice.properties << EOF
   security.protocol=SASL_SSL
   ssl.truststore.location=/tmp/truststore.p12
   ssl.truststore.password=$STOREPASS
   ssl.truststore.type=PKCS12
   sasl.mechanism=OAUTHBEARER
   sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.refresh.token="$REFRESH_TOKEN" \
     oauth.client.id="kafka-cli" \
     oauth.ssl.truststore.location="/tmp/truststore.p12" \
     oauth.ssl.truststore.password="$STOREPASS" \
     oauth.ssl.truststore.type="PKCS12" \
     oauth.token.endpoint.uri="https://$SSO_HOST/realms/kafka-authz/protocol/openid-connect/token" ;
   sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
   EOF

   The kafka-cli public client is used for the oauth.client.id in the sasl.jaas.config. Since it’s a public client it does not require a secret. The client authenticates with the refresh token that was authenticated in the previous step. The refresh token requests an access token behind the scenes, which is then sent to the Kafka broker for authentication.

1. Write to topic my-topic.

   bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic my-topic \
     --producer.config=/tmp/team-a-client.properties
   First message

   This request returns a Not authorized to access topics: [my-topic] error.

   team-a-client has a Dev Team A role that gives it permission to perform any supported actions on topics that start with a_, but can only write to topics that start with x_. The topic named my-topic matches neither of those rules.

2. Write to topic a_messages.

   bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic a_messages \
     --producer.config /tmp/team-a-client.properties
   First message
   Second message

   Messages are produced to Kafka successfully.

3. Press CTRL+C to exit the CLI application.

4. Check the Kafka container log for a debug log of Authorization GRANTED for the request.

   oc logs my-cluster-kafka-0 -f -n $NS

1. Fetch messages from topic a_messages.

   bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic a_messages \
     --from-beginning --consumer.config /tmp/team-a-client.properties

   The request returns an error because the Dev Team A role for team-a-client only has access to consumer groups that have names starting with a_.

2. Update the team-a-client properties to specify the custom consumer group it is permitted to use.

   bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic a_messages \
     --from-beginning --consumer.config /tmp/team-a-client.properties --group a_consumer_group_1

   The consumer receives all the messages from the a_messages topic.

1. List topics.

   bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties --list

   The a_messages topic is returned.

2. List consumer groups.

   bin/kafka-consumer-groups.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties --list

   The a_consumer_group_1 consumer group is returned.

   Fetch details on the cluster configuration.

   bin/kafka-configs.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties \
     --entity-type brokers --describe --entity-default

   The request returns an error because the operation requires cluster level permissions that team-a-client does not have.

1. Write to topic a_messages.

   bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic a_messages \
     --producer.config /tmp/team-b-client.properties
   Message 1

   This request returns a Not authorized to access topics: [a_messages] error.

2. Write to topic b_messages.

   bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic b_messages \
     --producer.config /tmp/team-b-client.properties
   Message 1
   Message 2
   Message 3

   Messages are produced to Kafka successfully.

3. Write to topic x_messages.

   bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --producer.config /tmp/team-b-client.properties
   Message 1

   A Not authorized to access topics: [x_messages] error is returned, The team-b-client can only read from topic x_messages.

4. Write to topic x_messages using team-a-client.

   bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --producer.config /tmp/team-a-client.properties
   Message 1

   This request returns a Not authorized to access topics: [x_messages] error. The team-a-client can write to the x_messages topic, but it does not have a permission to create a topic if it does not yet exist. Before team-a-client can write to the x_messages topic, an admin power user must create it with the correct configuration, such as the number of partitions and replicas.

1. Create the x_messages topic as alice.

   bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/alice.properties \
     --topic x_messages --create --replication-factor 1 --partitions 1

   The topic is created successfully.

2. List all topics as alice.

   bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/alice.properties --list
   bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties --list
   bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-b-client.properties --list

   Admin user alice can list all the topics, whereas team-a-client and team-b-client can only list the topics they have access to.

   The Dev Team A and Dev Team B roles both have Describe permission on topics that start with x_, but they cannot see the other team’s topics because they do not have Describe permissions on them.

3. Use the team-a-client to produce messages to the x_messages topic:

   bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --producer.config /tmp/team-a-client.properties
   Message 1
   Message 2
   Message 3

   As alice created the x_messages topic, messages are produced to Kafka successfully.

4. Use the team-b-client to produce messages to the x_messages topic.

   bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --producer.config /tmp/team-b-client.properties
   Message 4
   Message 5

   This request returns a Not authorized to access topics: [x_messages] error.

5. Use the team-b-client to consume messages from the x_messages topic:

   bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --from-beginning --consumer.config /tmp/team-b-client.properties --group x_consumer_group_b

   The consumer receives all the messages from the x_messages topic.

6. Use the team-a-client to consume messages from the x_messages topic.

   bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --from-beginning --consumer.config /tmp/team-a-client.properties --group x_consumer_group_a

   This request returns a Not authorized to access topics: [x_messages] error.

7. Use the team-a-client to consume messages from a consumer group that begins with a_.

   bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --from-beginning --consumer.config /tmp/team-a-client.properties --group a_consumer_group_a

   This request returns a Not authorized to access topics: [x_messages] error.

   Dev Team A has no Read access on topics that start with a x_.

8. Use alice to produce messages to the x_messages topic.

   bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --from-beginning --consumer.config /tmp/alice.properties

   Messages are produced to Kafka successfully.

   alice can read from or write to any topic.

9. Use alice to read the cluster configuration.

   bin/kafka-configs.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/alice.properties \
     --entity-type brokers --describe --entity-default

   The cluster configuration for this example is empty.