6.4. Kafka 구성

1. Kafka 클라이언트는 클라이언트 ID 및 시크릿과 선택적으로 새로 고침 토큰을 사용하여 권한 부여 서버에서 액세스 토큰을 요청합니다. 또는 클라이언트는 사용자 이름과 암호를 사용하여 인증할 수 있습니다.
2. 권한 부여 서버는 새 액세스 토큰을 생성합니다.
3. Kafka 클라이언트는 SASL OAUTHB>-<ER 메커니즘을 사용하여 액세스 토큰을 전달하는 Kafka 브로커로 인증합니다.
4. Kafka 브로커는 자체 클라이언트 ID 및 시크릿을 사용하여 권한 부여 서버에서 토큰 인트로스펙션 끝점을 호출하여 액세스 토큰을 검증합니다.
5. 토큰이 유효한 경우 Kafka 클라이언트 세션이 설정됩니다.

1. Kafka 클라이언트는 클라이언트 ID 및 시크릿을 사용하여 토큰 끝점에서 권한 부여 서버와 선택적으로 새로 고침 토큰을 사용하여 인증합니다. 또는 클라이언트는 사용자 이름과 암호를 사용하여 인증할 수 있습니다.
2. 권한 부여 서버는 새 액세스 토큰을 생성합니다.
3. Kafka 클라이언트는 SASL OAUTHB>-<ER 메커니즘을 사용하여 액세스 토큰을 전달하는 Kafka 브로커로 인증합니다.
4. Kafka 브로커는 JWT 토큰 서명 확인 및 로컬 토큰 인트로스펙션을 사용하여 액세스 토큰을 로컬에서 검증합니다.

1. Kafka 클라이언트는 SASL OAUTHB>-<ER 메커니즘을 사용하여 장기간의 액세스 토큰을 전달하는 Kafka 브로커로 인증합니다.
2. Kafka 브로커는 자체 클라이언트 ID 및 시크릿을 사용하여 권한 부여 서버에서 토큰 인트로스펙션 끝점을 호출하여 액세스 토큰을 검증합니다.
3. 토큰이 유효한 경우 Kafka 클라이언트 세션이 설정됩니다.

1. Kafka 클라이언트는 SASL OAUTHB>-<ER 메커니즘을 사용하여 장기간의 액세스 토큰을 전달하는 Kafka 브로커로 인증합니다.
2. Kafka 브로커는 JWT 토큰 서명 확인 및 로컬 토큰 인트로스펙션을 사용하여 로컬에서 액세스 토큰을 검증합니다.

1. Kafka 클라이언트는 clientId 를 사용자 이름으로, 시크릿 을 암호로 전달합니다.
2. Kafka 브로커는 토큰 끝점을 사용하여 clientId 및 시크릿 을 권한 부여 서버에 전달합니다.
3. 클라이언트 인증 정보가 유효하지 않은 경우 권한 부여 서버에서 새 액세스 토큰 또는 오류를 반환합니다.

4. Kafka 브로커는 다음 방법 중 하나로 토큰을 검증합니다.

   1. 토큰 인트로스펙션 엔드 포인트가 지정되면 Kafka 브로커는 권한 부여 서버에서 끝점을 호출하여 액세스 토큰을 검증합니다. 토큰 검증에 성공하면 세션이 설정됩니다.
   2. 로컬 토큰 인트로스펙션이 사용되는 경우 권한 부여 서버에 대한 요청이 수행되지 않습니다. Kafka 브로커는 JWT 토큰 서명 검사를 사용하여 로컬에서 액세스 토큰을 검증합니다.

1. Kafka 클라이언트는 사용자 이름과 암호를 전달합니다. password는 클라이언트를 실행하기 전에 수동으로 가져와 구성한 액세스 토큰의 값을 제공합니다.

