ksqlDB と role-based access control (RBAC)¶

Confluent Platform 5.3 以降、ksqlDB では、クラスターリソースに対するロールベースアクセス制御（RBAC）がサポートされています。

注釈

ksqlDB は、対話型モードでの実行時に ksqlDB クラスターへの認可を行うために、RBAC をサポートしています。ユーザー用の RBAC は、ヘッドレスモードでデプロイされた ksqlDB クラスターでは使用されません。

ユーザーのロールを作成するには、Confluent CLI または Control Center を使用します。

トピックで Avro、Protobuf、または JSON_SR スキーマが使用されている場合は、Kafka クラスターのクラスター ID、ksqlDB クラスターのサービス ID、および Schema Registry クラスターのクラスター ID が必要です。詳細については、「クラスターの識別子について」を参照してください。

参考

まず最初に、自動化された RBAC の例をお試しください。Confluent Platform における RBAC の機能を紹介しています。

ksqlDB service principal¶

ksqlDB アプリケーションは、特権を持つ "ksqlDB ユーザー" service principal に付与されたロールで実行されます。デフォルトで、ksqlDB service principal の名前は "ksql" です。ksqlDB service principal にデフォルトで付与されるロールは、ksqlDB を実行するために必要であり、ksqlDB に接続して参加する対話型ユーザーに付与されるロールとは異なります。

対話型ユーザーにロールを付与するときは、ksqlDB ユーザーの service principal にも同じロールを付与する必要があります。ksqlDB アプリケーションは、バックグラウンドおよび ksqlDB クラスター内の他のサーバーで ksqlDB service principal としてユーザーコマンドを実行し、一部の ksqlDB ステートメントは長時間実行される永続的なクエリを開始します。ksqlDB は、リソースにアクセスするために必要なアクセス許可を対話型ユーザーが持っているかどうかを確認しますが、クエリ自体は ksqlDB service principal のセキュリティコンテキストで実行されます。このため、ksqlDB ステートメントに必要なすべてのリソースに対する正しいアクセス許可が、対話型ユーザーと ksqlDB service principal の両方にあることが重要になります。

次のコマンド例は、対話型ユーザーと ksqlDB service principal に、トピックへの読み取り専用アクセス権限を付与する方法を示しています。これらのロールを付与するには、UserAdmin や ResourceOwner などの十分なアクセス許可で Confluent CLI にログインする必要があります。

# Grant read-only access for a user to a topic.
confluent iam rbac role-binding 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 rbac role-binding create \
  --principal User:ksql \
  --role DeveloperRead \
  --resource Topic:<topic-name> \
  --kafka-cluster-id <kafka-id>

ksqlDB サーバーの RBAC 用構成¶

ksqlDB サーバーのプロパティファイルで以下の構成設定を割り当てて、 RBAC 環境で ksqlDB を有効にします。

Confluent セキュリティサービスにアクセスするには、パブリックキー/プライベートキーのペアを生成する必要があります。public.key.path 設定を割り当てて、パブリックキーファイルへのパスを指定します。詳細については、「Metadata Service (MDS) の構成」を参照してください。

ちなみに

confluent secret コマンドを使用すると、パスワードやその他の構成データを安全に保管できます。詳細については、「シークレット管理」を参照してください。

以下の設定を ksqlDB サーバーの構成ファイルに追加します。デフォルトでは、このファイルの名前は ${CONFLUENT_HOME}/etc/ksqldb/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>

############ ksqlDB security settings for Confluent RBAC ######################

# Set the ksqlDB cluster ID to work on specific roles and permissions for this ksqlDB cluster scope.
ksql.service.id=<ksql-cluster-id>

# Enable the ksqlDB Confluent security plugin.
ksql.security.extension.class=io.confluent.ksql.security.KsqlConfluentSecurityExtension

# Enable permissions validations used for persistent queries requests.

# By default, this setting is set to 'auto', but DESCRIBE_CONFIGS ACLs on the Kafka cluster are required for the client user.
ksql.access.validator.enable=on

