14.5. OAuth 2.0 토큰 기반 권한 부여 사용

절차

1. Red Hat Single Sign-On 설명서의 서버 설치 및 구성에 설명된 대로 Red Hat Single Sign-On Operator를 사용하여 Red Hat Single Sign-On 서버를 설치합니다.
2. Red Hat Single Sign-On 인스턴스가 실행될 때까지 기다립니다.

3. 관리 콘솔에 액세스할 수 있도록 외부 호스트 이름을 가져옵니다.

   NS=sso
   oc get ingress keycloak -n $NS

   이 예에서는 Red Hat Single Sign-On 서버가 sso 네임스페이스에서 실행되고 있다고 가정합니다.

4. admin 사용자의 암호를 가져옵니다.

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

   암호는 시크릿으로 저장되므로 Red Hat Single Sign-On 인스턴스의 구성 YAML 파일을 가져와 시크릿 이름을 확인합니다(secretKeyRef.name).

5. 보안 이름을 사용하여 일반 텍스트 암호를 가져옵니다.

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

   이 예제에서는 시크릿 이름이 credentials -keycloak 이라고 가정합니다.

6. 사용자 이름 admin 과 가져온 암호를 사용하여 관리 콘솔에 로그인합니다.

   https://HOSTNAME 을 사용하여 Kubernetes Ingress 에 액세스합니다.

   이제 Admin Console을 사용하여 예제 영역을 Red Hat Single Sign-On에 업로드할 수 있습니다.

7. Add CloudEvent를 클릭하여 예제 영역을 가져옵니다.

8. examples/security/keycloak-authorization/kafka-authz-realm.json 파일을 추가한 다음 Create 를 클릭합니다.

   이제 관리 콘솔의 현재 영역으로 kafka-authz 가 있습니다.

   기본 보기는 마스터 영역을 표시합니다.

9. Red Hat Single Sign-On 관리 콘솔에서 클라이언트 > kafka > Authorization > Settings 로 이동하여 결정 전략이 승인으로 설정되어 있는지 확인합니다.

   검증 정책은 클라이언트가 Kafka 클러스터에 액세스하기 위해 하나 이상의 정책을 충족해야 함을 의미합니다.

10. Red Hat Single Sign-On 관리 콘솔에서 그룹,사용자,역할, 클라이언트로 이동 하여 영역 구성을 확인합니다.

    그룹

    그룹은 사용자 그룹을 생성하고 사용자 권한을 설정하는 데 사용됩니다. groups은 이름이 할당된 사용자 집합입니다. 사용자를 지역, 조직 또는 부서 단위로 구분하는 데 사용됩니다. 그룹을 LDAP ID 공급자에 연결할 수 있습니다. 예를 들어 Kafka 리소스에 대한 권한을 부여하기 위해 사용자 정의 LDAP 서버 관리자 사용자 인터페이스를 통해 사용자를 그룹 멤버로 만들 수 있습니다.

    사용자

    사용자는 사용자를 생성하는 데 사용됩니다. 이 예제에서는 alice 및 bob 가 정의됩니다. Alice 는 ClusterManager 그룹의 멤버이며 bob 는 ClusterManager-my-cluster 그룹의 멤버입니다. 사용자는 LDAP ID 공급자에 저장할 수 있습니다.

    역할

    역할은 사용자 또는 클라이언트를 특정 권한을 갖는 것으로 표시합니다. 역할은 그룹과 유사한 개념입니다. 일반적으로 조직의 역할을 가진 사용자를 태그 하고 필수 권한을 갖는 데 사용됩니다. 역할은 LDAP ID 공급자에 저장할 수 없습니다. LDAP가 요구 사항인 경우 대신 그룹을 사용하고 Red Hat Single Sign-On 역할을 그룹에 추가하여 사용자에게 해당 역할을 할당받을 수 있습니다.

    클라이언트

    클라이언트에 는 특정 구성이 있을 수 있습니다. 예에서는 kafka,kafka-cli,team-a-client 및 team-b-client 클라이언트가 구성됩니다.

    • kafka 클라이언트는 Kafka 브로커가 액세스 토큰 검증에 필요한 OAuth 2.0 통신을 수행하는 데 사용됩니다. 이 클라이언트에는 Kafka 브로커에서 권한을 수행하는 데 사용되는 권한 부여 서비스 리소스 정의, 정책 및 권한 부여 범위도 포함되어 있습니다. 권한 부여 구성은 Authorization Enabled 가 Settings 탭에서 전환될 때 표시되는 Authorization 탭의 kafka 클라이언트에 정의되어 있습니다.
    • kafka-cli 클라이언트는 액세스 토큰 또는 새로 고침 토큰을 가져오기 위해 사용자 이름 및 암호로 인증할 때 Kafka 명령줄 툴에서 사용하는 공용 클라이언트입니다.
    • team-a-client 및 team-b-client 클라이언트는 특정 Kafka 항목에 대한 부분 액세스 권한이 있는 서비스를 나타내는 기밀 클라이언트입니다.