2. 암호는 Kafka 브로커 리스너가 인증을 위해 토큰 끝점으로 구성되었는지에 따라 $accessToken: 문자열 접두사를 사용하여 전달됩니다.

   1. 토큰 끝점이 구성된 경우 브로커에 클라이언트 시크릿 대신 액세스 토큰이 포함되어 있음을 알리기 위해 암호 앞에 $accessToken: 접두사가 추가해야 합니다. Kafka 브로커는 사용자 이름을 계정 사용자 이름으로 해석합니다.
   2. Kafka 브로커 리스너에 토큰 끝점이 구성되지 않은 경우( no-client-credentials 모드강화) 암호는 접두사 없이 액세스 토큰을 제공해야 합니다. Kafka 브로커는 사용자 이름을 계정 사용자 이름으로 해석합니다. 이 모드에서 클라이언트는 클라이언트 ID와 시크릿을 사용하지 않으며 password 매개변수는 항상 원시 액세스 토큰으로 해석됩니다.

3. Kafka 브로커는 다음 방법 중 하나로 토큰을 검증합니다.

   1. 토큰 인트로스펙션 엔드 포인트가 지정되면 Kafka 브로커는 권한 부여 서버에서 끝점을 호출하여 액세스 토큰을 검증합니다. 토큰 검증에 성공하면 세션이 설정됩니다.
   2. 로컬 토큰 인트로스펙션을 사용하는 경우 권한 부여 서버에 대한 요청이 없습니다. Kafka 브로커는 JWT 토큰 서명 검사를 사용하여 로컬에서 액세스 토큰을 검증합니다.

절차

1. Red Hat Single Sign-On 설치.

   ZIP 파일에서 또는 RPM을 사용하여 설치할 수 있습니다.

2. Red Hat Single Sign-On 관리 콘솔에 로그인하여 AMQ Streams에 대한 OAuth 2.0 정책을 생성합니다.

   Red Hat Single Sign-On을 배포할 때 로그인 세부 정보가 제공됩니다.

3. 영역을 생성하고 활성화합니다.

   기존 마스터 영역을 사용할 수 있습니다.

4. 필요한 경우 영역의 세션 및 토큰 타임아웃을 조정합니다.
5. kafka-broker 라는 클라이언트를 생성합니다.

6. 설정 탭에서 다음을 설정합니다.

   • 기밀성에 대한 액세스 유형
   • 이 클라이언트에 대한 웹 로그인을 비활성화하려면 Standard Flow Enabled to OFF
   • 이 클라이언트가 자체 이름으로 인증할 수 있도록 서비스 계정이 ON 에 활성화됨
7. 계속하기 전에 저장 을 클릭합니다.
8. 인증 정보 탭에서 AMQ Streams Kafka 클러스터 구성에서 사용할 시크릿을 기록합니다.

9. Kafka 브로커에 연결할 모든 애플리케이션 클라이언트에 대해 클라이언트 생성 단계를 반복합니다.

   각 새 클라이언트에 대한 정의를 생성합니다.

   구성의 클라이언트 ID로 이름을 사용합니다.

절차

1. server.properties 파일에서 Kafka 브로커 리스너 구성을 구성합니다.

   예를 들어 OAUTHBLOGER 메커니즘을 사용합니다.

   sasl.enabled.mechanisms=OAUTHBEARER
   listeners=CLIENT://0.0.0.0:9092
   listener.security.protocol.map=CLIENT:SASL_PLAINTEXT
   listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER
   sasl.mechanism.inter.broker.protocol=OAUTHBEARER
   inter.broker.listener.name=CLIENT
   listener.name.client.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler
   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required ;
   listener.name.client.oauthbearer.sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

2. broker 연결 설정을 listener.name.client.oauthbearer.sasl.jaas.config 의 일부로 구성합니다.

   다음 예제에서는 연결 구성 옵션을 보여줍니다.

   예 1: JWKS 끝점 구성을 사용한 로컬 토큰 검증

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.valid.issuer.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME" \
     oauth.jwks.endpoint.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME/protocol/openid-connect/certs" \
     oauth.jwks.refresh.seconds="300" \
     oauth.jwks.refresh.min.pause.seconds="1" \
     oauth.jwks.expiry.seconds="360" \
     oauth.username.claim="preferred_username" \
     oauth.ssl.truststore.location="PATH-TO-TRUSTSTORE-P12-FILE" \
     oauth.ssl.truststore.password="TRUSTSTORE-PASSWORD" \
     oauth.ssl.truststore.type="PKCS12" ;
   listener.name.client.oauthbearer.connections.max.reauth.ms=3600000

   예 2: OAuth 2.0 인트로스펙션 끝점을 통해 인증 서버에 토큰 검증 제공

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.introspection.endpoint.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME/protocol/openid-connect/introspection" \
     # ...