# Assign the metadata URL and access credentials, which are shared by authentication security handlers and the ksqlDB 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 ksqlDB Basic+Bearer authentication.
ksql.authentication.plugin.class=io.confluent.ksql.security.VertxBearerOrBasicAuthenticationPlugin

# 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 ksqlDB user, which has the default user name of "ksql".
# These credentials authorize ksqlDB 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>";

ファイルを保存し、ksqlDB サーバーを再起動します。

ksqlDB での RBAC の設定は、Confluent Platform Metadata Service（MDS）および Confluent Server Authorizer に依存します。Metadata Service は、confluent.metadata 設定で構成します。Confluent Server Authorizer 用に ksqlDB を構成するには、ksql.security.extension.class 設定を使用します。詳細については、「Confluent Server Authorizer の構成」を参照してください。

以下の設定には、ksqlDB service principal 認証情報を使用します。

sasl.jaas.config: Confluent Server Authorizer で Kafka クラスターに対する認可を行うために使用します
confluent.metadata.basic.auth.user.info: MDS に対する認可を行うために使用します
ksql.schema.registry.basic.auth.user.info: Schema Registry に対する認可を行うために使用します

MDS と通信するクライアントが利用できる追加の構成については、Confluent Platform のセキュリティに関するドキュメントの「REST クライアントの構成」を参照してください。

ksqlDB へのアクセスのためのロールの割り当て¶

次のセクションでは、ユーザーが ksqlDB に接続し、一時クエリを含む SQL ステートメントを実行するために必要な、最小限のロールとリソースのセットを定義します。

セキュリティのためのベストプラクティスとして、ユーザーにはリソースへの必要最小限のアクセスレベルを提供します。たとえば、トピックからの読み取りを行うには、対話型ユーザーに対して ResourceOwner ロールではなく DeveloperRead ロールを付与することをお勧めします。

重要

トピックの読み取り、書き込み、リスト、および作成を試みる ksqlDB ステートメントをユーザーが使用するには、Kafka でのアクセス許可が必要です。Kafka クラスターおよび Schema Registry にあるリソースへの追加アクセス権限を明示的に付与する必要があります。詳細については、「ユーザーに対するリソースへのアクセス権限の付与」を参照してください。

これらのロールを付与するには、UserAdmin、ResourceOwner、または十分なアクセス許可のあるその他のロールとして Confluent CLI にログインする必要があります。

ksqlDB のロールマッピング¶

次の表は、RBAC の各ロールで許可されている ksqlDB の操作の一覧です。

ロール	ksqlDB クラスターへのアクセス	ksqlDB クラスターの終了	ksqlDB クラスターへの変更アクセス	ksqlDB クラスターからの表示アクセス
SystemAdmin	○	○	○	○
UserAdmin	×	×	○	○
ClusterAdmin	○	○	×	×
Operator	×	×	×	×
SecurityAdmin	×	×	×	○
ResourceOwner	○	○	○	×
DeveloperRead	×	×	×	×
DeveloperWrite	○	×	×	×
DeveloperManage	×	○	×	×

クラスター識別子¶

ロールバインディングを作成するには、Confluent Platform デプロイ内のコンポーネントのクラスター識別子が必要です。詳細については、「クラスターの識別子について」を参照してください。

以下のコマンドでは、クラスター ID に環境変数を使用しています。対話型ユーザー名とトピック名の変数も定義されています。

# Cluster identifiers
export KAFKA_ID=<kafka-cluster-id>
export KSQLDB_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>

登録済みクラスター名の使用¶

Confluent Platform 6.0 以降では、ksqlDB クラスターを登録し、わかりやすいクラスター名を指定できるため、より簡単にロールバインディングを作成できます。以下のすべてのコマンド例では、$KAFKA_ID および $KSQLDB_ID の代わりにクラスター名を使用できます。詳細については、「クラスターレジストリ」を参照してください。

