# Module 1: Deploy the Confluent Platform Demo Environment
`cp-demo` is a Docker environment and has all services running on one host.
It is meant exclusively to easily demo Confluent Platform.
Note that in production, you should not deploy all Confluent Platform services on a single host as shown in `cp-demo`.
Also, in production, Confluent Control Center should be deployed with a valid license in
[reduced infrastructure mode](https://docs.confluent.io/control-center/current/installation/configuration.html#mode-settings) in conjunction with
[Confluent Health+](../../health-plus/index.md#health-plus) for monitoring.
If you prefer non-Docker examples, go to [confluentinc/examples GitHub repository](https://github.com/confluentinc/examples).
After you complete the guided tutorial that follows, you can apply the concepts you learn to build your
own event streaming pipeline in Confluent Cloud, a fully managed, cloud-native event streaming platform powered
by Kafka. When you sign up for [Confluent Cloud](https://confluent.cloud), use the promo code `CPDEMO50`
to receive an additional $50 free usage ([details](https://www.confluent.io/confluent-cloud-promo-disclaimer)).
## Prerequisites
This example has been validated with:
- Docker engine version 20.10.12
- Docker Compose version 2.4.1
- Java version 11.0.8
- Ubuntu 18.04
- OpenSSL 1.1.1
- git
- curl
- jq
## Setup
You can run this demo locally with Docker or in a cloud IDE with [Gitpod](https://gitpod.io/#https://github.com/confluentinc/cp-demo/tree/current-post).
### Docker
This demo has been validated with Docker as described in [Prerequisites](#cp-demo-prereqs).
If you are using Docker, follow these steps:
1. In Docker’s advanced [settings](https://docs.docker.com/docker-for-mac/#advanced), increase the memory dedicated to Docker to at least eight GB (default is two GB) and ensure Docker is allocated at least two CPU cores.
2. Clone the [confluentinc/cp-demo GitHub repository](https://github.com/confluentinc/cp-demo):
```bash
git clone https://github.com/confluentinc/cp-demo
```
3. Navigate to the `cp-demo` directory and switch to the latest 8.0.x release branch. For example:
```bash
cd cp-demo
git checkout 8.1.0-post
```
### Gitpod
If you are using Gitpod, follow these steps:
This demo is enabled to run with Gitpod, but support for the Gitpod workflow is best effort from the [community](https://github.com/confluentinc/cp-demo/issues).
If you are using [Gitpod](https://gitpod.io/#https://github.com/confluentinc/cp-demo/tree/current-post), the demo will automatically start after the Gitpod workspace is ready. `VIZ=false` is used to save system resources.
Log in to Confluent Control Center (port `9021`) by clicking the `Open Browser` option in the pop-up:
or by selecting `Remote Explorer` on the left sidebar and then clicking the `Open Browser` option corresponding to the port you want to connect to:
## Start
Within the `cp-demo` directory, there is a single [script](https://github.com/confluentinc/cp-demo/tree/latest/scripts/start.sh) that runs the `cp-demo` workflow end-to-end.
It generates the keys and certificates, brings up the Docker containers, and configures and validates the environment.
You can run it with optional settings:
- `CLEAN`: controls whether certificates are regenerated
- `C3_KSQLDB_HTTPS`: controls whether Confluent Control Center and ksqlDB server use `HTTP` or `HTTPS` (default: `false` for `HTTP`). This option is not supported with [Gitpod](https://gitpod.io/#https://github.com/confluentinc/cp-demo/tree/current-post).
- `VIZ`: enables Elasticsearch and Kibana (default: `true`)
1. To run `cp-demo` the first time with defaults, run the following command. The very first run downloads all the required Docker images (~15 minutes) and sets up the environment (~five minutes).
```bash
./scripts/start.sh
```
2. On subsequent runs, if you have not deleted the generated certificates and the locally built Connect image, they will be reused. To force them to be rebuilt, you can set `CLEAN=true`.
```bash
CLEAN=true ./scripts/start.sh
```
3. `cp-demo` supports access to the Confluent Control Center GUI via either `http://` (the default) or secure `https://`, the latter employing a self-signed CA and certificates generated during deployment. In order to run ksqlDB queries from Confluent Control Center later in this tutorial, both ksqlDB and Confluent Control Center must be running in either `http` or `https` [mode](/platform/current/ksqldb/integrate-ksql-with-confluent-control-center.html#configuration-settings-for-ksqldb-and-c3-short). To run `cp-demo` in `https` mode, set `C3_KSQLDB_HTTPS=true` when starting `cp-demo`:
```bash
C3_KSQLDB_HTTPS=true ./scripts/start.sh
```
4. Elasticsearch and Kibana increase localhost memory requirements for `cp-demo`. For users who want to run `cp-demo` with a smaller memory footprint, opt-out of these components by setting `VIZ=false` when starting `cp-demo`.
```bash
VIZ=false ./scripts/start.sh
```
5. After the start script completes, run through the pre-flight checks below and follow the guided tutorial through this on-premises deployment.
## Pre-flight checks
Before going through the tutorial, check that the environment has started correctly.
If any of these pre-flight checks fails, consult the [Troubleshooting the scripted demo](teardown.md#cp-demo-troubleshooting) section.
1. Verify the status of the Docker containers show `Up` state.
```bash
docker compose ps
```
Your output should resemble:
```text
NAME IMAGE COMMAND SERVICE CREATED STATUS PORTS
alertmanager confluentinc/cp-enterprise-alertmanager:2.3.0 "alertmanager-start" alertmanager 39 minutes ago Up 39 minutes 0.0.0.0:9093->9093/tcp, [::]:9093->9093/tcp
connect localbuild/connect:8.1.0-8.1.0 "/etc/confluent/dock…" connect 39 minutes ago Up 39 minutes (healthy) 0.0.0.0:8083->8083/tcp, [::]:8083->8083/tcp
control-center confluentinc/cp-enterprise-control-center-next-gen:2.3.0 "/etc/confluent/dock…" control-center 39 minutes ago Up 39 minutes (healthy) 0.0.0.0:9021-9022->9021-9022/tcp, [::]:9021-9022->9021-9022/tcp
elasticsearch docker.elastic.co/elasticsearch/elasticsearch:8.19.4 "/bin/tini -- /usr/l…" elasticsearch 21 minutes ago Up 21 minutes (healthy) 0.0.0.0:9200->9200/tcp, [::]:9200->9200/tcp, 0.0.0.0:9300->9300/tcp, [::]:9300->9300/tcp
kafka1 confluentinc/cp-server:8.1.0 "bash -c 'if [ ! -f …" kafka1 40 minutes ago Up 40 minutes (healthy) 0.0.0.0:8091->8091/tcp, [::]:8091->8091/tcp, 0.0.0.0:9091->9091/tcp, [::]:9091->9091/tcp, 0.0.0.0:10091->10091/tcp, [::]:10091->10091/tcp, 0.0.0.0:11091->11091/tcp, [::]:11091->11091/tcp, 0.0.0.0:12091->12091/tcp, [::]:12091->12091/tcp
kafka2 confluentinc/cp-server:8.1.0 "bash -c 'if [ ! -f …" kafka2 40 minutes ago Up 40 minutes (healthy) 0.0.0.0:8092->8092/tcp, [::]:8092->8092/tcp, 0.0.0.0:9092->9092/tcp, [::]:9092->9092/tcp, 0.0.0.0:10092->10092/tcp, [::]:10092->10092/tcp, 0.0.0.0:11092->11092/tcp, [::]:11092->11092/tcp, 0.0.0.0:12092->12092/tcp, [::]:12092->12092/tcp
kibana docker.elastic.co/kibana/kibana:8.19.4 "/bin/tini -- /usr/l…" kibana 21 minutes ago Up 20 minutes (healthy) 0.0.0.0:5601->5601/tcp, [::]:5601->5601/tcp
ksqldb-cli confluentinc/cp-ksqldb-server:8.1.0 "/bin/sh" ksqldb-cli 38 minutes ago Up 38 minutes 8088/tcp
ksqldb-server confluentinc/cp-ksqldb-server:8.1.0 "/etc/confluent/dock…" ksqldb-server 39 minutes ago Up 39 minutes (healthy) 0.0.0.0:8088-8089->8088-8089/tcp, [::]:8088-8089->8088-8089/tcp
openldap osixia/openldap:1.3.0 "/container/tool/run…" openldap 40 minutes ago Up 40 minutes 389/tcp, 636/tcp
prometheus confluentinc/cp-enterprise-prometheus:2.3.0 "prometheus-start" prometheus 39 minutes ago Up 39 minutes 0.0.0.0:9090->9090/tcp, [::]:9090->9090/tcp
restproxy confluentinc/cp-kafka-rest:8.1.0 "/etc/confluent/dock…" restproxy 38 minutes ago Up 38 minutes 0.0.0.0:8086->8086/tcp, [::]:8086->8086/tcp
schemaregistry confluentinc/cp-schema-registry:8.1.0 "/etc/confluent/dock…" schemaregistry 39 minutes ago Up 39 minutes (healthy) 0.0.0.0:8085->8085/tcp, [::]:8085->8085/tcp
streams-demo cnfldemos/cp-demo-kstreams:0.0.12 "/app/start.sh" streams-demo 20 minutes ago Up 20 minutes 9092/tcp
tools cnfldemos/tools:0.3 "/bin/bash" tools 40 minutes ago Up 40 minutes
```
2. Jump to the end of the entire `cp-demo` pipeline and view the Kibana dashboard at [http://localhost:5601/app/dashboards#/view/Overview](http://localhost:5601/app/dashboards#/view/Overview) . This is a cool view and validates that the `cp-demo` start script completed successfully.
3. View the full Confluent Platform configuration in the [docker-compose.yml](https://github.com/confluentinc/cp-demo/tree/latest/docker-compose.yml) file.
4. View the Kafka Streams application configuration in the [client configuration](https://github.com/confluentinc/cp-demo/tree/latest/env_files/streams-demo.env) file, set with security parameters to the Kafka cluster and Schema Registry.
## Guided tutorial
### Log into Confluent Control Center
1. If you ran `cp-demo` with `C3_KSQLDB_HTTPS=false` (which is the default), log into the Confluent Control Center GUI from a web browser at the following URL:
```text
http://localhost:9021
```
2. If you ran `cp-demo` with `C3_KSQLDB_HTTPS=true` (not supported with [Gitpod](https://gitpod.io/#https://github.com/confluentinc/cp-demo/tree/current-post)), log into the Confluent Control Center GUI from a web browser at the following URL:
```text
https://localhost:9022
```
The browser will detect a self-signed, untrusted certificate and certificate authority, and issue a privacy warning as shown below. To proceed, accept this certificate using your browser’s process for this, which will then last for the duration of that browser session.
- **Chrome**: Click `Advanced` and when the window expands, click `Proceed to localhost (unsafe)`.
- **Safari**: Open a new private browsing window (`Shift + ⌘ + N`), click `Show Details` and when the window expands, click `visit this website`.
3. At the login screen, log into Confluent Control Center as `superUser` and password `superUser`, which has super user access to the cluster. You may also log in as [other users](https://github.com/confluentinc/cp-demo/tree/latest/scripts/security/ldap_users) to learn how each user’s view changes depending on their permissions.
### Brokers
1. Select the cluster named **Kafka Raleigh**.
2. Click **Brokers**.
3. View the status of the brokers in the cluster:
4. Click **Production** or **Consumption** to view:
* Production metrics
* Consumption metrics
* Broker uptime
* Partition replicas: Partitions, leaders, under replicated partitions, and under min in-sync replicas.
* Disk utilization
* System usage: Network pool and request pool
### Topics
Confluent Control Center can manage topics in a Kafka cluster.
1. Click **Topics**.
2. Scroll down and click the topic `wikipedia.parsed`.
3. View an overview of this topic:
- Throughput
- Partition replication status
4. View which brokers are leaders for which partitions and where all partitions reside.
5. Inspect messages for this topic in real-time.
6. View the schema for this topic. For `wikipedia.parsed`, the topic value is using a Schema registered with Schema Registry (the topic key is just a string).
7. View configuration settings for this topic.
8. Return to **Topics**, click `wikipedia.parsed.count-by-domain` to view the output topic from the Kafka Streams application.
9. Return to **Topics** view and click the **+ Add a topic** button to create a new topic in your Kafka cluster. You can also view and edit settings of Kafka topics in the cluster. Read more on Confluent Control Center [topic management](https://docs.confluent.io/control-center/current/topics/overview.html).
### Kafka Connect
This example runs two connectors:
- SSE source connector
- Elasticsearch sink connector
They are running on a Connect worker that is configured with Confluent Platform security features.
The Connect worker’s embedded producer is configured to be idempotent, exactly-once in order semantics per partition (in the event of an error that causes a producer retry, the same message—which is still sent by the producer multiple times—will only be written to the Kafka log on the broker once).
The Kafka Connect Docker container is running a custom image which has a specific set of connectors and transformations
needed by `cp-demo`. See [this Dockerfile](https://github.com/confluentinc/cp-demo/tree/latest/Dockerfile) for more details. Confluent Control Center uses the Kafka Connect API to manage multiple [connect clusters](../../connect/index.md#kafka-connect).
1. In the navigation bar, click **Connect**.
2. Select **connect1**, the name of the cluster of Connect workers.
3. Verify the connectors running in this example:
- source connector `wikipedia-sse`: view the example’s SSE source connector [configuration file](https://github.com/confluentinc/cp-demo/tree/latest/scripts/connectors/submit_wikipedia_sse_config.sh).
- sink connector `elasticsearch-ksqldb` consuming from the Kafka topic `WIKIPEDIABOT`: view the example’s Elasticsearch sink connector [configuration file](https://github.com/confluentinc/cp-demo/tree/latest/scripts/connectors/submit_elastic_sink_config.sh).
4. Click any connector name to view or modify any details of the connector configuration and custom transforms.
### ksqlDB
In this example, ksqlDB is authenticated and authorized to connect to the secured Kafka cluster, and it is already running queries as defined in the [ksqlDB command file](https://github.com/confluentinc/cp-demo/tree/latest/scripts/ksqlDB/statements.sql).
Its embedded producer is configured to be idempotent, exactly-once in order semantics per partition (in the event of an error that causes a producer retry, the same message—which is still sent by the producer multiple times—will only be written to the Kafka log on the broker once).
1. In the navigation bar, click **ksqlDB**.
2. From the list of ksqlDB applications, select `wikipedia`.
3. View the ksqlDB Flow to see the streams and tables created in the example, and how they relate to one another.
4. Use Confluent Control Center to interact with ksqlDB, or run ksqlDB CLI to get to the ksqlDB CLI prompt.
```bash
docker compose exec ksqldb-cli bash -c 'ksql -u ksqlDBUser -p ksqlDBUser http://ksqldb-server:8088'
```
5. View the existing ksqlDB streams. (If you are using the ksqlDB CLI, at the `ksql>` prompt, type `SHOW STREAMS;`)
6. Click **WIKIPEDIA** to describe the schema (fields or columns) of an existing ksqlDB stream. (If you are using the ksqlDB CLI, at the `ksql>` prompt, type `DESCRIBE WIKIPEDIA;`)
7. View the existing ksqlDB tables. (If you are using the ksqlDB CLI, at the `ksql>` prompt, type `SHOW TABLES;`). One table is called `WIKIPEDIA_COUNT_GT_1`, which counts occurrences within a [tumbling window](../../ksqldb/concepts/time-and-windows-in-ksqldb-queries.md#ksqldb-time-and-windows-tumbling-window).
8. View the existing ksqlDB queries, which are continuously running. (If you are using the ksqlDB CLI, at the `ksql>` prompt, type `SHOW QUERIES;`).
9. View messages from different ksqlDB streams and tables. Click on your stream of choice and then click **Query stream** to open the Query Editor. The editor shows a pre-populated query, like `select * from WIKIPEDIA EMIT CHANGES;`, and it shows results for newly arriving data.
10. Click **ksqlDB Editor** and run the `SHOW PROPERTIES;` statement. You can see the configured ksqlDB server properties and check these values with the [docker-compose.yml](https://github.com/confluentinc/cp-demo/tree/latest/docker-compose.yml) file.
11. The [ksqlDB processing log](../../ksqldb/reference/processing-log.md#ksqldb-reference-processing-log) captures per-record errors during processing to help developers debug their ksqlDB queries. In this example, the processing log uses mutual TLS (mTLS) authentication, as configured in the custom [log4j properties file](https://github.com/confluentinc/cp-demo/tree/latest/scripts/helper/log4j-secure.properties), to write entries into a Kafka topic. To see it in action, in the ksqlDB editor run the following “bad” query for 20 seconds:
```bash
SELECT 1/0 FROM wikipedia EMIT CHANGES;
```
No records should be returned from this query. ksqlDB writes errors into the processing log for each record. View the processing log topic `ksql-clusterksql_processing_log` with topic inspection (jump to offset 0/partition 0) or the corresponding ksqlDB stream `KSQL_PROCESSING_LOG` with the ksqlDB editor (set `auto.offset.reset=earliest`).
```bash
SELECT * FROM KSQL_PROCESSING_LOG EMIT CHANGES;
```
## Security
### Overview
All components and clients in `cp-demo` make full use of Confluent Platform’s extensive [security features](../../security/overview.md#security).
- [Role-Based Access Control (RBAC)](../../security/authorization/rbac/overview.md#rbac-overview) for authorization. Give principals access to resources using role-bindings.
#### NOTE
RBAC is powered by the [Metadata Service (MDS)](../../kafka/configure-mds/index.md#rbac-mds-config) which uses Confluent Server Authorizer to connect to an OpenLDAP directory service. This enables group-based authorization for scalable access management.
- [SSL](../../security/authentication/mutual-tls/overview.md#kafka-ssl-authentication) for encryption and mTLS for authentication. The example [automatically generates](https://github.com/confluentinc/cp-demo/tree/latest/scripts/security/certs-create.sh) SSL certificates and creates keystores, truststores, and secures them with a password.
- [HTTPS for Control Center](https://docs.confluent.io/platform/current/control-center/installation/configuration.html#https-settings).
- [HTTPS for Schema Registry](../../schema-registry/security/index.md#schemaregistry-security).
- [HTTPS for Connect](../../connect/security.md#connect-security).
You can see each component’s security configuration in the example’s [docker-compose.yml](https://github.com/confluentinc/cp-demo/tree/latest/docker-compose.yml) file.
#### NOTE
This example showcases a secure Confluent Platform for educational purposes and is not meant to be complete best practices. There are certain differences between what is shown in the example and what you should do in production:
* Authorize users only for operations that they need, instead of making all of them super users
* If the `PLAINTEXT` security protocol is used, these `ANONYMOUS` usernames should not be configured as super users
* Consider not even opening the `PLAINTEXT` port if `SSL` or `SASL_SSL` are configured
KRaft has two listener ports:
| Name | Protocol | In this example, used for … | Kraft |
|--------|-----------------|-------------------------------------------------------------------------------|---------|
| N/A | SASL/DIGEST-MD5 | Validating trial license for REST Proxy and Schema Registry. (no TLS support) | 2181 |
| N/A | mTLS | Broker communication (kafka1, kafka2) | 2182 |
Each broker has five listener ports:
| Name | Protocol | In this example, used for … | kafka1 | kafka2 |
|----------|----------------|---------------------------------------------------------------------------------------|----------|----------|
| N/A | MDS | Authorization via RBAC | 8091 | 8092 |
| INTERNAL | SASL_PLAINTEXT | CP Kafka clients (e.g. Confluent Metrics Reporter), SASL_PLAINTEXT | 9091 | 9092 |
| TOKEN | SASL_SSL | Confluent Platform service (e.g. Schema Registry) when they need to use impersonation | 10091 | 10092 |
| SSL | SSL | End clients, (e.g. stream-demo), with SSL no SASL | 11091 | 11092 |
| CLEAR | PLAINTEXT | No security, available as a backdoor; for demo and learning only | 12091 | 12092 |
End clients (non-CP clients):
- Authenticate using mTLS via the broker SSL listener.
- If they are also using Schema Registry, authenticate to Schema Registry via LDAP.
- Should never use the TOKEN listener which is meant only for internal communication between Confluent components.
#### NOTE
If you wish to authenticate clients with username and password via LDAP, you would create a new SASL PLAIN client listener with Confluent’s [LdapAuthenticateCallbackHandler](/platform/current/kafka/authentication_sasl/client-authentication-ldap.html). This is omitted from the demo for simplicity.
- See [client configuration](https://github.com/confluentinc/cp-demo/tree/latest/env_files/streams-demo.env/) used in the example by the `streams-demo` container running the Kafka Streams application `wikipedia-activity-monitor`.
### Broker listeners
1. Verify the ports on which the Kafka brokers are listening with the
following command, and they should match the table shown below:
```bash
docker compose logs kafka1 | grep "Registered broker 1"
docker compose logs kafka2 | grep "Registered broker 2"
```
2. For example only: Communicate with brokers via the PLAINTEXT port, client security configurations are not required
```bash
# CLEAR/PLAINTEXT port
docker compose exec kafka1 kafka-consumer-groups \
--list \
--bootstrap-server kafka1:12091
```
3. End clients: Communicate with brokers via the SSL port, and SSL parameters configured via the `--command-config` argument for command line tools or `--consumer.config` for kafka-console-consumer.
```bash
# SSL/SSL port
docker compose exec kafka1 kafka-consumer-groups \
--list \
--bootstrap-server kafka1:11091 \
```
4. If a client tries to communicate with brokers via the SSL port but does not specify the SSL parameters, it fails
```bash
# SSL/SSL port
docker compose exec kafka1 kafka-consumer-groups \
--list \
--bootstrap-server kafka1:11091
```
Your output should resemble:
```bash
ERROR Uncaught exception in thread 'kafka-admin-client-thread | adminclient-1': (org.apache.kafka.common.utils.KafkaThread)
java.lang.OutOfMemoryError: Java heap space
...
```
5. Communicate with brokers via the SASL_PLAINTEXT port, and SASL_PLAINTEXT parameters configured via the `--command-config` argument for command line tools or `--consumer.config` for kafka-console-consumer.
```bash
# INTERNAL/SASL_PLAIN port
docker compose exec kafka1 kafka-consumer-groups \
--list \
--bootstrap-server kafka1:9091 \
--command-config /etc/kafka/secrets/client_sasl_plain.config
```
### Authorization with RBAC
1. Verify which users are configured to be super users.
```bash
docker compose logs kafka1 | grep "super.users ="
```
Your output should resemble the following. Notice this authorizes each service name which authenticates as itself,
as well as the unauthenticated `PLAINTEXT` which authenticates as `ANONYMOUS` (for demo purposes only):
```bash
kafka1 | super.users = User:admin;User:mds;User:superUser;User:ANONYMOUS
```
2. From the Confluent Control Center UI, in the Administration menu, click the **Manage role assignments** option. Click **Assignments**, and then the Kafka cluster ID.
3. From the **Topic** list, verify that the LDAP user `appSA` is allowed to access a few topics, including any topic whose name starts with **wikipedia**. This role assignment was done during `cp-demo` startup in the [create-role-bindings.sh script](https://github.com/confluentinc/cp-demo/tree/latest/scripts/helper/create-role-bindings.sh).
4. Verify that LDAP user `appSA` (which is not a super user) can consume messages from topic `wikipedia.parsed`. Notice that it is configured to authenticate to brokers with mTLS and authenticate to Schema Registry with LDAP.
```bash
docker compose exec connect kafka-avro-console-consumer \
--bootstrap-server kafka1:11091,kafka2:11092 \
--consumer-property security.protocol=SSL \
--consumer-property ssl.truststore.location=/etc/kafka/secrets/kafka.appSA.truststore.jks \
--consumer-property ssl.truststore.password=confluent \
--consumer-property ssl.keystore.location=/etc/kafka/secrets/kafka.appSA.keystore.jks \
--consumer-property ssl.keystore.password=confluent \
--consumer-property ssl.key.password=confluent \
--property schema.registry.url=https://schemaregistry:8085 \
--property schema.registry.ssl.truststore.location=/etc/kafka/secrets/kafka.appSA.truststore.jks \
--property schema.registry.ssl.truststore.password=confluent \
--property basic.auth.credentials.source=USER_INFO \
--property basic.auth.user.info=appSA:appSA \
--group wikipedia.test \
--topic wikipedia.parsed \
--max-messages 5
```
5. Verify that LDAP user `badapp` cannot consume messages from topic `wikipedia.parsed`.
```bash
docker compose exec connect kafka-avro-console-consumer \
--bootstrap-server kafka1:11091,kafka2:11092 \
--consumer-property security.protocol=SSL \
--consumer-property ssl.truststore.location=/etc/kafka/secrets/kafka.badapp.truststore.jks \
--consumer-property ssl.truststore.password=confluent \
--consumer-property ssl.keystore.location=/etc/kafka/secrets/kafka.badapp.keystore.jks \
--consumer-property ssl.keystore.password=confluent \
--consumer-property ssl.key.password=confluent \
--property schema.registry.url=https://schemaregistry:8085 \
--property schema.registry.ssl.truststore.location=/etc/kafka/secrets/kafka.badapp.truststore.jks \
--property schema.registry.ssl.truststore.password=confluent \
--property basic.auth.credentials.source=USER_INFO \
--property basic.auth.user.info=badapp:badapp \
--group wikipedia.test \
--topic wikipedia.parsed \
--max-messages 5
```
Your output should resemble:
```bash
ERROR [Consumer clientId=consumer-wikipedia.test-1, groupId=wikipedia.test] Topic authorization failed for topics [wikipedia.parsed]
org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [wikipedia.parsed]
```
6. Create role bindings to permit `badapp` client to consume from topic `wikipedia.parsed` and its related subject in Schema Registry.
Get the Kafka cluster ID:
```none
KAFKA_CLUSTER_ID=$(curl -s https://localhost:8091/v1/metadata/id --tlsv1.2 --cacert scripts/security/snakeoil-ca-1.crt | jq -r ".id")
```
Create the role bindings:
```text
# Create the role binding for the topic ``wikipedia.parsed``
docker compose exec tools bash -c "confluent iam rbac role-binding create \
--principal User:badapp \
--role ResourceOwner \
--resource Topic:wikipedia.parsed \
--kafka-cluster-id $KAFKA_CLUSTER_ID"
# Create the role binding for the group ``wikipedia.test``
docker compose exec tools bash -c "confluent iam rbac role-binding create \
--principal User:badapp \
--role ResourceOwner \
--resource Group:wikipedia.test \
--kafka-cluster-id $KAFKA_CLUSTER_ID"
# Create the role binding for the subject ``wikipedia.parsed-value``, i.e., the topic-value (versus the topic-key)
docker compose exec tools bash -c "confluent iam rbac role-binding create \
--principal User:badapp \
--role ResourceOwner \
--resource Subject:wikipedia.parsed-value \
--kafka-cluster-id $KAFKA_CLUSTER_ID \
--schema-registry-cluster schema-registry"
```
7. Verify that LDAP user `badapp` now can consume messages from topic `wikipedia.parsed`.
```bash
docker compose exec connect kafka-avro-console-consumer \
--bootstrap-server kafka1:11091,kafka2:11092 \
--consumer-property security.protocol=SSL \
--consumer-property ssl.truststore.location=/etc/kafka/secrets/kafka.badapp.truststore.jks \
--consumer-property ssl.truststore.password=confluent \
--consumer-property ssl.keystore.location=/etc/kafka/secrets/kafka.badapp.keystore.jks \
--consumer-property ssl.keystore.password=confluent \
--consumer-property ssl.key.password=confluent \
--property schema.registry.url=https://schemaregistry:8085 \
--property schema.registry.ssl.truststore.location=/etc/kafka/secrets/kafka.badapp.truststore.jks \
--property schema.registry.ssl.truststore.password=confluent \
--property basic.auth.credentials.source=USER_INFO \
--property basic.auth.user.info=badapp:badapp \
--group wikipedia.test \
--topic wikipedia.parsed \
--max-messages 5
```
8. View all the role bindings that were configured for RBAC in this cluster.
```bash
./scripts/validate/validate_bindings.sh
```
9. Because the Kafka cluster is configured for [SASL](../../security/authentication/sasl/plain/overview.md#kafka-sasl-auth-plain), any administrative commands must authenticate directly with the Kafka brokers. This authentication is provided via a client properties file specified with the `--command-config` flag on the command-line tool itself. For example, to run a command like the [consumer throttle script](https://github.com/confluentinc/cp-demo/tree/latest/scripts/app/throttle_consumer.sh), you must include this flag pointing to a file with the correct security credentials. This replaces the previous method of relying on a pre-configured `KAFKA_OPTS` environment variable on a broker container. Consequently, the command is no longer restricted to running on a specific container like `kafka1` or `kafka2` and can be executed from any machine that has the configuration file and network access to the brokers.
10. Next step: Learn more about security with the [Security Tutorial](../../security/security_tutorial.md#security-tutorial).
## Data governance with Schema Registry
All the applications and connectors used in this example are configured to automatically read and write Avro-formatted data, leveraging the [Confluent Schema Registry](../../schema-registry/index.md#schemaregistry-intro).
The security in place between Schema Registry and the end clients, e.g. `appSA`, is as follows:
- Encryption: TLS, e.g. client has `schema.registry.ssl.truststore.*` configurations
- Authentication: bearer token authentication from HTTP basic auth headers, e.g. client has `basic.auth.user.info` and `basic.auth.credentials.source` configurations
- Authorization: Schema Registry uses the bearer token with RBAC to authorize the client
1. View the Schema Registry subjects for topics that have registered schemas for their keys and/or values. Notice the `curl` arguments include (a) TLS information required to interact with Schema Registry which is listening for HTTPS on port 8085, and (b) authentication credentials required for RBAC (using superUser:superUser to see all of them).
```text
docker exec schemaregistry curl -s -X GET \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
-u superUser:superUser \
https://schemaregistry:8085/subjects | jq .
```
Your output should resemble:
```JSON
[
"WIKIPEDIA_COUNT_GT_1-value",
"wikipedia-activity-monitor-KSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-value",
"wikipedia.parsed.replica-value",
"WIKIPEDIABOT-value",
"WIKIPEDIANOBOT-value",
"_confluent-ksql-ksql-clusterquery_CTAS_WIKIPEDIA_COUNT_GT_1_7-Aggregate-GroupBy-repartition-value",
"wikipedia.parsed.count-by-domain-value",
"wikipedia.parsed-value",
"_confluent-ksql-ksql-clusterquery_CTAS_WIKIPEDIA_COUNT_GT_1_7-Aggregate-Aggregate-Materialize-changelog-value"
]
```
2. Instead of using the superUser credentials, now use client credentials noexist:noexist (user does not exist in LDAP) to try to register a new Avro schema (a record with two fields `username` and `userid`) into Schema Registry for the value of a new topic `users`. It should fail due to an authorization error.
```text
docker compose exec schemaregistry curl -X POST \
-H "Content-Type: application/vnd.schemaregistry.v1+json" \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
--data '{ "schema": "[ { \"type\":\"record\", \"name\":\"user\", \"fields\": [ {\"name\":\"userid\",\"type\":\"long\"}, {\"name\":\"username\",\"type\":\"string\"} ]} ]" }' \
-u noexist:noexist \
https://schemaregistry:8085/subjects/users-value/versions
```
Your output should resemble:
```JSON
{"error_code":401,"message":"Unauthorized"}
```
3. Instead of using credentials for a user that does not exist, now use the client credentials appSA:appSA (the user appSA exists in LDAP) to try to register a new Avro schema (a record with two fields `username` and `userid`) into Schema Registry for the value of a new topic `users`. It should fail due to an authorization error, with a different message than above.
```text
docker compose exec schemaregistry curl -X POST \
-H "Content-Type: application/vnd.schemaregistry.v1+json" \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
--data '{ "schema": "[ { \"type\":\"record\", \"name\":\"user\", \"fields\": [ {\"name\":\"userid\",\"type\":\"long\"}, {\"name\":\"username\",\"type\":\"string\"} ]} ]" }' \
-u appSA:appSA \
https://schemaregistry:8085/subjects/users-value/versions
```
Your output should resemble:
```JSON
{"error_code":40301,"message":"User is denied operation Write on Subject: users-value"}
```
4. Create a role binding for the `appSA` client permitting it access to Schema Registry.
Get the Kafka cluster ID:
```none
KAFKA_CLUSTER_ID=$(curl -s https://localhost:8091/v1/metadata/id --tlsv1.2 --cacert scripts/security/snakeoil-ca-1.crt | jq -r ".id")
```
Create the role binding:
```text
# Create the role binding for the subject ``users-value``, i.e., the topic-value (versus the topic-key)
docker compose exec tools bash -c "confluent iam rbac role-binding create \
--principal User:appSA \
--role ResourceOwner \
--resource Subject:users-value \
--kafka-cluster-id $KAFKA_CLUSTER_ID \
--schema-registry-cluster schema-registry"
```
5. Again try to register the schema. It should pass this time. Note the schema id that it returns, e.g. below schema id is `9`.
```text
docker compose exec schemaregistry curl -X POST \
-H "Content-Type: application/vnd.schemaregistry.v1+json" \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
--data '{ "schema": "[ { \"type\":\"record\", \"name\":\"user\", \"fields\": [ {\"name\":\"userid\",\"type\":\"long\"}, {\"name\":\"username\",\"type\":\"string\"} ]} ]" }' \
-u appSA:appSA \
https://schemaregistry:8085/subjects/users-value/versions
```
Your output should resemble:
```JSON
{"id":9}
```
6. View the new schema for the subject `users-value`. From Confluent Control Center, click **Topics**. Scroll down to and click on the topic users and select “SCHEMA”.
You may alternatively request the schema via the command line:
```text
docker exec schemaregistry curl -s -X GET \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
-u appSA:appSA \
https://schemaregistry:8085/subjects/users-value/versions/1 | jq .
```
Your output should resemble:
```JSON
{
"subject": "users-value",
"version": 1,
"id": 9,
"schema": "{\"type\":\"record\",\"name\":\"user\",\"fields\":[{\"name\":\"username\",\"type\":\"string\"},{\"name\":\"userid\",\"type\":\"long\"}]}"
}
```
7. Describe the topic `users`. Notice that it has a special configuration `confluent.value.schema.validation=true` which enables [Schema Validation](../../schema-registry/schema-validation.md#schema-validation), a data governance feature in Confluent Server that gives operators a centralized location within the Kafka cluster itself to enforce data format correctness. Enabling Schema ID Validation allows brokers configured with `confluent.schema.registry.url` to validate that data produced to the topic is using a valid schema.
```bash
docker compose exec kafka1 kafka-topics \
--describe \
--topic users \
--bootstrap-server kafka1:9091 \
--command-config /etc/kafka/secrets/client_sasl_plain.config
```
Your output should resemble:
```bash
Topic: users PartitionCount: 2 ReplicationFactor: 2 Configs: confluent.value.schema.validation=true
Topic: users Partition: 0 Leader: 1 Replicas: 1,2 Isr: 1,2 Offline:
Topic: users Partition: 1 Leader: 2 Replicas: 2,1 Isr: 2,1 Offline:
```
8. Now produce a non-Avro message to this topic using `kafka-console-producer`.
```bash
docker compose exec connect kafka-console-producer \
--topic users \
--broker-list kafka1:11091 \
--producer-property security.protocol=SSL \
--producer-property ssl.truststore.location=/etc/kafka/secrets/kafka.appSA.truststore.jks \
--producer-property ssl.truststore.password=confluent \
--producer-property ssl.keystore.location=/etc/kafka/secrets/kafka.appSA.keystore.jks \
--producer-property ssl.keystore.password=confluent \
--producer-property ssl.key.password=confluent
```
After starting the console producer, it will wait for input. Enter a few characters and press enter.
It should result in a failure with an error message that resembles:
```bash
ERROR Error when sending message to topic users with key: null, value: 5 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.InvalidRecordException: This record has failed the validation on broker and hence be rejected.
```
Close the console producer by entering `CTRL+C`.
9. Describe the topic `wikipedia.parsed`, which is the topic that the kafka-connect-sse source connector is writing to. Notice that it also has enabled Schema ID Validation.
```bash
docker compose exec kafka1 kafka-topics \
--describe \
--topic wikipedia.parsed \
--bootstrap-server kafka1:9091 \
--command-config /etc/kafka/secrets/client_sasl_plain.config
```
10. Next step: Learn more about Schema Registry with the [Schema Registry Tutorial](../../schema-registry/schema_registry_onprem_tutorial.md#schema-registry-tutorial).
## Confluent REST Proxy
The [Confluent REST Proxy](../../kafka-rest/index.md#kafkarest-intro) is running for optional client access.
This demo showcases Confluent REST Proxy in two modes:
- Standalone service, listening for HTTPS requests on port 8086
- Embedded service on the Kafka brokers, listening for HTTPS requests on port 8091 on `kafka1` and on port 8092 on `kafka2` (these REST Proxy ports are shared with the broker’s Metadata Service (MDS) listener)
### Standalone REST Proxy
For the next few steps, use the REST Proxy that is running as a standalone service.
1. Use the standalone REST Proxy to try to produce a message to the topic `users`, referencing schema id `9`. This schema was registered in Schema Registry in the previous section. It should fail due to an authorization error.
```text
docker compose exec restproxy curl -X POST \
-H "Content-Type: application/vnd.kafka.avro.v2+json" \
-H "Accept: application/vnd.kafka.v2+json" \
--cert /etc/kafka/secrets/restproxy.certificate.pem \
--key /etc/kafka/secrets/restproxy.key \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
--data '{"value_schema_id": 9, "records": [{"value": {"user":{"userid": 1, "username": "Bunny Smith"}}}]}' \
-u appSA:appSA \
https://restproxy:8086/topics/users
```
Your output should resemble:
```JSON
{"offsets":[{"partition":null,"offset":null,"error_code":40301,"error":"Not authorized to access topics: [users]"}],"key_schema_id":null,"value_schema_id":9}
```
2. Create a role binding for the client permitting it produce to the topic `users`.
Get the Kafka cluster ID:
```none
KAFKA_CLUSTER_ID=$(curl -s https://localhost:8091/v1/metadata/id --tlsv1.2 --cacert scripts/security/snakeoil-ca-1.crt | jq -r ".id")
```
Create the role binding:
```text
# Create the role binding for the topic ``users``
docker compose exec tools bash -c "confluent iam rbac role-binding create \
--principal User:appSA \
--role DeveloperWrite \
--resource Topic:users \
--kafka-cluster-id $KAFKA_CLUSTER_ID"
```
3. Again try to produce a message to the topic `users`. It should pass this time.
```text
docker compose exec restproxy curl -X POST \
-H "Content-Type: application/vnd.kafka.avro.v2+json" \
-H "Accept: application/vnd.kafka.v2+json" \
--cert /etc/kafka/secrets/restproxy.certificate.pem \
--key /etc/kafka/secrets/restproxy.key \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
--data '{"value_schema_id": 9, "records": [{"value": {"user":{"userid": 1, "username": "Bunny Smith"}}}]}' \
-u appSA:appSA \
https://restproxy:8086/topics/users
```
Your output should resemble:
```JSON
{"offsets":[{"partition":1,"offset":0,"error_code":null,"error":null}],"key_schema_id":null,"value_schema_id":9}
```
4. Create consumer instance `my_avro_consumer`.
```text
docker compose exec restproxy curl -X POST \
-H "Content-Type: application/vnd.kafka.v2+json" \
--cert /etc/kafka/secrets/restproxy.certificate.pem \
--key /etc/kafka/secrets/restproxy.key \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
--data '{"name": "my_consumer_instance", "format": "avro", "auto.offset.reset": "earliest"}' \
-u appSA:appSA \
https://restproxy:8086/consumers/my_avro_consumer
```
Your output should resemble:
```text
{"instance_id":"my_consumer_instance","base_uri":"https://restproxy:8086/consumers/my_avro_consumer/instances/my_consumer_instance"}
```
5. Subscribe `my_avro_consumer` to the `users` topic.
```text
docker compose exec restproxy curl -X POST \
-H "Content-Type: application/vnd.kafka.v2+json" \
--cert /etc/kafka/secrets/restproxy.certificate.pem \
--key /etc/kafka/secrets/restproxy.key \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
--data '{"topics":["users"]}' \
-u appSA:appSA \
https://restproxy:8086/consumers/my_avro_consumer/instances/my_consumer_instance/subscription
```
6. Try to consume messages for `my_avro_consumer` subscriptions. It should fail due to an authorization error.
```text
docker compose exec restproxy curl -X GET \
-H "Accept: application/vnd.kafka.avro.v2+json" \
--cert /etc/kafka/secrets/restproxy.certificate.pem \
--key /etc/kafka/secrets/restproxy.key \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
-u appSA:appSA \
https://restproxy:8086/consumers/my_avro_consumer/instances/my_consumer_instance/records
```
Your output should resemble:
```text
{"error_code":40301,"message":"Not authorized to access group: my_avro_consumer"}
```
7. Create a role binding for the client permitting it access to the consumer group `my_avro_consumer`.
Get the Kafka cluster ID:
```none
KAFKA_CLUSTER_ID=$(curl -s https://localhost:8091/v1/metadata/id --tlsv1.2 --cacert scripts/security/snakeoil-ca-1.crt | jq -r ".id")
```
Create the role binding:
```text
# Create the role binding for the group ``my_avro_consumer``
docker compose exec tools bash -c "confluent iam rbac role-binding create \
--principal User:appSA \
--role ResourceOwner \
--resource Group:my_avro_consumer \
--kafka-cluster-id $KAFKA_CLUSTER_ID"
```
8. Again try to consume messages for `my_avro_consumer` subscriptions. It should fail due to a different authorization error.
```text
# Note: Issue this command twice due to https://github.com/confluentinc/kafka-rest/issues/432
docker compose exec restproxy curl -X GET \
-H "Accept: application/vnd.kafka.avro.v2+json" \
--cert /etc/kafka/secrets/restproxy.certificate.pem \
--key /etc/kafka/secrets/restproxy.key \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
-u appSA:appSA \
https://restproxy:8086/consumers/my_avro_consumer/instances/my_consumer_instance/records
docker compose exec restproxy curl -X GET \
-H "Accept: application/vnd.kafka.avro.v2+json" \
--cert /etc/kafka/secrets/restproxy.certificate.pem \
--key /etc/kafka/secrets/restproxy.key \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
-u appSA:appSA \
https://restproxy:8086/consumers/my_avro_consumer/instances/my_consumer_instance/records
```
Your output should resemble:
```JSON
{"error_code":40301,"message":"Not authorized to access topics: [users]"}
```
9. Create a role binding for the client permitting it access to the topic `users`.
Get the Kafka cluster ID:
```none
KAFKA_CLUSTER_ID=$(curl -s https://localhost:8091/v1/metadata/id --tlsv1.2 --cacert scripts/security/snakeoil-ca-1.crt | jq -r ".id")
```
Create the role binding:
```text
# Create the role binding for the group my_avro_consumer
docker compose exec tools bash -c "confluent iam rbac role-binding create \
--principal User:appSA \
--role DeveloperRead \
--resource Topic:users \
--kafka-cluster-id $KAFKA_CLUSTER_ID"
```
10. Again try to consume messages for `my_avro_consumer` subscriptions. It should pass this time.
```text
# Note: Issue this command twice due to https://github.com/confluentinc/kafka-rest/issues/432
docker compose exec restproxy curl -X GET \
-H "Accept: application/vnd.kafka.avro.v2+json" \
--cert /etc/kafka/secrets/restproxy.certificate.pem \
--key /etc/kafka/secrets/restproxy.key \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
-u appSA:appSA \
https://restproxy:8086/consumers/my_avro_consumer/instances/my_consumer_instance/records
docker compose exec restproxy curl -X GET \
-H "Accept: application/vnd.kafka.avro.v2+json" \
--cert /etc/kafka/secrets/restproxy.certificate.pem \
--key /etc/kafka/secrets/restproxy.key \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
-u appSA:appSA \
https://restproxy:8086/consumers/my_avro_consumer/instances/my_consumer_instance/records
```
Your output should resemble:
```JSON
[{"topic":"users","key":null,"value":{"userid":1,"username":"Bunny Smith"},"partition":1,"offset":0}]
```
11. Delete the consumer instance `my_avro_consumer`.
```text
docker compose exec restproxy curl -X DELETE \
-H "Content-Type: application/vnd.kafka.v2+json" \
--cert /etc/kafka/secrets/restproxy.certificate.pem \
--key /etc/kafka/secrets/restproxy.key \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
-u appSA:appSA \
https://restproxy:8086/consumers/my_avro_consumer/instances/my_consumer_instance
```
### Embedded REST Proxy
For the next few steps, use the REST Proxy that is embedded on the Kafka brokers. Only [REST Proxy API v3 for Confluent Platform](../../kafka-rest/api.md#rest-proxy-v3) is supported.
1. Create a role binding for the client to be granted `ResourceOwner` role for the topic `dev_users`.
Get the Kafka cluster ID:
```none
KAFKA_CLUSTER_ID=$(curl -s https://localhost:8091/v1/metadata/id --tlsv1.2 --cacert scripts/security/snakeoil-ca-1.crt | jq -r ".id")
```
Create the role binding:
```text
# Create the role binding for the topic ``dev_users``
docker compose exec tools bash -c "confluent iam rbac role-binding create \
--principal User:appSA \
--role ResourceOwner \
--resource Topic:dev_users \
--kafka-cluster-id $KAFKA_CLUSTER_ID"
```
2. Create the topic `dev_users` with embedded REST Proxy.
Get the Kafka cluster ID:
```none
KAFKA_CLUSTER_ID=$(curl -s https://localhost:8091/v1/metadata/id --tlsv1.2 --cacert scripts/security/snakeoil-ca-1.crt | jq -r ".id")
```
Use `curl` to create the topic:
```text
docker exec restproxy curl -s -X POST \
-H "Content-Type: application/json" \
-H "accept: application/json" \
-d "{\"topic_name\":\"dev_users\",\"partitions_count\":64,\"replication_factor\":2,\"configs\":[{\"name\":\"cleanup.policy\",\"value\":\"compact\"},{\"name\":\"compression.type\",\"value\":\"gzip\"}]}" \
--cert /etc/kafka/secrets/mds.certificate.pem \
--key /etc/kafka/secrets/mds.key \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
-u appSA:appSA \
"https://kafka1:8091/kafka/v3/clusters/$KAFKA_CLUSTER_ID/topics" | jq
```
3. List topics with embedded REST Proxy to find the newly created `dev_users`.
Get the Kafka cluster ID:
```none
KAFKA_CLUSTER_ID=$(curl -s https://localhost:8091/v1/metadata/id --tlsv1.2 --cacert scripts/security/snakeoil-ca-1.crt | jq -r ".id")
```
Use `curl` to list the topics:
```text
docker exec restproxy curl -s -X GET \
-H "Content-Type: application/json" \
-H "accept: application/json" \
--cert /etc/kafka/secrets/mds.certificate.pem \
--key /etc/kafka/secrets/mds.key \
--tlsv1.2 \
--cacert /etc/kafka/secrets/snakeoil-ca-1.crt \
-u appSA:appSA \
https://kafka1:8091/kafka/v3/clusters/$KAFKA_CLUSTER_ID/topics | jq '.data[].topic_name'
```
Your output should resemble below. Output may vary, depending on other topics you may have created, but at least you should see the topic `dev_users` created in the previous step.
```text
"_confluent-monitoring"
"dev_users"
"users"
"wikipedia-activity-monitor-KSTREAM-AGGREGATE-STATE-STORE-0000000003-changelog"
"wikipedia-activity-monitor-KSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition"
"wikipedia.failed"
"wikipedia.parsed"
"wikipedia.parsed.count-by-domain"
"wikipedia.parsed.replica"
```
## Failed broker
To simulate a failed broker, stop the Docker container running one of
the two Kafka brokers.
1. Stop the Docker container running Kafka broker 2.
```bash
docker compose stop kafka2
```
2. After a few minutes, observe the Broker summary show that the number of brokers
has decreased from 2 to 1, and there are many under replicated
partitions.
3. View Topic information details to see that there are out of sync replicas.
4. Restart the Docker container running Kafka broker 2.
```bash
docker compose start kafka2
```
5. After about a minute, observe the broker summary in Confluent Control Center.
The broker count has recovered to 2, and the topic
partitions are back to reporting no under replicated partitions.
6. Click the broker count `2` inside the **Brokers** box and when
the **Brokers overview** pane appears, click inside the **Partitioning
and replication** box to view when broker counts changed.
## Alerting
There are many types of Control Center [alerts](https://docs.confluent.io/control-center/current/alerts/concepts.htmlcontrolcenter_userguide_alerts)
and many ways to configure them. Use the Alerts management page to
define triggers and actions, or click on individual resources
to setup alerts from there.
1. This example already has pre-configured triggers and actions. To view configuration details, navigate to the **Alerts Triggers** screen, and then click **Edit** for each trigger.
- The trigger `Under Replicated Partitions` happens when a broker
reports non-zero under replicated partitions, and it causes an
action `Email Administrator`.
- The trigger `Consumption Difference` happens when consumption
difference for the Elasticsearch connector consumer group is
greater than `0`, and it causes an action
`Email Administrator`.
2. If you followed the steps in the [failed broker](#failed-broker)
section, view the **Alerts History** to see that the trigger
`Under Replicated Partitions` happened and caused an alert when you
stopped broker 2.
3. You can also trigger the `Consumption Difference` trigger. In the
Kafka Connect -> Sinks screen, edit the running Elasticsearch sink
connector.
4. In the Connect view, pause the Elasticsearch sink connector in Settings by
pressing the pause icon in the top right. This stops consumption
for the related consumer group.
5. View the alert history to see that this trigger happened and caused
an alert.
## Confluent Self-Balancing Clusters
[Self-Balancing Clusters](../../clusters/sbc/index.md#sbc) automates your resource workload balancing, provides failure detection and self-healing, and allows you to add or decommission brokers as needed, with no manual tuning required. This simplifies scale-up and scale-down operations, ensuring workload is assigned to new brokers and automating recovery in case of a failure.
This section showcases two features of Self-Balancing Clusters:
- Adding a new broker to the cluster (scale-up): observe Self-Balancing Clusters rebalance the cluster by assigning existing partitions to the new broker.
- Simulating a failure by killing a broker: observe Self-Balancing Clusters reassign the failed broker’s replicas to the remaining brokers.
Before running this section:
- Self-Balancing requires 15 minutes to initialize and collect metrics from brokers in the cluster, so after starting `cp-demo`, wait at least this time before proceeding.
- Because these steps add a third broker, ensure you have adequate resources allocated to Docker.
1. Run [scripts/sbc/add-broker.sh](https://github.com/confluentinc/cp-demo/tree/latest/scripts/sbc/add-broker.sh) to add a new broker `kafka3` to the cluster.
```bash
./scripts/sbc/add-broker.sh
```
The script returns when Self-Balancing Clusters has acknowledged the broker addition and started a rebalance task.
2. Open Control Center at [http://localhost:9021](http://localhost:9021) and navigate to **Brokers**. The Self-balancing panel shows 1 in-progress task.
Click the panel and locate the in-progress add-broker task for broker with ID **broker.3**.
3. The add-broker rebalance task progresses through stages `PLAN_COMPUTATION`, `REASSIGNMENT` and finally `COMPLETED`. The time spent in each phase varies depending on the hardware you are running.
Check the status using the following scripts to read the broker logs:
```bash
./scripts/sbc/validate_sbc_add_broker_plan_computation.sh
./scripts/sbc/validate_sbc_add_broker_reassignment.sh
./scripts/sbc/validate_sbc_add_broker_completed.sh
```
4. After a few minutes, when rebalancing has completed, the add-broker rebalance task in Confluent Control Center moves to `Success`.
Run the following command to view replica placements for all topic-partitions in the cluster:
```bash
docker compose exec kafka1 kafka-replica-status \
--bootstrap-server kafka1:9091 \
--admin.config /etc/kafka/secrets/client_sasl_plain.config
```
Look for instances of `3` in the `Replica` column, showing that rebalancing has assigned partition replicas (leaders and followers) to the new broker.
5. Simulate a broker-failure by running [scripts/sbc/kill-broker.sh](https://github.com/confluentinc/cp-demo/tree/latest/scripts/sbc/kill-broker.sh) to kill the previously-added broker `kafka3`:
```bash
./scripts/sbc/kill-broker.sh
```
This script returns when Self-Balancing Clusters has detected the broker failure and the recovery wait-time `KAFKA_CONFLUENT_BALANCER_HEAL_BROKER_FAILURE_THRESHOLD_MS` has expired, triggering replica reassignment from the `kafka3` broker to the original two brokers.
Note that in `cp-demo`, the threshold time has been set to 30 seconds, which is artificially low but useful in a demo environment.
6. Monitor the progress of replica-reassignment from the failed broker, which eventually reduces the under-replicated partitions in the cluster back to zero. To track completion of self-healing, check the **Self-Balancing** panel in Confluent Control Center, or run the following scripts:
```bash
./scripts/sbc/validate_sbc_kill_broker_started.sh
./scripts/sbc/validate_sbc_kill_broker_completed.sh
```
7. When self-healing has completed, the Kafka cluster should no longer have any under-replicated partitions that were previously assigned to failed broker `kafka3`.
Confirm this by running this command and verifying no output, meaning no out-of-sync replicas:
```bash
docker exec kafka1 kafka-replica-status \
--bootstrap-server kafka1:9091 \
--admin.config /etc/kafka/secrets/client_sasl_plain.config \
--verbose | grep "IsInIsr: false"
```
## Monitoring
This tutorial has demonstrated how Confluent Control Center helps users manage their Confluent Platform deployment and how it provides monitoring capabilities for the cluster and applications.
For a practical guide to optimizing your Kafka deployment for various service goals including throughput, latency, durability and availability, and useful metrics to monitor for performance and cluster health for on-premises Kafka clusters, see the [Optimizing Your Apache Kafka Deployment](https://www.confluent.io/white-paper/optimizing-your-apache-kafka-deployment/) whitepaper.
For most Confluent Platform users the Confluent Control Center monitoring and integrations are sufficient for production usage in their on-premises Apache Kafka® deployments.
There are additional monitoring solutions for various use cases, as described below.
### Metrics API
The Confluent Cloud Metrics API is a REST API you can use to query timeseries metrics.
You can use the Metrics API to get telemetry data for both the on-premises Confluent Platform cluster as well as the Confluent Cloud cluster.
- On-premises metrics (enabled by Telemetry Reporter) using the endpoint [https://api.telemetry.confluent.cloud/v2/metrics/hosted-monitoring/query](https://api.telemetry.confluent.cloud/v2/metrics/hosted-monitoring/query)
#### NOTE
The hosted monitoring endpoint is in preview and the endpont will eventually be renamed [https://api.telemetry.confluent.cloud/v2/metrics/health-plus/query](https://api.telemetry.confluent.cloud/v2/metrics/health-plus/query)
- Confluent Cloud metrics using the endpoint [https://api.telemetry.confluent.cloud/v2/metrics/cloud/query](https://api.telemetry.confluent.cloud/v2/metrics/cloud/query)
- See the [Confluent Cloud Metrics API Reference](https://api.telemetry.confluent.cloud/docs) for more information.
The Metrics API and Telemetry Reporter powers [Health+](/platform/current/health-plus/index.html), the fully-managed monitoring
solution for Confluent Platform. You can enable Health+ for free and add premium capabilities as you see fit.
Popular [third-party monitoring tools](https://docs.confluent.io/cloud/current/monitoring/metrics-api.html#integrate-with-third-party-monitoring)
like Datadog and Grafana Cloud integrate with the Metrics API out-of-the-box,
or if you manage your own Prometheus database, the Metrics API can also export metrics in Prometheus format.
See [Module 2: Deploy Hybrid Confluent Platform and Confluent Cloud Environment](hybrid-cloud.md#cp-demo-hybrid) to play hands-on with the Metrics API.
### JMX
Some users wish to integrate with other monitoring solutions like Prometheus, Grafana, Datadog, and Splunk.
The following JMX-based monitoring stacks help users setup a ‘single pane of glass’ monitoring solution for all their organization’s services and applications, including Kafka.
Here are some examples of monitoring stacks that integrate with Confluent Platform:
1. [JMX Exporter + Prometheus + Grafana](https://github.com/confluentinc/jmx-monitoring-stacks) (runnable with cp-demo from [https://github.com/confluentinc/jmx-monitoring-stacks](https://github.com/confluentinc/jmx-monitoring-stacks)):
2. [Jolokia + Elasticsearch + Kibana](https://github.com/confluentinc/jmx-monitoring-stacks) (runnable with cp-demo from [https://github.com/confluentinc/jmx-monitoring-stacks](https://github.com/confluentinc/jmx-monitoring-stacks)):
3. [Monitoring Confluent Platform with Datadog](https://www.confluent.io/blog/confluent-datadog-integration-kafka-monitoring-metrics):
## Related content
- [Scripted Confluent Platform Demo](overview.md#scripted-demo)
- [Module 2: Deploy Hybrid Confluent Platform and Confluent Cloud Environment](hybrid-cloud.md#cp-demo-hybrid)
- [Troubleshoot and Stop the Confluent Platform Demo](teardown.md#cp-demo-teardown)