RHEL에서 Apache Kafka에 Streams 사용

프로세스

1. 클러스터의 모든 Kafka 브로커에 대한 TLS 인증서를 생성합니다. 인증서에는 일반 이름 또는 주체 대체 이름에 광고 및 부트스트랩 주소가 있어야 합니다.

2. 다음을 위해 모든 클러스터 노드에서 Kafka 구성 속성 파일을 편집합니다.

   • listener.security.protocol.map 필드를 변경하여 TLS 암호화를 사용하려는 리스너의 SSL 프로토콜을 지정합니다.
   • 브로커 인증서와 함께 JKS 키 저장소의 경로로 ssl.keystore.location 옵션을 설정합니다.

   • ssl.keystore.password 옵션을 키 저장소를 보호하는 데 사용한 암호로 설정합니다.

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

     listeners=UNENCRYPTED://:9092,ENCRYPTED://:9093,REPLICATION://:9094
     listener.security.protocol.map=UNENCRYPTED:PLAINTEXT,ENCRYPTED:SSL,REPLICATION:PLAINTEXT
     ssl.keystore.location=/path/to/keystore/server-1.jks
     ssl.keystore.password=123456

     Copy to Clipboard Toggle word wrap
3. Kafka 브로커 시작

프로세스

1. 사용자 인증서에 서명하는 데 사용되는 CA(Certification Authority)의 공개 키가 포함된 JKS(Java 키 저장소) 신뢰 저장소를 준비합니다.

2. 다음과 같이 모든 클러스터 노드에서 Kafka 구성 속성 파일을 편집합니다.

   • ssl.truststore.location 속성을 사용하여 JKS 신뢰 저장소의 경로를 지정합니다.
   • 신뢰 저장소가 암호로 보호되는 경우 ssl.truststore.password 속성을 사용하여 암호를 설정합니다.

   • ssl.client.auth 속성을 required 로 설정합니다.

     TLS 클라이언트 인증 구성

     ssl.truststore.location=/path/to/truststore.jks
     ssl.truststore.password=123456
     ssl.client.auth=required

     Copy to Clipboard Toggle word wrap

3. Kafka 브로커를 시작합니다.

프로세스

1. /opt/kafka/config/jaas.conf JAAS 구성 파일을 편집하거나 생성하여 PlainLoginModule 을 활성화하고 허용된 사용자 이름과 암호를 지정합니다.

   이 파일이 모든 Kafka 브로커에서 동일한지 확인합니다.

   JAAS 구성

   KafkaServer {
       org.apache.kafka.common.security.plain.PlainLoginModule required
       user_admin="123456"
       user_user1="123456"
       user_user2="123456";
   };

   Copy to Clipboard Toggle word wrap

2. 다음과 같이 모든 클러스터 노드에서 Kafka 구성 속성 파일을 편집합니다.

   • listener.security.protocol.map 속성을 사용하여 특정 리스너에서 SASL PLAIN 인증을 활성화합니다. SASL_PLAINTEXT 또는 SASL_SSL 을 지정합니다.

   • sasl.enabled.mechanisms 속성을 PLAIN 으로 설정합니다.

     SASL 일반 구성

     listeners=INSECURE://:9092,AUTHENTICATED://:9093,REPLICATION://:9094
     listener.security.protocol.map=INSECURE:PLAINTEXT,AUTHENTICATED:SASL_PLAINTEXT,REPLICATION:PLAINTEXT
     sasl.enabled.mechanisms=PLAIN

     Copy to Clipboard Toggle word wrap

3. (re) KAFKA_OPTS 환경 변수를 사용하여 Kafka 브로커를 시작하여 JAAS 구성을 Kafka 브로커에 전달합니다.

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

   Copy to Clipboard Toggle word wrap

프로세스

1. /opt/kafka/config/jaas.conf JAAS 구성 파일을 편집하거나 생성하여 ScramLoginModule 을 활성화합니다.

   이 파일이 모든 Kafka 브로커에서 동일한지 확인합니다.

   JAAS 구성

   KafkaServer {
       org.apache.kafka.common.security.scram.ScramLoginModule required;
   };

   Copy to Clipboard Toggle word wrap