11. Red Hat Single Sign-On 관리 콘솔에서 권한 부여 > 권한으로 이동하여 영역에 정의된 리소스 및 정책을 사용하는 승인된 권한을 확인합니다.

    예를 들어 kafka 클라이언트에는 다음과 같은 권한이 있습니다.

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

    Dev Team A

    Dev Team A realm 역할은 모든 클러스터에서 x_ 로 시작하는 항목에 쓸 수 있습니다. 이를 통해 Topic:x_*, Describe 및 Write 범위, Dev Team A 정책이 결합되어 있습니다. Dev Team A 정책은 Dev Team A 라는 영역 역할이 있는 모든 사용자와 일치합니다.

    Dev Team B

    Dev Team B 영역 역할은 모든 클러스터에서 x_ 로 시작하는 주제에서 읽을 수 있습니다. 이를 통해 Topic:x_*, Group:x_* 리소스, 설명 및 읽기 범위, Dev Team B 정책이 결합되어 있습니다. Dev Team B 정책은 Dev Team B 라는 영역 역할이 있는 모든 사용자와 일치합니다. 사용자 및 클라이언트 일치에는 주제에서 읽고 x_ 로 시작하는 이름이 있는 주제 및 소비자 그룹에 대해 소비된 오프셋을 업데이트할 수 있습니다.

절차

1. 배포된 Red Hat Single Sign-On 인스턴스의 호스트 이름을 사용하여 Kafka 브로커가 Red Hat Single Sign-On 서버와 통신할 수 있도록 신뢰 저장소 인증서를 준비합니다.

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

   Kubernetes Ingress 를 사용하여 보안(HTTPS)을 연결하면 인증서가 필요합니다.

   일반적으로 단일 인증서는 없지만 인증서 체인이 있습니다. /tmp/sso.pem 파일에 마지막으로 나열된 최상위 발급자 CA만 제공해야 합니다. 수동으로 추출하거나 다음 명령을 사용할 수 있습니다.

   인증서 체인에서 최상위 CA 인증서를 추출하는 명령 예

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

   참고

   신뢰할 수 있는 CA 인증서는 일반적으로 openssl 명령을 사용하는 것이 아니라 신뢰할 수 있는 소스에서 가져옵니다.

2. OpenShift에 인증서를 시크릿으로 배포합니다.

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

3. 호스트 이름을 환경 변수로 설정

   SSO_HOST=SSO-HOSTNAME

4. 예제 Kafka 클러스터를 생성하고 배포합니다.

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

절차

1. AMQ Streams Kafka 이미지를 사용하여 새 대화형 Pod 컨테이너를 실행하여 실행 중인 Kafka 브로커에 연결합니다.

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

   참고

   이미지 다운로드에서 oc 시간이 초과되면 후속 시도로 인해 AlreadyExists 오류가 발생할 수 있습니다.

2. Pod 컨테이너에 연결합니다.

   oc attach -ti kafka-cli -n $NS

3. Red Hat Single Sign-On 인스턴스의 호스트 이름을 사용하여 TLS를 사용한 클라이언트 연결 인증서를 준비합니다.

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

   일반적으로 단일 인증서는 없지만 인증서 체인이 있습니다. /tmp/sso.pem 파일에 마지막으로 나열된 최상위 발급자 CA만 제공해야 합니다. 수동으로 추출하거나 다음 명령을 사용할 수 있습니다.

   인증서 체인에서 최상위 CA 인증서를 추출하는 명령 예

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

   참고

   신뢰할 수 있는 CA 인증서는 일반적으로 openssl 명령을 사용하는 것이 아니라 신뢰할 수 있는 소스에서 가져옵니다.

