.. _ksql-rbac:

KSQL and |rbac-long|
####################

Starting in |cp| 5.3, KSQL supports :ref:`role-based access control (RBAC) <rbac-overview>`
to cluster resources.

.. note::

   KSQL supports |rbac| to provide authorization to KSQL clusters when running
   in interactive mode. |rbac| for users isn't used in KSQL clusters that are
   deployed in headless mode.

Use the :ref:`Confluent CLI <confluent_iam_rolebinding_create>` or  
:ref:`Control Center <c3_rbac_manage_roles_ui>` to create roles for users.

You need the cluster ID of the |ak| cluster, the service ID of the KSQL
cluster, and the cluster ID of the |sr| cluster, if your topics have Avro
schemas. For more information, see :ref:`rbac-get-cluster-ids`.

.. include:: ../../includes/rbac-demo.rst

KSQL |rbac-sa|
**************

KSQL applications run with the roles granted to a privileged *KSQL user*
|rbac-sa|. By default, the KSQL |rbac-sa| is named "ksql". The roles granted to
the KSQL |rbac-sa| by default are required to run KSQL and are distinct from
the roles granted to an interactive user who wants to connect and contribute
to KSQL.

When you grant roles to an interactive user, you must also grant the same roles
to the KSQL user |rbac-sa|. KSQL applications execute user commands as the KSQL
|rbac-sa| in the background and on other servers in the KSQL cluster, and some
KSQL statements start long-running persistent queries. KSQL checks that the
interactive user has the required permissions to access any resources required,
but the query itself runs in the security context of the KSQL |rbac-sa|. For
this reason, it's important that both the interactive user and the KSQL
|rbac-sa| have the correct permissions to all resources required by any KSQL
statements.

The following example commands show how to grant an interactive user and the
KSQL |rbac-sa| read-only access to a topic. To grant these roles, you must be
logged in to the |confluent-cli| with sufficient permissions, for example as
``UserAdmin`` or ``ResourceOwner``.

.. code:: bash

    # Grant read-only access for a user to a topic.
    confluent iam rolebinding create \
      --principal User:<user-name> \
      --role DeveloperRead \
      --resource Topic:<topic-name> \
      --kafka-cluster-id <kafka-id>

    # Grant read-only access for the ksql service principal to a topic.
    confluent iam rolebinding create \
      --principal User:ksql \
      --role DeveloperRead \
      --resource Topic:<topic-name> \
      --kafka-cluster-id <kafka-id>

Configure KSQL Server for |rbac|
********************************

Assign the following configuration settings in the KSQL Server properties
file to enable KSQL in an :ref:`RBAC environment <rbac-configuration>`.

To access the Confluent security service, you must generate a public/private
key pair. Specify the path to the public key file by assigning the
``public.key.path`` setting. For more information, see :ref:`rbac-mds-config`.

.. include:: ../../includes/secret-tip.rst

Append the following settings to the KSQL Server configuration file, which by
default is named ``${CONFLUENT_HOME}/etc/ksql/ksql-server.properties``.

::

    #------ External service configuration -------

    # The set of Kafka brokers to bootstrap Kafka cluster information from
    bootstrap.servers=<kafka-hostname>:9092

    # Schema Registry authentication
    ksql.schema.registry.url=http://<sr-hostname>:8081
    # Credentials for accessing Schema Registry
    ksql.schema.registry.basic.auth.user.info=<ksql-user>:<ksql-user-password>

    ############ KSQL security settings for Confluent RBAC ######################
    
    # Set the KSQL cluster ID to work on specific roles and permissions for this KSQL cluster scope.
    ksql.service.id=<ksql-cluster-id>
    
    # Enable the KSQL Confluent security plugin.
    ksql.security.extension.class=io.confluent.ksql.security.KsqlConfluentSecurityExtension
    
    # Assign the metadata URL and access credentials, which are shared by authentication security handlers and the KSQL security extension.
    # May be the same as the Kafka cluster hostname.
    confluent.metadata.bootstrap.server.urls=http://<metadata-server-hostname>:8090
    confluent.metadata.http.auth.credentials.provider=BASIC
    # Credentials for accessing the metadata server
    confluent.metadata.basic.auth.user.info=<ksql-user>:<ksql-user-password>
    
    # Enable KSQL Basic+Bearer authentication.
    rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.InstallBearerOrBasicSecurityHandler
    websocket.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.InstallBearerOrBasicSecurityHandler
    public.key.path=<path/to/publickey.pem>
    
    # Assign credentials for Kafka access.
    # These settings may vary, depending on your Kafka cluster authentication settings.
    security.protocol=SASL_PLAINTEXT
    sasl.mechanism=OAUTHBEARER
    sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler
    # Credentials for the KSQL user, which has the default user name of "ksql".
    # These credentials authorize KSQL Server to access the Kafka cluster.
    sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
        metadataServerUrls="http://<metadata-server-hostname>" username="<ksql-user>" password="<ksql-user-password>";

