6.4. Kafka 구성

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

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

1. Kafka 클라이언트는 SASL OAUTHBEARER 메커니즘을 사용하여 Kafka 브로커로 인증하여 장기 액세스 토큰을 전달합니다.
2. Kafka 브로커는 자체 클라이언트 ID와 시크릿을 사용하여 권한 부여 서버에서 토큰 인트로스펙션 엔드포인트를 호출하여 액세스 토큰을 검증합니다.
3. 토큰이 유효한 경우 Kafka 클라이언트 세션이 설정됩니다.

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

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

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

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

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

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

   1. 토큰 끝점이 구성된 경우 브로커에 password 매개변수에 클라이언트 시크릿이 아닌 액세스 토큰이 포함되어 있음을 알 수 있도록 암호 앞에 $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. 설정 탭에서 다음을 설정합니다.

   • 액세스 유형 ( Confidential)
   • 이 클라이언트에 대한 웹 로그인을 비활성화하려면 표준 흐름을 OFF 로 활성화
   • 이 클라이언트가 고유한 이름으로 인증할 수 있도록 활성화됨
7. 계속하기 전에 저장을 클릭합니다.
8. Credentials 탭에서 AMQ Streams Kafka 클러스터 구성에서 사용할 시크릿을 기록합니다.

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

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

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

프로세스

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

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

   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://<oauth_server_address>/auth/realms/<realm_name>" \
     oauth.jwks.endpoint.uri="https://<oauth_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://<oauth_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://<oauth_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://<oauth_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
     oauth.include.accept.header="false" ; 15

   1

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

   2

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

   3

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

   4

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

   5

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

   6

   유효한 발행자 URI입니다. 이 발행자가 발행한 액세스 토큰만 허용됩니다. (필수 사항)

   7

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

   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)입니다.

   15

   (선택 사항) oauth.include.accept.header 를 false 로 설정하여 요청에서 Accept 헤더를 제거합니다. 헤더를 포함하면 권한 부여 서버와 통신할 때 문제가 발생하는 경우 이 설정을 사용할 수 있습니다.

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://<oauth_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)을 지정하려면 이 대체 옵션을 사용합니다. 필요한 경우 "['client.info'].['client.id']" 와 같은 JsonPath 표현식을 사용하여 토큰 내의 중첩된 JSON 속성에서 대체 사용자 이름을 검색할 수 있습니다.

   3

   oauth.fallback.username.claim 이 적용되는 경우 사용자 이름 클레임 값과 대체 사용자 이름 클레임의 이름 충돌을 방지할 수도 있습니다. producer 라는 클라이언트가 존재하지만 producer 라는 일반 사용자도 존재하는 상황을 고려하십시오. 이 속성을 사용하여 클라이언트의 사용자 ID에 접두사를 추가할 수 있습니다.

   4

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

   5

   ( oauth.introspection.endpoint.uri) 사용 시에만 적용할 수 있습니다. 인증 서버는 인트로스펙션 엔드포인트 응답에서 식별 가능한 정보를 제공하지 않기 위해 인증 서버를 구성하거나 구현할 수 있습니다. 사용자 ID를 얻으려면 userinfo 끝점의 URI를 폴백으로 구성할 수 있습니다. oauth.fallback.username.claim,oauth.fallback.username.claim, oauth.fallback.username.prefix 설정은 userinfo 엔드포인트 응답에 적용됩니다.

프로세스

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

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

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

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

   • 사용 방법에 따른 JAAS 모듈 및 인증 속성

     예를 들어 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
     sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

     1

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

     2

     OAUTHBEARER 또는 PLAIN 으로 지정된 SASL 메커니즘 .

     3

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

     4

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

     5

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

     6

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

     7

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

     8

     truststore에 액세스하기 위한 암호입니다.

     9

     truststore 유형입니다.

     10

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

     11

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

     암호는 메커니즘 속성 부여

     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>" ;
     sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

     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" ;
     sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

     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" ;
     sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

     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 브로커에 액세스할 수 있는지 확인합니다.