3. 필요한 경우 권한 부여 서버에 대한 액세스를 구성합니다.

   서비스 메시 와 같은 기술을 사용하여 컨테이너 외부의 보안 채널을 구성하지 않는 한, 프로덕션 환경에 이 단계가 일반적으로 필요합니다.

   1. 보안 권한 부여 서버에 연결하기 위한 사용자 정의 신뢰 저장소를 제공합니다. 권한 부여 서버에 액세스하려면 항상 SSL이 필요합니다.

      속성을 설정하여 신뢰 저장소를 구성합니다.

      예를 들면 다음과 같습니다.

      listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
        # ...
        oauth.client.id="kafka-broker" \
        oauth.client.secret="kafka-broker-secret" \
        oauth.ssl.truststore.location="PATH-TO-TRUSTSTORE-P12-FILE" \
        oauth.ssl.truststore.password="TRUSTSTORE-PASSWORD" \
        oauth.ssl.truststore.type="PKCS12" ;

   2. 인증서 호스트 이름이 액세스 URL 호스트 이름과 일치하지 않으면 인증서 호스트 이름 검증을 비활성화할 수 있습니다.

      oauth.ssl.endpoint.identification.algorithm=""

      이 검사를 통해 권한 부여 서버에 대한 클라이언트 연결이 인증됨을 확인합니다. 프로덕션 환경 외 환경에서 검증을 끄는 것이 좋습니다.