Save the file and restart KSQL Server.

|rbac| for KSQL depends on the |cp| :ref:`Metadata Service (MDS) <metadata-service>`
and the |csa|. The ``confluent.metadata`` settings configure the Metadata
Service. The ``ksql.security.extension.class`` setting configures KSQL for the
|csa|. For more information, see :ref:`confluent_server_authorizer`.

Use the KSQL |rbac-sa| credentials for the following settings.

- ``sasl.jaas.config`` for authorizing to the Kafka cluster with |csa|
- ``confluent.metadata.basic.auth.user.info`` for authorizing to MDS
- ``ksql.schema.registry.basic.auth.user.info`` for authorizing to |sr|

Assign roles for KSQL access
****************************

The following section defines the minimal set of roles and resources required
for a user to connect to KSQL and execute KSQL statements, including transient
queries.

As a best practice for security, give users the minimum necessary access level
to resources. For example, prefer granting an interactive user the
``DeveloperRead`` role over the ``ResourceOwner`` role for reading from a topic. 

.. important::

     Users require permissions in Kafka to use any KSQL statements that attempt
     to read, write, list, and create topics. Additional access to resources on
     the Kafka cluster and |sr| must be granted explicitly. For more information,
     see :ref:`config-security-ksql-grant-access`.

To grant these roles, you must be logged in to the |confluent-cli| as a
``UserAdmin``, ``ResourceOwner``, or other role with sufficient permissions.

Cluster identifiers
===================

To create role bindings, you need the cluster identifiers for the components in
your |cp| deployment. For more information, see :ref:`rbac-get-cluster-ids`.

The following commands use these environment variables for the cluster IDs. 
Variables for the interactive user name and topic name are also defined.

.. code:: bash

    # Cluster identifiers
    export KAFKA_ID=<kafka-cluster-id>
    export KSQL_ID=<ksql-cluster-id>
    export SR_ID=<schema-registry-cluster-id>

    # The name of the interactive user and
    # the name of a topic for the user to access. 
    export USER_NAME=<interactive-user-name>
    export TOPIC_NAME=<kafka-topic-name>

.. _ksql-rbac-assign-startup-roles:

Assign roles for starting KSQL Server
=====================================

To start KSQL Server and connect to a secured Kafka cluster, a user must have
the following roles granted by a ``UserAdmin``.

KSQL cluster scope
------------------

Run the following commands to assign the ``ResourceOwner`` role on the KSQL
cluster, which binds ownership of the KSQL cluster's resources to the user.
A user who has the ``ResourceOwner`` or ``ClusterAdmin`` role on a KSQL cluster
is a *cluster owner*.

.. code:: bash

    confluent iam rolebinding create \
      --principal User:<cluster-owner-name> \
      --role ResourceOwner \
      --kafka-cluster-id $KAFKA_ID \
      --ksql-cluster-id $KSQL_ID \
      --resource KsqlCluster:ksql-cluster

    confluent iam rolebinding create \
      --principal User:<cluster-owner-name> \
      --role SecurityAdmin \
      --kafka-cluster-id $KAFKA_CLUSTER_ID \
      --ksql-cluster-id $KSQL_ID

Run the following commands to assign the ``ResourceOwner`` role to the KSQL
|rbac-sa|.

.. code:: bash

    confluent iam rolebinding create \
      --principal User:ksql \
      --role ResourceOwner \
      --kafka-cluster-id $KAFKA_ID \
      --ksql-cluster-id $KSQL_ID \
      --resource KsqlCluster:ksql-cluster

Kafka cluster scope
-------------------

Grant the ``ResourceOwner`` role to the KSQL |rbac-sa| on the following internal
KSQL resources. By default, these roles are granted to the KSQL user named
"ksql". 