4. Kafka 브로커에 대한 TLS 연결에 대한 신뢰 저장소를 생성합니다.

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

5. Kafka 부트스트랩 주소를 Kafka 브로커의 호스트 이름으로 사용하고 tls 리스너 포트(9093)를 사용하여 Kafka 브로커에 대한 인증서를 준비합니다.

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

   가져온 .pem 파일은 일반적으로 단일 인증서가 아니라 인증서 체인입니다. /tmp/my-cluster-kafka.pem 파일에 마지막으로 나열된 최상위 발급자 CA만 제공해야 합니다. 수동으로 추출하거나 다음 명령을 사용할 수 있습니다.

   인증서 체인에서 최상위 CA 인증서를 추출하는 명령 예

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

   참고

   신뢰할 수 있는 CA 인증서는 일반적으로 openssl 명령을 사용하는 것이 아니라 신뢰할 수 있는 소스에서 가져옵니다. 이 예제에서는 클라이언트가 Kafka 클러스터가 배포된 동일한 네임스페이스의 Pod에서 실행되고 있다고 가정합니다. 클라이언트가 OpenShift 클러스터 외부에서 Kafka 클러스터에 액세스하는 경우 먼저 부트스트랩 주소를 확인해야 합니다. 이 경우 OpenShift 시크릿에서 직접 클러스터 인증서를 가져올 수도 있으며 openssl 이 필요하지 않습니다. 자세한 내용은 13장. Kafka 클러스터에 대한 클라이언트 액세스 설정의 내용을 참조하십시오.

6. Kafka 브로커의 인증서를 신뢰 저장소에 추가합니다.

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

   권한이 부여된 액세스 권한을 확인하기 위해 세션을 열린 상태로 유지합니다.

클라이언트 및 관리자 구성 설정

1. team-a-client 클라이언트에 대한 인증 속성을 사용하여 Kafka 구성 파일을 준비합니다.

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

   SASL OAUTHBLOGER 메커니즘이 사용됩니다. 이 메커니즘을 사용하려면 클라이언트 ID 및 클라이언트 보안이 필요합니다. 즉, 클라이언트가 먼저 Red Hat Single Sign-On 서버에 연결하여 액세스 토큰을 가져옵니다. 그러면 클라이언트는 Kafka 브로커에 연결하고 액세스 토큰을 사용하여 인증합니다.

2. team-b-client 클라이언트에 대한 인증 속성을 사용하여 Kafka 구성 파일을 준비합니다.

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

3. curl 을 사용하여 admin 사용자 alice 를 인증하고 암호 부여 인증을 수행하여 새로 고침 토큰을 가져옵니다.

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

   새로 고침 토큰은 수명이 길며 만료되지 않는 오프라인 토큰입니다.

4. admin 사용자의 인증 속성을 사용하여 Kafka 구성 파일을 준비합니다.

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

   kafka-cli 공용 클라이언트는 sasl.jaas.config 의 oauth.client.id 에 사용됩니다. 이는 공용 클라이언트이므로 보안이 필요하지 않습니다. 클라이언트는 이전 단계에서 인증된 새로 고침 토큰으로 인증됩니다. 새로 고침 토큰은 인증을 위해 Kafka 브로커로 전송되는 뒤에서 액세스 토큰을 요청합니다.

1. my-topic 에 주제를 작성합니다.

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

   이 요청은 주제에 액세스할 수 있는 Not authorized to access topics: [my-topic] 오류를 반환합니다.

   team-a-client 에는 a_ 로 시작하는 항목에 대해 지원되는 모든 작업을 수행할 수 있는 권한을 부여하는 Dev Team A 역할이 있지만 x_ 로 시작하는 항목에만 쓸 수 있습니다. my-topic 이라는 주제는 해당 규칙과 일치하지 않습니다.

2. a_ECDHE에 주제를 작성합니다.

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

   Kafka에 메시지가 성공적으로 생성됩니다.

3. Ctrl+C를 눌러 CLI 애플리케이션을 종료합니다.

4. 요청에 대한 Authorization GRANTED 의 디버그 로그가 Kafka 컨테이너 로그를 확인합니다.

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