4. 선택한 인증 흐름에 따라 추가 속성을 구성합니다.

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     # ...
     oauth.token.endpoint.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME/protocol/openid-connect/token" \ 1
     oauth.custom.claim.check="@.custom == 'custom-value'" \ 2
     oauth.scope="SCOPE" \ 3
     oauth.check.audience="true" \ 4
     oauth.audience="AUDIENCE" \ 5
     oauth.valid.issuer.uri="https://https://AUTH-SERVER-ADDRESS/auth/REALM-NAME" \ 6
     oauth.client.id="kafka-broker" \ 7
     oauth.client.secret="kafka-broker-secret" \ 8
     oauth.connect.timeout.seconds=60 \ 9
     oauth.read.timeout.seconds=60 \ 10
     oauth.http.retries=2 \ 11
     oauth.http.retry.pause.millis=300 \ 12
     oauth.groups.claim="$.groups" \ 13
     oauth.groups.claim.delimiter="," ; 14

   1

   권한 부여 서버에 대한 OAuth 2.0 토큰 끝점 URL입니다. 프로덕션의 경우 항상 https:// URL을 사용합니다. KeycloakRBACAuthorizer 를 사용하거나 OAuth 2.0이 활성화된 리스너가 연결 간 통신에 사용되는 경우 필수입니다.

   2

   (선택 사항) 사용자 정의 클레임 확인. 검증 중에 추가 사용자 지정 규칙을 JWT 액세스 토큰에 적용하는 JsonPath 필터 쿼리입니다. 액세스 토큰에 필요한 데이터가 포함되어 있지 않으면 거부됩니다. 인트로스펙션 끝점 방법을 사용하는 경우 사용자 정의 검사가 인트로스펙션 끝점 응답 JSON에 적용됩니다.

   3

   (선택 사항) 토큰 끝점에 전달되는 범위 매개변수입니다. 범위는broker 간 인증을 위한 액세스 토큰을 가져올 때 사용됩니다. 또한 clientId 및 시크릿 을 사용하여 PLAIN 클라이언트 인증을 통해 OAuth 2.0에 대한 클라이언트 이름에 사용됩니다. 권한 부여 서버에 따라 토큰과 토큰의 콘텐츠만 가져올 수 있습니다. 리스너의 토큰 검증 규칙에는 영향을 미치지 않습니다.

   4

   (선택 사항) #177 검사. 권한 부여 서버에서 aud (audience) 클레임을 제공하고 대상 검사를 적용하려면 ouath.check.audience 를 true 로 설정합니다. 대상 검사에서 의도된 토큰 수신자를 식별합니다. 결과적으로 Kafka 브로커는 aud 클레임에 clientId 가 없는 토큰을 거부합니다. 기본값은 false 입니다.

   5

   (선택 사항) 토큰 끝점에 전달된 대상 매개변수입니다. 구독자 간 인증을 위한 액세스 토큰을 가져올 때 대상 사용자가 사용됩니다.An audience is used when obtaining an access token for inter-broker authentication. 또한 clientId 및 시크릿 을 사용하여 PLAIN 클라이언트 인증을 통해 OAuth 2.0에 대한 클라이언트 이름에 사용됩니다. 권한 부여 서버에 따라 토큰과 토큰의 콘텐츠만 가져올 수 있습니다. 리스너의 토큰 검증 규칙에는 영향을 미치지 않습니다.

   6

   유효한 발급자 URI입니다. 이 발행자에서 발행한 액세스 토큰만 허용됩니다. (알림이 필요합니다.)

   7

   모든 브로커에 대해 동일한 Kafka 브로커의 구성된 클라이언트 ID입니다. kafka-broker로 권한 부여 서버에 등록된 클라이언트 입니다. 인트로스펙션 엔드 포인트가 토큰 검증에 사용되거나 KeycloakRBACAuthorizer 가 사용되는 경우에 필요합니다.

   8

   모든 브로커에 대해 동일한 Kafka 브로커에 대해 구성된 시크릿입니다. 브로커가 권한 부여 서버에 인증해야 하는 경우 클라이언트 시크릿, 액세스 토큰 또는 새로 고침 토큰을 지정해야 합니다.

   9

   (선택 사항) 권한 부여 서버에 연결할 때 연결 제한 시간(초)입니다. 기본값은 60입니다.

   10

   (선택 사항) 권한 부여 서버에 연결할 때 읽기 제한 시간(초)입니다. 기본값은 60입니다.

   11

   권한 부여 서버에 실패한 HTTP 요청을 재시도하는 최대 횟수입니다. 기본값은 0이며, 즉 재시도를 수행하지 않습니다. 이 옵션을 효과적으로 사용하려면 oauth.connect.timeout.seconds 및 oauth.read.timeout.seconds 옵션에 대한 시간 초과 시간을 줄이는 것이 좋습니다. 그러나 재시도하면 현재 작업자 스레드를 다른 요청에서 사용할 수 없게 할 수 있으며 요청이 너무 많으면 Kafka 브로커가 응답하지 않을 수 있습니다.

   12

   권한 부여 서버에 실패한 HTTP 요청의 다른 재시도를 시도하기 전에 대기하는 시간입니다. 기본적으로 이 시간은 0으로 설정되어 일시 중지가 적용되지 않습니다. 실패한 요청으로 인해 많은 문제가 요청별 네트워크 결함 또는 프록시 문제로 신속하게 해결될 수 있기 때문입니다. 그러나 권한 부여 서버가 과부하 상태에 있거나 트래픽이 높은 경우 이 옵션을 100ms 이상으로 설정하여 서버의 부하를 줄이고 재시도 횟수 가능성을 높일 수 있습니다.

   13

   JWT 토큰 또는 인트로스펙션 끝점 응답에서 그룹 정보를 추출하는 데 사용되는 JsonPath 쿼리입니다. 기본적으로 설정되지 않습니다. 사용자 정의 작성자가 사용자 그룹을 기반으로 권한 결정을 내리는 데 사용할 수 있습니다.

   14

   구분된 단일 문자열로 반환되는 경우 그룹 정보를 구문 분석하는 데 사용되는 구분 기호입니다. 기본값은 ','(comma)입니다.