- KSQL command topic: ``_confluent-ksql-${KSQL_ID}_command_topic``
- KSQL processing log topic: ``${KSQL_ID}-ksql_processing_log`` 
- Consumer groups: Prefixed with ``_confluent-ksql-${KSQL_ID}``

Use the following command to grant full access for the KSQL |rbac-sa| to the
KSQL command topic.

.. code:: bash

  # Grant full access for the KSQL service principal to the KSQL command topic. 
  confluent iam rolebinding create \
    --principal User:ksql \
    --role ResourceOwner \
    --resource Topic:_confluent-ksql-${KSQL_ID}_command_topic \
    --kafka-cluster-id $KAFKA_ID

Use the following command to grant full access for the KSQL |rbac-sa| to the
KSQL processing log topic.

.. code:: bash

  # Grant full access for the KSQL service principal to the KSQL processing log topic. 
  confluent iam rolebinding create \
    --principal User:ksql \
    --role ResourceOwner \
    --resource Topic:${KSQL_ID}-ksql_processing_log \
    --kafka-cluster-id $KAFKA_ID

Use the following command to grant full access for the KSQL user to the
KSQL consumer groups.

.. code:: bash

  # Grant full access for the KSQL service principal to the KSQL consumer groups. 
  confluent iam rolebinding create \
    --principal User:ksql \
    --role ResourceOwner \
    --resource Group:_confluent-ksql-${KSQL_ID} \
    --prefix \
    --kafka-cluster-id $KAFKA_ID

Assign roles for terminating a KSQL cluster
===========================================

To terminate a KSQL cluster by calling the REST endpoint at ``/ksql/terminate``,
a user must have the ``DeveloperManage`` role granted by a ``UserAdmin``. For
more information, see :ref:`ksql_cluster_terminate`.

KSQL cluster scope
------------------

Use the following command to assign a ``DeveloperManage`` role on the KSQL
cluster.

.. code:: bash

    confluent iam rolebinding create \
        --principal User:<cluster-owner-name> \
        --role DeveloperManage \
        --kafka-cluster-id $KAFKA_ID \
        --ksql-cluster-id $KSQL_ID \
        --resource KsqlCluster:ksql-cluster

.. _ksql-rbac-assign-roles-to-contribute:

Assign roles for contributing to a KSQL application
===================================================

To access the KSQL cluster for development, an interactive user must have the
following roles.

KSQL cluster scope
------------------

Use the following command to assign the ``DeveloperWrite`` role on the KSQL
cluster.

.. code:: bash

    # Grant write access for a user to resources on the KSQL cluster.
    confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role DeveloperWrite \
        --kafka-cluster-id $KAFKA_ID \
        --ksql-cluster-id $KSQL_ID \
        --resource KsqlCluster:ksql-cluster

Kafka cluster scope
-------------------

Use the following command to assign the ``DeveloperRead`` role on the internal
consumer groups for transient queries.

.. code:: bash

    # Grant read-only access for a user to the KSQL consumer groups.
    confluent iam rolebinding create \
      --principal User:$USER_NAME \
      --role DeveloperRead \
      --resource Group:_confluent-ksql-${KSQL_ID} \
      --prefix \
      --kafka-cluster-id $KAFKA_ID

Assign the following role if the interactive user needs access to the KSQL
processing log.

.. code:: bash

    # Grant access for a user to the processing log. 
    confluent iam rolebinding create \
      --principal User:$USER_NAME \
      --role DeveloperRead \
      --resource Topic:${KSQL_ID}-ksql_processing_log \
      --kafka-cluster-id $KAFKA_CLUSTER_ID

.. _config-security-ksql-grant-access:

Grant a user access to resources
================================

In addition to the minimum necessary roles, an interactive user needs roles to
access specific resources like topics, consumer groups, and subjects.

.. important:: 
    
    When you grant roles to a user, remember to grant the same roles to the
    KSQL |rbac-sa|, which is named "ksql" by default.

Prefixed resource names
-----------------------

Use the ``prefix`` option in the |confluent-cli| to grant access to all
resources with names that start with the specified prefix string. For example,
to grant read-only access to all topics that start with ``<topic-name-prefix->``,
you could use the following command.

.. code:: bash

    # Grant read-only access for a user to a topic.
    confluent iam rolebinding create \
      --principal User:$USER_NAME \
      --role DeveloperRead \
      --resource Topic:<topic-name-prefix-> \
      --prefix \
      --kafka-cluster-id $KAFKA_ID