たとえば、以下のコマンドをアップデートできます。

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

ksqlDB クラスターがクラスターレジストリに $KAFKA_ID および $KSQLDB_ID で登録されていても、これらの ID の代わりにわかりやすい名前を使用できます。登録されたクラスター名（この例では <ksqldb-cluster-name> ）を使用して、前のコマンドを次のようにアップデートします。

 confluent iam rbac role-binding create \
   --principal User:<cluster-owner-name> \
   --role ResourceOwner \
   --cluster-name <ksqldb-cluster-name>
   --resource KsqlCluster:ksql-cluster

ksqlDB サーバーを起動するためのロールの割り当て¶

ksqlDB サーバーを起動し、セキュアな Kafka クラスターに接続するには、UserAdmin によってユーザーに以下のロールが付与されている必要があります。

ksqlDB クラスターのスコープ¶

次のコマンドを実行して、ksqlDB クラスターに対する ResourceOwner ロールを割り当てます。これにより、ksqlDB クラスターのリソースの所有権がユーザーにバインドされます。ksqlDB クラスターに対する ResourceOwner または ClusterAdmin ロールを持つユーザーは "クラスターオーナー" です。

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

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

次のコマンドを実行して、ResourceOwner ロールを ksqlDB service principal に割り当てます。

confluent iam rbac role-binding create \
  --principal User:ksql \
  --role ResourceOwner \
  --kafka-cluster-id $KAFKA_ID \
  --ksql-cluster-id $KSQLDB_ID \
  --resource KsqlCluster:ksql-cluster

Kafka クラスターのスコープ¶

以下の内部 ksqlDB リソースに対する ResourceOwner ロールを ksqlDB service principal に付与します。デフォルトでは、これらのロールは "ksql" という名前の ksqlDB ユーザーに付与されます。

ksqlDB コマンドトピック : _confluent-ksql-${KSQLDB_ID}_command_topic
ksqlDB 処理ログのトピック : ${KSQLDB_ID}ksql_processing_log
コンシューマーグループ : _confluent-ksql-${KSQLDB_ID} というプレフィックスを使用

また、TransactionalId リソースに対する DeveloperWrite ロールを ksqlDB service principal に付与します。

TransactionalId リソース : --resource TransactionalId:${KSQLDB_ID}

ksqlDB コマンドトピックへのフルアクセス権限を ksqlDB service principal に付与するには、次のコマンドを使用します。

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

ksqlDB 処理ログトピックへのフルアクセス権限を ksqlDB service principal に付与するには、次のコマンドを使用します。

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

TransactionalId リソースへの DeveloperWrite アクセス権限を ksqlDB service principal に付与するには、次のコマンドを使用します。

# Grant DeveloperWrite access for the ksql service principal to the TransactionalId resource.
confluent iam rbac role-binding create \
  --principal User:ksql \
  --role DeveloperWrite \
  --resource TransactionalId:${KSQLDB_ID} \
  --kafka-cluster-id $KAFKA_ID

ksqlDB コンシューマーグループへのフルアクセス権限を ksql ユーザーに付与するには、次のコマンドを使用します。

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

べき等性を持つプロデューサーへのロールの割り当て¶

べき等性を持つプロデューサーおよび "厳密に 1 回" のセマンティクス（EOS）を有効にするには、Kafka クラスターに対する Write アクセス権限が ksqlDB service principal に必要です。

注釈

べき等性を持つプロデューサーによって使用されるすべてのプリンシパルは、クラスターに対する Write として認可される必要があります。Kafka クラスターに対する RBAC ロール DeveloperWrite または ResourceOwner をバインドすると、Write アクセス許可が付与されますが、2 つのロールのうち許容度が低いのは DeveloperWrite であり、こちらが第一に推奨されます。消費については、べき等性を持つコンシューマーになるために追加の Kafka のアクセス許可は必要ありません。

以下のロールバインディングにより、Write でクラスターにアクセスできるようになります。

