Chapter 7. Setting up client access to a Kafka cluster

Procedure

1. Deploy a Kafka producer.

   oc run kafka-producer -ti --image=registry.redhat.io/amq7/amq-streams-kafka-33-rhel8:2.3.0 --rm=true --restart=Never -- bin/kafka-console-producer.sh --bootstrap-server cluster-name-kafka-bootstrap:9092 --topic my-topic

2. Type a message into the console where the producer is running.
3. Press Enter to send the message.

4. Deploy a Kafka consumer.

   oc run kafka-consumer -ti --image=registry.redhat.io/amq7/amq-streams-kafka-33-rhel8:2.3.0 --rm=true --restart=Never -- bin/kafka-console-consumer.sh --bootstrap-server cluster-name-kafka-bootstrap:9092 --topic my-topic --from-beginning

5. Confirm that you see the incoming messages in the consumer console.

1. An external listener is configured for the Kafka cluster, with TLS encryption and mTLS authentication, and Kafka simple authorization enabled.
2. A KafkaUser is created for the client, with mTLS authentication, and Access Control Lists (ACLs) defined for simple authorization.

Procedure

1. Configure the Kafka cluster with a Kafka listener.

   • Define the authentication required to access the Kafka broker through the listener.

   • Enable authorization on the Kafka broker.

     Example listener configuration

     apiVersion: kafka.strimzi.io/v1beta2
     kind: Kafka
     metadata:
       name: my-cluster
       namespace: myproject
     spec:
       kafka:
         # ...
         listeners: 1
         - name: external 2
           port: 9094 3
           type: <listener_type> 4
           tls: true 5
           authentication:
             type: tls 6
           configuration: 7
             #...
         authorization: 8
           type: simple
           superUsers:
             - super-user-name 9
       # ...

     1

     Configuration options for enabling external listeners are described in the Generic Kafka listener schema reference.

     2

     Name to identify the listener. Must be unique within the Kafka cluster.

     3

     Port number used by the listener inside Kafka. The port number has to be unique within a given Kafka cluster. Allowed port numbers are 9092 and higher with the exception of ports 9404 and 9999, which are already used for Prometheus and JMX. Depending on the listener type, the port number might not be the same as the port number that connects Kafka clients.

     4

     External listener type specified as route, loadbalancer, nodeport or ingress. An internal listener is specified as internal or cluster-ip.

     5

     Required. TLS encryption on the listener. For route and ingress type listeners it must be set to true. For mTLS authentication, also use the authentication property.

     6

     Client authentication mechanism on the listener. For server and client authentication using mTLS, you specify tls: true and authentication.type: tls.

     7

     (Optional) Depending on the requirements of the listener type, you can specify additional listener configuration.

     8

     Authorization specified as simple, which uses the AclAuthorizer Kafka plugin.

     9

     (Optional) Super users can access all brokers regardless of any access restrictions defined in ACLs.

     Warning

     An OpenShift Route address comprises the name of the Kafka cluster, the name of the listener, and the name of the namespace it is created in. For example, my-cluster-kafka-listener1-bootstrap-myproject (CLUSTER-NAME-kafka-LISTENER-NAME-bootstrap-NAMESPACE). If you are using a route listener type, be careful that the whole length of the address does not exceed a maximum limit of 63 characters.

2. Create or update the Kafka resource.

   oc apply -f <kafka_configuration_file>

   The Kafka cluster is configured with a Kafka broker listener using mTLS authentication.

   A service is created for each Kafka broker pod.

   A service is created to serve as the bootstrap address for connection to the Kafka cluster.

   A service is also created as the external bootstrap address for external connection to the Kafka cluster using nodeport listeners.

   The cluster CA certificate to verify the identity of the kafka brokers is also created in the secret <cluster_name>-cluster-ca-cert.

   Note

   If you scale your Kafka cluster while using external listeners, it might trigger a rolling update of all Kafka brokers. This depends on the configuration.