1. 주제 a_ECDHE에서 메시지를 가져옵니다.

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

   team-a-client 의 Dev Team A 역할은 a_ 로 시작하는 이름이 있는 소비자 그룹에만 액세스할 수 있기 때문에 요청이 오류를 반환합니다.

2. team-a-client 속성을 업데이트하여 사용할 수 있는 사용자 지정 소비자 그룹을 지정합니다.

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

   소비자는 a_ECDHE 주제에서 모든 메시지를 수신합니다.

1. 주제 나열.

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

   a_ECDHE 주제가 반환됩니다.

2. 소비자 그룹을 나열합니다.

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

   a_consumer_group_1 소비자 그룹이 반환됩니다.

   클러스터 구성에 대한 세부 정보를 가져옵니다.

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

   작업에 team-a-client 가 없는 클러스터 수준 권한이 필요하므로 요청이 오류를 반환합니다.

1. a_ECDHE에 주제를 작성합니다.

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

   이 요청은 항목에 액세스할 수 있는 Not authorized to access topics: [a_ECDHE] 오류를 반환합니다.

2. 기록 b_ECDHE.

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

   Kafka에 메시지가 성공적으로 생성됩니다.

3. x_ECDHE 주제를 작성할 수 있습니다.

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

   항목에 액세스할 수 있는 권한이 없음: [x_ECDHE] 오류가 반환되고 team-b-client 는 주제 x_ECDHE 에서만 읽을 수 있습니다.

4. team-a-client 를 사용하여 x_ ECDHE 주제를 작성합니다.

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

   이 요청은 항목에 액세스할 수 있는 Not authorized to access topics: [x_ECDHE] 오류를 반환합니다. team-a-client 는 x_ ECDHE 항목에 쓸 수 있지만 아직 없는 경우 주제를 만들 수 있는 권한이 없습니다. team-a-client 가 x_ ECDHE 항목에 쓰기 전에 관리자의 고급 사용자가 파티션 및 복제본 수와 같은 올바른 구성으로 생성해야 합니다.

1. x_ECDHE 주제를 alice 로 만듭니다.

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

   주제가 성공적으로 생성되었습니다.

2. 모든 주제를 alice 로 나열합니다.

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

   admin 사용자 alice 는 모든 주제를 나열할 수 있지만 team-a-client 및 team-b-client 는 액세스할 수 있는 주제만 나열할 수 있습니다.

   Dev Team A 및 Dev Team B 역할에는 모두 x_ 로 시작하는 항목에 대한 자세한 권한이 있지만 해당 역할에 대한 자세한 권한이 없기 때문에 다른 팀의 주제를 볼 수 없습니다.

3. team-a-client 를 사용하여 x_ECDHE 항목에 대한 메시지를 생성합니다.

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

   alice 가 x_ECDHE 주제를 만들었으므로 Kafka에 메시지가 성공적으로 생성됩니다.

4. team-b-client 를 사용하여 x_ECDHE 항목에 대한 메시지를 생성합니다.

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

   이 요청은 항목에 액세스할 수 있는 Not authorized to access topics: [x_ECDHE] 오류를 반환합니다.

5. team-b-client 를 사용하여 x_ECDHE 주제의 메시지를 사용합니다.

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

   소비자는 x_ECDHE 주제에서 모든 메시지를 수신합니다.

6. team-a-client 를 사용하여 x_ECDHE 주제의 메시지를 사용합니다.

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

   이 요청은 항목에 액세스할 수 있는 Not authorized to access topics: [x_ECDHE] 오류를 반환합니다.

7. team-a-client 를 사용하여 a_ 로 시작하는 소비자 그룹의 메시지를 사용합니다.

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

   이 요청은 항목에 액세스할 수 있는 Not authorized to access topics: [x_ECDHE] 오류를 반환합니다.

   Dev Team A 에는 x_ 로 시작하는 항목에 대한 읽기 액세스 권한이 없습니다.

8. alice 를 사용하여 x_ECDHE 항목에 대한 메시지를 생성합니다.

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

   Kafka에 메시지가 성공적으로 생성됩니다.

   Alice 는 모든 주제에서 읽거나 쓸 수 있습니다.

9. alice 를 사용하여 클러스터 구성을 읽습니다.

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

   이 예제의 클러스터 구성은 비어 있습니다.