5. OAuth 2.0 인증 및 사용 권한 부여 서버 유형에 따라 추가 구성 설정을 추가합니다.

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     # ...
     oauth.check.issuer=false \ 1
     oauth.fallback.username.claim="CLIENT-ID" \ 2
     oauth.fallback.username.prefix="CLIENT-ACCOUNT" \ 3
     oauth.valid.token.type="bearer" \ 4
     oauth.userinfo.endpoint.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME/protocol/openid-connect/userinfo" ; 5

   1

   권한 부여 서버에서 iss 클레임을 제공하지 않으면 발급자 검사를 수행할 수 없습니다. 이 경우 oauth.check.issuer 를 false 로 설정하고 oauth.valid.issuer.uri 를 지정하지 마십시오. 기본값은 true 입니다.

   2

   권한 부여 서버는 일반 사용자와 클라이언트를 모두 식별하는 단일 속성을 제공하지 않을 수 있습니다. 클라이언트가 자체 이름으로 인증되면 서버에서 클라이언트 ID 를 제공할 수 있습니다. 사용자 이름과 암호를 사용하여 인증할 때 새로 고침 토큰 또는 액세스 토큰을 얻기 위해 서버는 클라이언트 ID 외에도 username 속성을 제공할 수 있습니다. 기본 사용자 ID 속성을 사용할 수 없는 경우 사용할 사용자 이름 클레임(attribute)을 지정하려면 이 대체 옵션을 사용합니다.

   3

   oauth. fallbackback.username.claim 이 적용되는 경우 사용자 이름 클레임 값과 대체 사용자 이름 클레임의 값 간에 이름 충돌을 방지해야 할 수도 있습니다. 생산자라는 클라이언트뿐만 아니라 생산자 라는 일반 사용자도 존재하는 상황을 고려하십시오. 이 속성을 사용하여 클라이언트의 사용자 ID에 접두사를 추가할 수 있습니다.You can use this property to add a prefix to the user ID of the client.

   4

   (사용 중인 권한 부여 서버에 따라 oauth.introspection.endpoint.uri를 사용하는 경우에만 적용 가능) 인트로스펙션 끝점이 토큰 유형 특성을 반환하거나 반환하지 않거나 다른 값을 포함할 수 있습니다. 인트로스펙션 끝점의 응답에 포함해야 하는 유효한 토큰 유형 값을 지정할 수 있습니다.

   5

   ( oauth.introspection.endpoint.uri사용 시에만 적용 가능) 권한 부여 서버가 인트로스펙션 끝점 응답에서 식별 가능한 정보를 제공하지 않도록 구성하거나 구현할 수 있습니다. 사용자 ID를 가져오려면 userinfo 끝점의 URI를 폴백으로 구성할 수 있습니다. oauth.knativeback.username.claim,oauth.entryback.username.claim, oauth. fallbackback.username.prefix 설정이 userinfo 끝점의 응답에 적용됩니다.

절차

1. OAuth 2.0 지원이 포함된 클라이언트 라이브러리를 Kafka 클라이언트의 pom.xml 파일에 추가합니다.

   <dependency>
    <groupId>io.strimzi</groupId>
    <artifactId>kafka-oauth-client</artifactId>
    <version>0.12.0.redhat-00006</version>
   </dependency>