# Grant Write access for the ksql service principal to the cluster.
confluent iam rbac role-binding create \
  --principal User:ksql \
  --role DeveloperWrite \
  --resource Cluster:kafka-cluster \
  --kafka-cluster-id $KAFKA_ID

# Grant Write access for the ksql service principal to the TransactionalId resource.
confluent iam rbac role-binding create \
  --principal User:ksql \
  --role DeveloperWrite \
  --resource TransactionalId:${KSQLDB_ID} \
  --prefix \
  --kafka-cluster-id $KAFKA_ID

ksqlDB クラスターを終了するためのロールの割り当て¶

/ksql/terminate で REST エンドポイントを呼び出して ksqlDB クラスターを終了するには、UserAdmin によってユーザーに DeveloperManage ロールが付与されている必要があります。詳細については、「クラスターの終了」を参照してください。

ksqlDB クラスターのスコープ¶

ksqlDB クラスターに対する DeveloperManage ロールを割り当てるには、次のコマンドを使用します。

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

ksqlDB アプリケーションに参加するためのロールの割り当て¶

開発のために ksqlDB クラスターにアクセスする場合、対話型ユーザーには以下のロールが必要になります。

ksqlDB クラスターのスコープ¶

ksqlDB クラスターに対する DeveloperWrite ロールを割り当てるには、次のコマンドを使用します。

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

Kafka クラスターのスコープ¶

一時クエリ用に内部コンシューマーグループに対する DeveloperRead ロールを割り当てるには、次のコマンドを使用します。

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

対話型ユーザーが ksqlDB 処理ログにアクセスする必要がある場合は、次のロールを割り当てます。

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

ユーザーに対するリソースへのアクセス権限の付与¶

必要最小限のロールに加えて、対話型ユーザーには、トピック、コンシューマーグループ、サブジェクトなど特定のリソースにアクセスするためのロールが必要です。

重要

ユーザーにロールを付与するときは、ksqlDB service principal にも同じロールを必ず付与してください。この名前は、デフォルトで "ksql" となっています。

プレフィックス付きのリソース名¶

指定されたプレフィックス文字列で始まる名前を持つすべてのリソースへのアクセス権限を付与するには、Confluent CLI の prefix オプションを使用します。たとえば、<topic-name-prefix-> で始まるすべてのトピックへの読み取り専用アクセス権限を付与するには、次のコマンドを使用できます。

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

トピック、ストリーム、テーブルにアクセスするためのロール¶

ストリームとテーブルの作成に必要なロールを対話型ユーザーに付与するには、以下の Confluent CLI コマンドを使用します。

トピックの SHOW または PRINT

Kafka のトピックに対する DeveloperRead ロール
Schema Registry サブジェクトに対する DeveloperRead ロール（トピックで Avro、Protobuf、または JSON_SR スキーマが使用されている場合）

対話型ユーザーは、このロールにより SHOW ステートメントと PRINT ステートメントを使用して、指定されたトピックを表示できます。また、ユーザーはこれらのトピックからストリームとテーブルの CREATE を実行できます。

これらのステートメントで、ksqlDB service principal には、トピックに対するロールが必要ありません。

注釈

サブジェクトの名前は、トピックの名前の末尾に -value を付加したものです。

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

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

ストリームまたはテーブルからの SELECT

ソーストピックに対する DeveloperRead ロール
_confluent-ksql-${KSQLDB_ID} コンシューマーグループに対する DeveloperRead ロール
_confluent-ksql-${KSQLDB_ID}transient 一時トピックに対する ResourceOwner ロール（テーブル用）
Schema Registry サブジェクトに対する DeveloperRead ロール（トピックで Avro、Protobuf、または JSON_SR スキーマが使用されている場合）
_confluent-ksql-<cluster-id>transient* サブジェクトに対する ResourceOwner ロール（Avro を使用するテーブル用、ストリームには不要）