Roles for accessing topics, streams, and tables
-----------------------------------------------

Use the following |confluent-cli| commands to give an interactive user the
necessary roles for creating streams and table.

SHOW or PRINT a topic
  - ``DeveloperRead`` role on the Kafka topic
  - ``DeveloperRead`` role on the |sr| subject, if the topic has an Avro schema
  
  This role enables an interactive user to display the specified topic by using
  the SHOW and PRINT statements. Also, users can CREATE streams and tables from
  these topics.

  The KSQL |rbac-sa| doesn't need a role on the topic for these statements.
  
  .. note::

      The subject's name is the topic's name appended with ``-value``.

  .. code:: bash

      # Grant read-only access for a user to read a topic.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role DeveloperRead \
        --resource Topic:$TOPIC_NAME \
        --kafka-cluster-id $KAFKA_ID

  .. code:: bash
  
      # Grant read-only access for a user to read a subject.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role DeveloperRead \
        --resource Subject:${TOPIC_NAME}-value
        --kafka-cluster-id $KAFKA_ID \
        --schema-registry-cluster-id $SR_ID

SELECT from a stream or table
  - ``DeveloperRead`` role on the source topic
  - ``DeveloperRead`` role on the ``_confluent-ksql-${KSQL_ID}`` consumer group
  - ``ResourceOwner`` role on the ``_confluent-ksql-${KSQL_ID}transient``
    transient topics, for tables
  - ``DeveloperRead`` role on the |sr| subject, if the topic has an Avro schema

  These roles enable a user to read from a stream or a table by using the
  SELECT statement. If a SELECT statement contains a JOIN that uses an
  unauthorized topic, the SELECT fails with an authorization error.

  .. code:: bash

      # Grant read-only access for a user to read a topic.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role DeveloperRead \
        --resource Topic:$TOPIC_NAME \
        --kafka-cluster-id $KAFKA_ID

  .. code:: bash

      # For tables: grant access to the transient query topics.
      # This is a limitation of KSQL tables. Giving this permission to
      # the prefixed topics lets the user view tables from other queries.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role ResourceOwner \
        --resource Topic:_confluent-ksql-${KSQL_ID}transient \
        --prefix \
        --kafka-cluster-id $KAFKA_CLUSTER_ID

  .. code:: bash

      # Grant read-only access for a user to read from a consumer group.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role DeveloperRead \
        --resource Group:_confluent-ksql-${KSQL_ID} \
        --prefix \
        --kafka-cluster-id $KAFKA_ID

  .. code:: bash
  
      # Grant read-only access for a user to read a subject.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role DeveloperRead \
        --resource Subject:${TOPIC_NAME}-value
        --kafka-cluster-id $KAFKA_ID \
        --schema-registry-cluster-id $SR_ID


Write to a topic with INSERT
  - ``DeveloperWrite`` role on the Kafka topic
  - ``ResourceOwner`` or ``DeveloperWrite`` role on the |sr| subject,
    if the topic has an Avro schema
  
  These roles enable a user to write data by using INSERT statements.
  The INSERT INTO statement contains a SELECT clause that requires the user
  to have read permissions on the topic in the query.

  .. code:: bash

      # Grant write access for a user to a topic.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role DeveloperWrite \
        --resource Topic:$TOPIC_NAME \
        --kafka-cluster-id $KAFKA_ID

  .. code:: bash

      # Grant full access for a user to create a subject and write to it.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role ResourceOwner \
        --resource Subject:${TOPIC_NAME}-value
        --kafka-cluster-id $KAFKA_ID \
        --schema-registry-cluster-id $SR_ID        