3. Retrieve the bootstrap address you can use to access the Kafka cluster from the status of the Kafka resource.

   oc get kafka <kafka_cluster_name> -o=jsonpath='{.status.listeners[?(@.name=="<listener_name>")].bootstrapServers}{"\n"}'

   For example:

   oc get kafka my-cluster -o=jsonpath='{.status.listeners[?(@.name=="external")].bootstrapServers}{"\n"}'

   Use the bootstrap address in your Kafka client to connect to the Kafka cluster.

4. Create or modify a user representing the client that requires access to the Kafka cluster.

   • Specify the same authentication type as the Kafka listener.

   • Specify the authorization ACLs for simple authorization.

     Example user configuration

     apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaUser
     metadata:
       name: my-user
       labels:
         strimzi.io/cluster: my-cluster 1
     spec:
       authentication:
         type: tls 2
       authorization:
         type: simple
         acls: 3
           - resource:
               type: topic
               name: my-topic
               patternType: literal
             operations:
               - Describe
               - Read
           - resource:
               type: group
               name: my-group
               patternType: literal
             operations:
               - Read

     1

     The label must match the label of the Kafka cluster.

     2

     Authentication specified as mutual tls.

     3

     Simple authorization requires an accompanying list of ACL rules to apply to the user. The rules define the operations allowed on Kafka resources based on the username (my-user).

5. Create or modify the KafkaUser resource.

   oc apply -f USER-CONFIG-FILE

   The user is created, as well as a secret with the same name as the KafkaUser resource. The secret contains a public and private key for mTLS authentication.

   Example secret

   apiVersion: v1
   kind: Secret
   metadata:
     name: my-user
     labels:
       strimzi.io/kind: KafkaUser
       strimzi.io/cluster: my-cluster
   type: Opaque
   data:
     ca.crt: <public_key> # Public key of the clients CA
     user.crt: <user_certificate> # Public key of the user
     user.key: <user_private_key> # Private key of the user
     user.p12: <store> # PKCS #12 store for user certificates and keys
     user.password: <password_for_store> # Protects the PKCS #12 store

6. Extract the cluster CA certificate from the <cluster_name>-cluster-ca-cert secret of the Kafka cluster.

   oc get secret <cluster_name>-cluster-ca-cert -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt

7. Extract the user CA certificate from the <user_name> secret.

   oc get secret <user_name> -o jsonpath='{.data.user\.crt}' | base64 -d > user.crt

8. Extract the private key of the user from the <user_name> secret.

   oc get secret <user_name> -o jsonpath='{.data.user\.key}' | base64 -d > user.key

9. Configure your client with the bootstrap address hostname and port for connecting to the Kafka cluster:

   props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "<hostname>:<port>");

10. Configure your client with the truststore credentials to verify the identity of the Kafka cluster.

    Specify the public cluster CA certificate.

    Example truststore configuration

    props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
    props.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "PEM");
    props.put(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG, "<ca.crt_file_content>");

    SSL is the specified security protocol for mTLS authentication. Specify SASL_SSL for SCRAM-SHA-512 authentication over TLS. PEM is the file format of the truststore.

11. Configure your client with the keystore credentials to verify the user when connecting to the Kafka cluster.

    Specify the public certificate and private key.

    Example keystore configuration

    props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
    props.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "PEM");
    props.put(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG, "<user.crt_file_content>");
    props.put(SslConfigs.SSL_KEYSTORE_KEY_CONFIG, "<user.key_file_content>");

    Add the keystore certificate and the private key directly to the configuration. Add as a single-line format. Between the BEGIN CERTIFICATE and END CERTIFICATE delimiters, start with a newline character (\n). End each line from the original certificate with \n too.

    Example keystore configuration

    props.put(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG, "-----BEGIN CERTIFICATE----- \n<user_certificate_content_line_1>\n<user_certificate_content_line_n>\n-----END CERTIFICATE---");
    props.put(SslConfigs.SSL_KEYSTORE_KEY_CONFIG, "----BEGIN PRIVATE KEY-----\n<user_key_content_line_1>\n<user_key_content_line_n>\n-----END PRIVATE KEY-----");

    Additional resources

    • Listener authentication options
    • Kafka authorization options
    • If you are using an authorization server, you can use token-based OAuth 2.0 authentication and OAuth 2.0 authorization.