ユーザーはこれらのロールにより、SELECT ステートメントを使用してストリームまたはテーブルから読み取ることができます。SELECT ステートメントに、認可されていないトピックを使用する JOIN が含まれている場合、この SELECT は認可エラーで失敗します。

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

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

# For tables that use Avro: grant access to the transient subjects.
confluent iam rbac role-binding create \
  --principal User:$USER_NAME \
  --role ResourceOwner \
  --resource Subject:_confluent-ksql-${KSQLDB_ID}transient \
  --prefix \
  --kafka-cluster-id $KAFKA_CLUSTER_ID \
  --schema-registry-cluster-id $SR_ID

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

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

INSERT によるトピックへの書き込み

Kafka のトピックに対する DeveloperWrite ロール
Schema Registry サブジェクトに対する ResourceOwner または DeveloperWrite ロール（トピックで Avro、Protobuf、または JSON_SR スキーマが使用されている場合）

ユーザーはこれらのロールにより、INSERT ステートメントを使用してデータを書き込むことができます。INSERT INTO ステートメントには、SELECT 句が含まれています。ユーザーがこれを使用するには、クエリ内のトピックに対する読み取りアクセス許可が必要になります。

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

# Grant full access for a user to create a subject and write to it.
confluent iam rbac role-binding 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 ロール
_confluent-ksql-${KSQLDB_ID} コンシューマーグループに対する DeveloperRead ロール
Schema Registry サブジェクトに対する ResourceOwner または DeveloperWrite ロール（ソーストピックで Avro、Protobuf、または JSON_SR スキーマが使用されている場合）

対話型ユーザーはこれらのロールにより、CREATE STREAM ステートメントを使用して、指定されたトピックにストリームまたはテーブルを登録できます。

トピックに Avro、Protobuf、または JSON_SR スキーマが使用されている場合、対話型ユーザーと ksqlDB service principal には、Schema Registry のサブジェクトへのフルアクセス権限が必要になります。

# Grant read-only access for a user to a topic.
confluent iam rbac role-binding 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 rbac role-binding create \
  --principal User:ksql \
  --role DeveloperRead \
  --resource Topic:$TOPIC_NAME \
  --kafka-cluster-id $KAFKA_ID

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

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

# Grant full access for a user to create a subject and write to it.
confluent iam rbac role-binding 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 ksqlDB service principal to create a subject and write to it.
confluent iam rbac role-binding 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 ロール
_confluent-ksql-${KSQLDB_ID}transient 一時トピックに対する ResourceOwner ロール
Schema Registry サブジェクトに対する ResourceOwner または DeveloperWrite ロール（ソーストピックで Avro、Protobuf、または JSON_SR スキーマが使用されている場合）
_confluent-ksql-${KSQLDB_ID}* サブジェクトに対する ResourceOwner ロール（Avro、Protobuf、または JSON_SR を使用するテーブル用）

対話型ユーザーはこれらのロールにより、CREATE TABLE ステートメントを使用して、指定されたトピックにテーブルを登録できます。

トピックに Avro、Protobuf、または JSON_SR スキーマが使用されている場合、対話型ユーザーと ksqlDB service principal には、Schema Registry のサブジェクトへのフルアクセス権限が必要になります。

注釈

一時トピックに対する ResourceOwner ロールは、KSQL テーブルの制限です。プレフィックスの付いたトピックに対してこのアクセス許可を付与すると、ユーザーは他のクエリのテーブルを表示できます。

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

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

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

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

# Grant full access for a user to create a subject and write to it.
confluent iam rbac role-binding 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 rbac role-binding create \
  --principal User:ksql \
  --role ResourceOwner \
  --resource Subject:${TOPIC_NAME}-value \
  --kafka-cluster-id $KAFKA_ID \
  --schema-registry-cluster-id $SR_ID