CREATE STREAM  
  - ``DeveloperRead`` role on the source topic
  - ``DeveloperRead`` role on the ``_confluent-ksql-${KSQL_ID}`` consumer groups
  - ``ResourceOwner`` or ``DeveloperWrite`` role on the |sr| subject,
    if the source topic has an Avro schema

  These roles enable an interactive user to register a stream or table on the
  specified topic by using the CREATE STREAM statement.

  If the topic has an Avro schema, the interactive user and the KSQL |rbac-sa|
  must have full access for the subject in |sr|.

  .. code:: bash

      # Grant read-only access for a user to a topic.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role DeveloperRead \
        --resource Topic:$TOPIC_NAME \
        --kafka-cluster-id $KAFKA_ID

      # Grant read-only access for the KSQL service principal to a topic.
      confluent iam rolebinding create \
        --principal User:ksql \
        --role DeveloperRead \
        --resource Topic:$TOPIC_NAME \
        --kafka-cluster-id $KAFKA_ID

  .. code:: bash

      # Grant read-only access for a user to the KSQL consumer groups.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role DeveloperRead \
        --resource Group:_confluent-ksql-${KSQL_ID} \
        --prefix \
        --kafka-cluster-id $KAFKA_ID

      # Grant read-only access for the KSQL service principal to the KSQL consumer groups.
      confluent iam rolebinding create \
        --principal User:ksql \
        --role DeveloperRead \
        --resource Group:_confluent-ksql-${KSQL_ID} \
        --prefix \
        --kafka-cluster-id $KAFKA_ID

  .. code:: bash

      # Grant full access for a user to create a subject and write to it.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role ResourceOwner \
        --resource Subject:${TOPIC_NAME}-value \
        --kafka-cluster-id $KAFKA_ID \
        --schema-registry-cluster-id $SR_ID

      # Grant full access for the KSQL service principal to create a subject and write to it.
      confluent iam rolebinding create \
        --principal User:ksql \
        --role ResourceOwner \
        --resource Subject:${TOPIC_NAME}-value \
        --kafka-cluster-id $KAFKA_ID \
        --schema-registry-cluster-id $SR_ID

CREATE TABLE  
  - ``DeveloperRead`` role on the source topic
  - ``ResourceOwner`` role on the ``_confluent-ksql-${KSQL_ID}transient`` transient topics
  - ``ResourceOwner`` or ``DeveloperWrite`` role on the |sr| subject,
    if the source topic has an Avro schema

  These roles enable an interactive user to register a table on the
  specified topic by using the CREATE TABLE statement.

  If the topic has an Avro schema, the interactive user and the KSQL |rbac-sa|
  must have full access for the subject in |sr|.

  .. note::

     The ``ResourceOwner`` role on the transient topics is a limitation of KSQL
     tables. Giving this permission to the prefixed topics lets the user view
     tables from other queries.

  .. code:: bash

      # Grant read-only access for a user to a topic.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role DeveloperRead \
        --resource Topic:$TOPIC_NAME \
        --kafka-cluster-id $KAFKA_ID

      # Grant read-only access for the KSQL service principal to a topic.
      confluent iam rolebinding create \
        --principal User:ksql \
        --role DeveloperRead \
        --resource Topic:$TOPIC_NAME \
        --kafka-cluster-id $KAFKA_ID

  .. code:: bash

      # Grant full access for a user to the transient query topics.
      # This is a limitation of KSQL tables. Giving this permission to
      # the prefixed topics lets the user view tables from other queries.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role ResourceOwner \
        --resource Topic:_confluent-ksql-${KSQL_ID}transient \
        --prefix \
        --kafka-cluster-id $KAFKA_ID

      # Grant full access for the KSQL service principal to KSQL transient topics.
      confluent iam rolebinding create \
        --principal User:ksql \
        --role ResourceOwner \
        --resource Topic:_confluent-ksql-${KSQL_ID}transient \
        --prefix \
        --kafka-cluster-id $KAFKA_ID

  .. code:: bash

      # Grant full access for a user to create a subject and write to it.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role ResourceOwner \
        --resource Subject:${TOPIC_NAME}-value \
        --kafka-cluster-id $KAFKA_ID \
        --schema-registry-cluster-id $SR_ID

      # Grant full access for the KSQL service principal to create a subject and write to it.
      confluent iam rolebinding create \
        --principal User:ksql \
        --role ResourceOwner \
        --resource Subject:${TOPIC_NAME}-value \
        --kafka-cluster-id $KAFKA_ID \
        --schema-registry-cluster-id $SR_ID