2. 다음과 같이 모든 클러스터 노드에서 Kafka 구성 속성 파일을 편집합니다.

   • listener.security.protocol.map 속성을 사용하여 특정 리스너에서 SASL SCRAM 인증을 활성화합니다. SASL_PLAINTEXT 또는 SASL_SSL 을 지정합니다.

   • sasl.enabled.mechanisms 옵션을 SCRAM-SHA-256 또는 SCRAM-SHA-512 로 설정합니다.

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

     listeners=INSECURE://:9092,AUTHENTICATED://:9093,REPLICATION://:9094
     listener.security.protocol.map=INSECURE:PLAINTEXT,AUTHENTICATED:SASL_PLAINTEXT,REPLICATION:PLAINTEXT
     sasl.enabled.mechanisms=SCRAM-SHA-512

     Copy to Clipboard Toggle word wrap

3. (re) KAFKA_OPTS 환경 변수를 사용하여 Kafka 브로커를 시작하여 JAAS 구성을 Kafka 브로커에 전달합니다.

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

   Copy to Clipboard Toggle word wrap

프로세스

1. sasl.mechanism.inter.broker.protocol 속성을 사용하여 Kafka 구성에서 inter-broker SASL 메커니즘을 지정합니다.

   broker SASL 메커니즘

   sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

   Copy to Clipboard Toggle word wrap

2. (선택 사항) SCRAM 메커니즘을 사용하는 경우 SCRAM 사용자를 추가하여 Kafka 클러스터에서 SCRAM 자격 증명을 등록합니다.

   이렇게 하면 Kafka 클러스터가 실행되기 전에 인증에 대한 인증 정보가 부트스트랩을 위해 업데이트됩니다.

3. username 및 password 필드를 사용하여 KafkaServer JAAS 컨텍스트에서broker 간 통신의 사용자 이름과 암호를 지정합니다.

   broker JAAS 컨텍스트

   KafkaServer {
       org.apache.kafka.common.security.plain.ScramLoginModule required
       username="admin"
       password="123456"
       # ...
   };

   Copy to Clipboard Toggle word wrap

1. 서비스 주체
2. Kerberos 로그인을 사용하는 Kafka 브로커
3. Kerberos 로그인을 사용하는 Zookeeper
4. Kerberos 인증을 사용하여 Kafka에 액세스할 수 있는 생산자 및 소비자 클라이언트

1. Kerberos KDC를 통해 주 키를 저장하는 keytab과 서비스 주체를 생성합니다.

   Kerberos 주체의 도메인 이름이 대문자인지 확인합니다.

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

   • zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM
   • kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM
   • producer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM

   • consumer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM

     Zoo Cryostat 서비스 주체는 Kafka config/server.properties 파일의 zookeeper.connect 구성과 동일한 호스트 이름을 사용해야 합니다.

     zookeeper.connect=node1.example.redhat.com:2181

     Copy to Clipboard Toggle word wrap

     호스트 이름이 동일하지 않으면 localhost 가 사용되고 인증이 실패합니다.

2. 호스트에 디렉터리를 만들고 keytab 파일을 추가합니다.

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

   /opt/kafka/krb5/zookeeper-node1.keytab
   /opt/kafka/krb5/kafka-node1.keytab
   /opt/kafka/krb5/kafka-producer1.keytab
   /opt/kafka/krb5/kafka-consumer1.keytab

   Copy to Clipboard Toggle word wrap

3. kafka 사용자가 디렉터리에 액세스할 수 있는지 확인합니다.

   chown kafka:kafka -R /opt/kafka/krb5

   Copy to Clipboard Toggle word wrap