# For tables that use Avro, Protobuf, or JSON_SR:
# Grant full access for the ksql service principal to all internal ksql subjects.
confluent iam rbac role-binding create \
  --principal User:ksql \
  --role ResourceOwner \
  --resource Subject:_confluent-ksql-${KSQLDB_ID} \
  --prefix \
  --kafka-cluster-id $KAFKA_ID \
  --schema-registry-cluster-id $SR_ID

永続的なクエリを使用したストリームまたはテーブルの作成

ソーストピックに対する DeveloperRead ロール
ksqlDB シンクトピックに対する ResourceOwner ロール
Schema Registry シンクサブジェクトに対する ResourceOwner または DeveloperWrite ロール（ソーストピックで Avro、Protobuf、または JSON_SR スキーマが使用されている場合）
_confluent-ksql-${KSQLDB_ID}* サブジェクトに対する DeveloperRead ロール（Avro、Protobuf、または JSON_SR を使用するテーブル用）

ユーザーはこれらのロールにより、永続的なクエリを使用してストリームとテーブルを作成できます。ksqlDB では新しい "シンクトピック" を作成するため、ユーザーにはシンクトピックの作成、読み取り、書き込みのための十分なアクセス許可が必要になります。対話型ユーザーと ksqlDB service principal には、シンクトピック（まだ存在しない場合）を作成するためのアクセス許可が必要であるため、シンクトピックに対する ResourceOwner ロールが必要になります。

注釈

シンクトピックはストリームまたはテーブルと同じ名前で、すべて大文字です。

トピックに Avro スキーマが使用されている場合、対話型ユーザーと ksqlDB service principal には、Schema Registry のシンクトピックのサブジェクトへの ResourceOwner または DeveloperWrite アクセス許可が必要になります。

永続的なクエリで作成され、Avro、Protobuf、または JSON_SR が使用されれているテーブルの場合、ksqlDB service principal には、すべての内部サブジェクトに対する DeveloperRead アクセス許可が必要になります。

注釈

サブジェクトの名前は、シンクトピックの名前の末尾に -value を付加したものです。

# Grant read-only access for a user to a topic.
confluent iam rbac role-binding 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 rbac role-binding create \
  --principal User:ksql \
  --role DeveloperRead \
  --resource Topic:$SOURCE_TOPIC_NAME \
  --kafka-cluster-id $KAFKA_ID

# Grant read-only access for a user to a topic.
confluent iam rbac role-binding 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 rbac role-binding create \
  --principal User:ksql \
  --role ResourceOwner \
  --resource Topic:$SINK_TOPIC_NAME \
  --kafka-cluster-id $KAFKA_ID

# Grant full access for a user to create a subject and write to it.
confluent iam rbac role-binding 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 rbac role-binding 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, Protobuf, or JSON_SR created with a persistent query:
# Grant read access for the ksql service principal to all internal ksql subjects.
confluent iam rbac role-binding create \
  --principal User:ksql \
  --role DeveloperRead \
  --resource Subject:_confluent-ksql-${KSQLDB_ID} \
  --prefix \
  --kafka-cluster-id $KAFKA_ID \
  --schema-registry-cluster-id $SR_ID

トピックとスキーマに対するフルコントロール

Kafka のトピックに対する ResourceOwner ロール
Schema Registry サブジェクトに対する ResourceOwner ロール（トピックで Avro、Protobuf、または JSON_SR スキーマが使用されている場合）

トピックとスキーマを削除するアクセス許可を含め、トピックとスキーマに対するフルコントロールをユーザーに付与するには、以下のコマンドを使用します。

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

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

トピックの削除

Kafka のトピックに対する DeveloperManage ロール
Schema Registry サブジェクトに対する DeveloperManage ロール（トピックで Avro、Protobuf、または JSON_SR スキーマが使用されている場合）

ユーザーはこれらのロールにより、DROP STREAM/TABLE [DELETE TOPIC] ステートメントを使用してトピックを削除できます。

トピックおよび対応するスキーマへの削除アクセス権限をユーザーに付与するには、以下のコマンドを使用します。

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

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