2. 속성 파일에 다음 구성을 지정하여 클라이언트 속성을 구성합니다.

   • 보안 프로토콜
   • SASL 메커니즘

   • 사용 중인 방법에 따른 module 및 인증 속성

     예를 들어 client.properties 파일에 다음을 추가할 수 있습니다.

     클라이언트 자격 증명 메커니즘 속성

     security.protocol=SASL_SSL 1
     sasl.mechanism=OAUTHBEARER 2
     ssl.truststore.location=/tmp/truststore.p12 3
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \ 4
       oauth.client.id="<client_id>" \ 5
       oauth.client.secret="<client_secret>" \ 6
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \ 7
       oauth.ssl.truststore.password="$STOREPASS" \ 8
       oauth.ssl.truststore.type="PKCS12" \ 9
       oauth.scope="<scope>" \ 10
       oauth.audience="<audience>" ; 11

     1

     TLS 암호화 연결을 위한 SASL_SSL 보안 프로토콜. 로컬 개발에만 암호화되지 않은 연결에서는 SASL_PLAINTEXT 를 사용하십시오.

     2

     SASL 메커니즘을 OAUTHB learnER 또는 PLAIN 으로 지정합니다.

     3

     Kafka 클러스터에 대한 보안 액세스를 위한 신뢰 저장소 구성입니다.

     4

     권한 부여 서버 토큰 끝점의 URI입니다.

     5

     클라이언트 ID: 권한 부여 서버에서 클라이언트를 생성할 때 사용되는 이름입니다.

     6

     권한 부여 서버에서 클라이언트를 생성할 때 생성된 클라이언트 시크릿입니다.

     7

     위치에는 권한 부여 서버의 공개 키 인증서(truststore.p12)가 포함됩니다.

     8

     신뢰 저장소에 액세스하기 위한 암호입니다.

     9

     truststore 유형입니다.

     10

     (선택 사항) 토큰 끝점에서 토큰을 요청하는 범위입니다. 권한 부여 서버에 범위를 지정하기 위해 클라이언트가 필요할 수 있습니다.

     11

     (선택 사항) 토큰 끝점에서 토큰을 요청하는 대상입니다. 권한 부여 서버에 대상을 지정하기 위해 클라이언트가 필요할 수 있습니다.

     password grant mechanism properties

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.client.id="<client_id>" \ 1
       oauth.client.secret="<client_secret>" \ 2
       oauth.password.grant.username="<username>" \ 3
       oauth.password.grant.password="<password>" \ 4
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" \
       oauth.scope="<scope>" \
       oauth.audience="<audience>" ;

     1

     클라이언트 ID: 권한 부여 서버에서 클라이언트를 생성할 때 사용되는 이름입니다.

     2

     (선택 사항) 권한 부여 서버에서 클라이언트를 생성할 때 생성된 클라이언트 시크릿입니다.

     3

     암호 부여 인증에 대한 사용자 이름입니다. OAuth 암호 부여 구성(사용자 이름 및 암호)은 OAuth 2.0 암호 부여 방법을 사용합니다. 암호 부여를 사용하려면 제한된 권한이 있는 권한 부여 서버에서 클라이언트에 대한 사용자 계정을 만듭니다. 계정이 서비스 계정처럼 작동해야 합니다. 인증에 사용자 계정이 필요하지만 새로 고침 토큰을 사용하는 것이 필요한 환경에서 사용합니다.

     4

     암호 부여 인증의 암호입니다.

     참고

     SASL PLAIN은 OAuth 2.0 암호 부여 방법을 사용하여 사용자 이름과 암호(암호 부여) 전달을 지원하지 않습니다.

     액세스 토큰 속성

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.access.token="<access_token>" ; 1
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" \

     1

     Kafka 클라이언트의 장기 액세스 토큰입니다.

     토큰 속성 새로 고침

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.client.id="<client_id>" \ 1
       oauth.client.secret="<client_secret>" \ 2
       oauth.refresh.token="<refresh_token>" ; 3
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" \

     1

     클라이언트 ID: 권한 부여 서버에서 클라이언트를 생성할 때 사용되는 이름입니다.

     2

     (선택 사항) 권한 부여 서버에서 클라이언트를 생성할 때 생성된 클라이언트 시크릿입니다.

     3

     Kafka 클라이언트의 장기 새로 고침 토큰입니다.

3. OAUTH 2.0 인증에 대한 클라이언트 속성을 Java 클라이언트 코드로 입력합니다.

   클라이언트 속성 입력 표시 예

   Properties props = new Properties();
   try (FileReader reader = new FileReader("client.properties", StandardCharsets.UTF_8)) {
     props.load(reader);
   }

4. Kafka 클라이언트가 Kafka 브로커에 액세스할 수 있는지 확인합니다.