1. opt/kafka/config/jaas.conf 파일을 생성하거나 수정하여 Zoo Cryostat 클라이언트 및 서버 작업을 지원합니다.

   Client {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true 

   1

   
       storeKey=true 

   2

   
       useTicketCache=false 

   3

   
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab" 

   4

   
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM"; 

   5

   
   };
   
   Server {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       useTicketCache=false
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   
   QuorumServer {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   
   QuorumLearner {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };

   Copy to Clipboard Toggle word wrap

   1

   키탭에서 기본 키를 가져오려면 true 로 설정합니다.

   2

   기본 키를 저장하려면 true 로 설정합니다.

   3

   티켓 캐시에서 티켓 부여 티켓(TGT)을 받으려면 true 로 설정합니다.

   4

   keyTab 속성은 Kerberos KDC에서 복사한 키탭 파일의 위치를 가리킵니다. 위치와 파일은 kafka 사용자가 읽을 수 있어야 합니다.

   5

   principal 속성은 SERVICE-NAME/FULLY-QUALIFIED-HOST-NAME@DOMAIN-NAME 형식을 따르는 KDC 호스트에서 생성된 정규화된 주체 이름과 일치하도록 구성됩니다.

2. 업데이트된 JAAS 구성을 사용하도록 opt/kafka/config/zookeeper.properties 를 편집합니다.

   # ...
   
   requireClientAuthScheme=sasl
   jaasLoginRenew=3600000 

   1

   
   kerberos.removeHostFromPrincipal=false 

   2

   
   kerberos.removeRealmFromPrincipal=false 

   3

   
   quorum.auth.enableSasl=true 

   4

   
   quorum.auth.learnerRequireSasl=true 

   5

   
   quorum.auth.serverRequireSasl=true
   quorum.auth.learner.loginContext=QuorumLearner 

   6

   
   quorum.auth.server.loginContext=QuorumServer
   quorum.auth.kerberos.servicePrincipal=zookeeper/_HOST 

   7

   
   quorum.cnxn.threads.size=20

   Copy to Clipboard Toggle word wrap

   1

   로그인 갱신 빈도를 밀리초 단위로 제어하며 티켓 갱신 간격에 맞게 조정할 수 있습니다. 기본값은 1시간입니다.

   2

   호스트 이름이 로그인 주체 이름의 일부로 사용되는지 여부를 지정합니다. 클러스터의 모든 노드에 단일 키탭을 사용하는 경우 true 로 설정됩니다. 그러나 문제 해결을 위해 각 브로커 호스트에 대해 별도의 keytab과 정규화된 주체를 생성하는 것이 좋습니다.

   3

   Kerberos 협상의 주체 이름에서 영역 이름이 제거되는지 여부를 제어합니다. 이 설정을 false 로 설정하는 것이 좋습니다.

   4

   Zoo Cryostat 서버 및 클라이언트에 대한 SASL 인증 메커니즘을 활성화합니다.

   5

   RequireSasl 속성은 마스터 선택과 같은 쿼럼 이벤트에 SASL 인증이 필요한지 여부를 제어합니다.

   6

   loginContext 속성은 지정된 구성 요소의 인증 구성에 사용되는 JAAS 구성에서 로그인 컨텍스트의 이름을 식별합니다. loginContext 이름은 opt/kafka/config/jaas.conf 파일의 관련 섹션 이름에 해당합니다.

   7

   식별에 사용되는 보안 주체 이름을 구성하는 데 사용할 이름 지정 규칙을 제어합니다.Controls the naming convention to be used to form the principal name used for identification. 자리 표시자 _HOST 는 런타임 시 server.1 속성에 정의된 호스트 이름으로 자동으로 확인됩니다.

3. JVM 매개변수로 Zoo Cryostat를 시작하여 Kerberos 로그인 구성을 지정합니다.

   su - kafka
   export EXTRA_ARGS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

   Copy to Clipboard Toggle word wrap

   기본 서비스 이름( Zookeeper )을 사용하지 않는 경우 -Dzookeeper.sasl.client.username=NAME 매개변수를 사용하여 이름을 추가합니다.

   참고

   /etc/krb5.conf 위치를 사용하는 경우 Zoo Cryostat, Kafka 또는 Kafka 생산자 및 소비자를 시작할 때 -Djava.security.krb5.conf=/etc/krb5.conf 를 지정할 필요가 없습니다.

1. 다음 요소를 사용하여 opt/kafka/config/jaas.conf 파일을 수정합니다.

   KafkaServer {
       com.sun.security.auth.module.Krb5LoginModule required
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/kafka-node1.keytab"
       principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   KafkaClient {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       useTicketCache=false
       keyTab="/opt/kafka/krb5/kafka-node1.keytab"
       principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };

   Copy to Clipboard Toggle word wrap

2. 리스너가 SASL/GSSAPI 로그인을 사용하도록 config/server.properties 파일에서 리스너 구성을 수정하여 Kafka 클러스터에서 각 브로커를 구성합니다.

   SASL 프로토콜을 리스너의 보안 프로토콜 맵에 추가하고 원하지 않는 프로토콜을 제거합니다.

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

   # ...
   broker.id=0
   # ...
   listeners=SECURE://:9092,REPLICATION://:9094 

   1

   
   inter.broker.listener.name=REPLICATION
   # ...
   listener.security.protocol.map=SECURE:SASL_PLAINTEXT,REPLICATION:SASL_PLAINTEXT 

   2

   
   # ..
   sasl.enabled.mechanisms=GSSAPI 

   3

   
   sasl.mechanism.inter.broker.protocol=GSSAPI 

   4

   
   sasl.kerberos.service.name=kafka 

   5

   
   ...

   Copy to Clipboard Toggle word wrap

   1

   두 개의 리스너: 클라이언트와의 범용 통신을 위한 보안 리스너( communication을 위해 TLS 지원) 및 브루커 간 통신을 위한 복제 리스너가 구성됩니다.

   2

   TLS 지원 리스너의 경우 프로토콜 이름은 SASL_PLAINTEXT입니다. TLS 지원 커넥터의 경우 프로토콜 이름은 SASL_PLAINTEXT입니다. SSL이 필요하지 않은 경우 ssl.* 속성을 제거할 수 있습니다.

   3

   Kerberos 인증을 위한 SASL 메커니즘은 GSSAPI 입니다.

   4

   브랜드 간 통신을 위한 Kerberos 인증.

   5

   인증 요청에 사용되는 서비스의 이름은 동일한 Kerberos 구성을 사용하는 다른 서비스와 구별하기 위해 지정됩니다.

3. JVM 매개변수를 사용하여 Kafka 브로커를 시작하여 Kerberos 로그인 구성을 지정합니다.

   su - kafka
   export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

   Copy to Clipboard Toggle word wrap

   브로커 및 Zoo Cryostat 클러스터가 이전에 구성되어 Kerberos 기반 인증 시스템으로 작업하는 경우 Zoo Cryostat 및 브로커 클러스터를 시작하고 로그에서 구성 오류를 확인할 수 있습니다.

   브로커 및 Zookeeper 인스턴스를 시작한 후 이제 Kerberos 인증을 위해 클러스터가 구성됩니다.

1. Kerberos 구성을 생산자 또는 소비자 구성 파일에 추가합니다.

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

   /opt/kafka/config/producer.properties

   # ...
   sasl.mechanism=GSSAPI 

   1

   
   security.protocol=SASL_PLAINTEXT 

   2

   
   sasl.kerberos.service.name=kafka 

   3

   
   sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \ 

   4

   
       useKeyTab=true  \
       useTicketCache=false \
       storeKey=true  \
       keyTab="/opt/kafka/krb5/producer1.keytab" \
       principal="producer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   # ...

   Copy to Clipboard Toggle word wrap

   1

   Kerberos(GSSAPI) 인증을 위한 구성

   2

   Kerberos는 SASL 일반 텍스트(사용자 이름/암호) 보안 프로토콜을 사용합니다.

   3

   Kerberos KDC에 구성된 Kafka의 서비스 주체(사용자)입니다.

   4

   jaas.conf 에 정의된 것과 동일한 속성을 사용하는 JAAS 구성

   /opt/kafka/config/consumer.properties

   # ...
   sasl.mechanism=GSSAPI
   security.protocol=SASL_PLAINTEXT
   sasl.kerberos.service.name=kafka
   sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
       useKeyTab=true  \
       useTicketCache=false \
       storeKey=true  \
       keyTab="/opt/kafka/krb5/consumer1.keytab" \
       principal="consumer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   # ...

   Copy to Clipboard Toggle word wrap

2. 클라이언트를 실행하여 Kafka 브로커에서 메시지를 보내고 받을 수 있는지 확인합니다.

   생산자 클라이언트:

   export KAFKA_HEAP_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Dsun.security.krb5.debug=true"; /opt/kafka/bin/kafka-console-producer.sh --producer.config /opt/kafka/config/producer.properties  --topic topic1 --bootstrap-server node1.example.redhat.com:9094

   Copy to Clipboard Toggle word wrap

   소비자 클라이언트:

   export KAFKA_HEAP_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Dsun.security.krb5.debug=true"; /opt/kafka/bin/kafka-console-consumer.sh --consumer.config /opt/kafka/config/consumer.properties  --topic topic1 --bootstrap-server node1.example.redhat.com:9094

   Copy to Clipboard Toggle word wrap

SASL>-<GEST-MD5 인증 활성화

1. 모든 Kafka 브로커 노드에서 /opt/kafka/config/jaas.conf JAAS 구성 파일을 생성하거나 편집하고 다음 컨텍스트를 추가합니다.

   Client {
       org.apache.kafka.common.security.plain.PlainLoginModule required
       username="<Username>"
       password="<Password>";
   };

   Copy to Clipboard Toggle word wrap

   사용자 이름과 암호는 Zoo Cryostat에 구성된 것과 동일해야 합니다.

   다음 예제에서는 클라이언트 컨텍스트를 보여줍니다.

   Client {
       org.apache.kafka.common.security.plain.PlainLoginModule required
       username="kafka"
       password="123456";
   };

   Copy to Clipboard Toggle word wrap

2. 모든 Kafka 브로커 노드를 하나씩 다시 시작합니다. JAAS 구성을 Kafka 브로커에 전달하려면 KAFKA_OPTS 환경 변수를 사용합니다.

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

   Copy to Clipboard Toggle word wrap

   멀티 노드 클러스터에서 브로커를 다시 시작하는 방법에 대한 자세한 내용은 4.3절. “Kafka 브로커의 정상 롤링 재시작 수행” 을 참조하십시오.

프로세스

1. Kafka 구성 속성 파일을 편집하여 모든 클러스터 노드에서 zookeeper.set.acl 필드를 true 로 설정합니다.

   zookeeper.set.acl=true

   Copy to Clipboard Toggle word wrap
2. Kafka 브로커를 시작합니다.

Zoo Cryostat ACL 활성화

1. Kafka 구성 속성 파일을 편집하여 모든 클러스터 노드에서 zookeeper.set.acl 필드를 true 로 설정합니다.

   zookeeper.set.acl=true

   Copy to Clipboard Toggle word wrap

2. 모든 Kafka 브로커를 하나씩 다시 시작합니다.

   멀티 노드 클러스터에서 브로커를 다시 시작하는 방법에 대한 자세한 내용은 4.3절. “Kafka 브로커의 정상 롤링 재시작 수행” 을 참조하십시오.

3. Zookeeper -security-migration.sh 도구를 사용하여 기존의 모든 Zoo Cryostat znodes 에서 ACL을 설정합니다.

   su - kafka
   cd /opt/kafka
   KAFKA_OPTS="-Djava.security.auth.login.config=./config/jaas.conf"; ./bin/zookeeper-security-migration.sh --zookeeper.acl=secure --zookeeper.connect=<ZooKeeperURL>
   exit

   Copy to Clipboard Toggle word wrap

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

   su - kafka
   cd /opt/kafka
   KAFKA_OPTS="-Djava.security.auth.login.config=./config/jaas.conf"; ./bin/zookeeper-security-migration.sh --zookeeper.acl=secure --zookeeper.connect=zoo1.my-domain.com:2181
   exit

   Copy to Clipboard Toggle word wrap

1. Apache Kafka 클러스터 및 Kafka 클라이언트의 Streams에 대한 OAuth 2.0 권한 부여 서버 구성
2. OAuth 2.0을 사용하도록 구성된 Kafka 브로커 리스너를 사용하여 Kafka 클러스터를 배포하거나 업데이트합니다.
3. OAuth 2.0을 사용하도록 Java 기반 Kafka 클라이언트를 업데이트