Create a stream or table with a persistent query
  - ``DeveloperRead`` role on the source topic
  - ``ResourceOwner`` role on the KSQL sink topic
  - ``ResourceOwner`` or ``DeveloperWrite`` role on the |sr| sink subject,
    if the source topic has an Avro schema
  - ``DeveloperRead`` role on ``_confluent-ksql-${KSQL_ID}*`` subjects (for
    tables that use Avro)
  
  These roles enable a user to create streams and tables with persistent
  queries. Because KSQL creates a new *sink topic*, the user must have
  sufficient permissions to create, read, and write to the sink topic. The
  ``ResourceOwner`` role is necessary on the sink topic, because the interactive
  user and the KSQL |rbac-sa| need permissions to create the sink topic if it
  doesn't exist already.

  .. note::

      The sink topic has the same name as the stream or table and is all
      uppercase.

  If the topic has an Avro schema, the interactive user and the KSQL |rbac-sa|
  must have ``ResourceOwner`` or ``DeveloperWrite`` permission on the sink
  topic's subject in |sr|.
  
  For tables that are created with a persistent query and use Avro, the KSQL
  |rbac-sa| must have ``DeveloperRead`` permission on all internal KSQL
  subjects.
  
  .. note::

      The subject's name is the sink topic's name appended with ``-value``.

  .. code:: bash

      # Grant read-only access for a user to a topic.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role DeveloperRead \
        --resource Topic:$SOURCE_TOPIC_NAME \
        --kafka-cluster-id $KAFKA_ID
      
      # Grant read-only access for the KSQL service principal to a topic.
      confluent iam rolebinding create \
        --principal User:ksql \
        --role DeveloperRead \
        --resource Topic:$SOURCE_TOPIC_NAME \
        --kafka-cluster-id $KAFKA_ID

  .. code:: bash

      # Grant read-only access for a user to a topic.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role ResourceOwner \
        --resource Topic:$SINK_TOPIC_NAME \
        --kafka-cluster-id $KAFKA_ID

      # Grant read-only access for the KSQL service principal to a topic.
      confluent iam rolebinding create \
        --principal User:ksql \
        --role ResourceOwner \
        --resource Topic:$SINK_TOPIC_NAME \
        --kafka-cluster-id $KAFKA_ID

  .. code:: bash

      # Grant full access for a user to create a subject and write to it.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role ResourceOwner \
        --resource Subject:${SINK_TOPIC_NAME}-value \
        --kafka-cluster-id $KAFKA_ID \
        --schema-registry-cluster-id $SR_ID

      # Grant full access for the KSQL service principal to create a subject and write to it.
      confluent iam rolebinding create \
        --principal User:ksql \
        --role ResourceOwner \
        --resource Subject:${SINK_TOPIC_NAME}-value \
        --kafka-cluster-id $KAFKA_ID \
        --schema-registry-cluster-id $SR_ID

      # For tables that use Avro created with a persistent query:
      # Grant read access for the KSQL service principal to all internal ksql subjects.
      confluent iam rolebinding create \
        --principal User:ksql \
        --role DeveloperRead \
        --resource Subject:_confluent-ksql-${KSQL_ID} \
        --prefix \
        --kafka-cluster-id $KAFKA_ID \
        --schema-registry-cluster-id $SR_ID

Full control over a topic and schema
  - ``ResourceOwner`` role on the Kafka topic
  - ``ResourceOwner`` role on the |sr| subject, if the topic has an Avro schema

  Use the following commands to grant a user full control over a topic and its
  schema, including permissions to delete the topic and schema.

  .. code:: bash

      # Grant full access for a user to manage a topic. 
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role ResourceOwner \
        --resource Topic:$TOPIC_NAME \
        --kafka-cluster-id $KAFKA_ID

  .. code:: bash
  
      # Grant full access for a user to manage a subject.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role ResourceOwner \
        --resource Subject:${TOPIC_NAME}-value
        --kafka-cluster-id $KAFKA_ID \
        --schema-registry-cluster-id $SR_ID

Delete a topic
  - ``DeveloperManage`` role on the Kafka topic
  - ``DeveloperManage`` role on the |sr| subject, if the topic has an Avro schema
  
  These roles enable a user to delete a topic by using the DROP STREAM/TABLE
  [DELETE TOPIC] statements.

  Use the following commands to grant a user delete access to a topic and
  corresponding schema.

  .. code:: bash

      # Grant delete access for a user to a topic.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role DeveloperManage \
        --resource Topic:$TOPIC_NAME \
        --kafka-cluster-id $KAFKA_ID        

  .. code:: bash

      # Grant delete access for a user to a subject.
      confluent iam rolebinding create \
        --principal User:$USER_NAME \
        --role DeveloperManage \
        --resource Subject:${TOPIC_NAME}-value
        --kafka-cluster-id $KAFKA_ID \
        --schema-registry-cluster-id $SR_ID