24.2. 파티션을 다시 할당하기 위한 재할당 JSON 파일 생성

프로세스

1. Kafka 클러스터의 <cluster _name> -cluster-ca-cert 시크릿에서 클러스터 CA 인증서 및 암호를 추출합니다.

   oc get secret <cluster_name>-cluster-ca-cert -o jsonpath='{.data.ca\.p12}' | base64 -d > ca.p12

   oc get secret <cluster_name>-cluster-ca-cert -o jsonpath='{.data.ca\.password}' | base64 -d > ca.password

   & lt;cluster_name >을 Kafka 클러스터 이름으로 바꿉니다. Kafka 리소스를 사용하여 Kafka 를 배포할 때 클러스터 CA 인증서가 있는 시크릿은 Kafka 클러스터 이름(<cluster_name> -cluster-ca-cert)을 사용하여 생성됩니다. 예를 들면 my-cluster-cluster-ca-cert 입니다.

2. Kafka 이미지를 사용하여 실행 중인 Kafka 브로커에 연결합니다.

   oc run --restart=Never --image=registry.redhat.io/amq-streams/kafka-40-rhel9:3.0.1 <interactive_pod_name> -- /bin/sh -c "sleep 3600"

   & lt;interactive_pod_name& gt;을 Pod 이름으로 바꿉니다.

3. 클러스터 CA 인증서를 대화형 Pod 컨테이너에 복사합니다.

   oc cp ca.p12 <interactive_pod_name>:/tmp

4. Kafka 브로커에 액세스할 수 있는 권한이 있는 Kafka 사용자의 시크릿에서 사용자 CA 인증서 및 암호를 추출합니다.

   oc get secret <kafka_user> -o jsonpath='{.data.user\.p12}' | base64 -d > user.p12

   oc get secret <kafka_user> -o jsonpath='{.data.user\.password}' | base64 -d > user.password

   & lt;kafka_user >를 Kafka 사용자의 이름으로 바꿉니다. KafkaUser 리소스를 사용하여 Kafka 사용자를 생성하면 사용자 CA 인증서가 있는 시크릿이 Kafka 사용자 이름으로 생성됩니다. 예를 들면 my-user 입니다.

5. 사용자 CA 인증서를 대화형 Pod 컨테이너에 복사합니다.

   oc cp user.p12 <interactive_pod_name>:/tmp

   CA 인증서를 사용하면 대화형 Pod 컨테이너가 TLS를 사용하여 Kafka 브로커에 연결할 수 있습니다.

6. config.properties 파일을 생성하여 Kafka 클러스터에 대한 연결을 인증하는 데 사용되는 truststore 및 키 저장소를 지정합니다.

   이전 단계에서 추출한 인증서 및 암호를 사용합니다.

   bootstrap.servers=<kafka_cluster_name>-kafka-bootstrap:9093 

   1

   
   security.protocol=SSL 

   2

   
   ssl.truststore.location=/tmp/ca.p12 

   3

   
   ssl.truststore.password=<truststore_password> 

   4

   
   ssl.keystore.location=/tmp/user.p12 

   5

   
   ssl.keystore.password=<keystore_password> 

   6

   1

   Kafka 클러스터에 연결할 부트스트랩 서버 주소입니다. < kafka_cluster_name>을 교체하려면 자체 Kafka 클러스터 이름을 사용합니다.

   2

   암호화에 TLS를 사용할 때 보안 프로토콜 옵션입니다.

   3

   truststore 위치에는 Kafka 클러스터의 공개 키 인증서(ca.p12)가 포함되어 있습니다.

   4

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

   5

   키 저장소 위치에는 Kafka 사용자의 공개 키 인증서(user.p12)가 포함되어 있습니다.

   6

   키 저장소에 액세스하기 위한 암호(user.password)입니다.

7. config.properties 파일을 대화형 Pod 컨테이너에 복사합니다.

   oc cp config.properties <interactive_pod_name>:/tmp/config.properties

8. 이동할 주제를 지정하는 topics.json 이라는 JSON 파일을 준비합니다.

   주제 이름을 쉼표로 구분된 목록으로 지정합니다.

   my-topic의 모든 파티션을 다시 할당하는 JSON 파일의 예

   {
     "version": 1,
     "topics": [
       { "topic": "my-topic"}
     ]
   }

   이 파일을 사용하여 주제의 복제 요소를 변경할 수도 있습니다.

9. topics.json 파일을 대화형 Pod 컨테이너에 복사합니다.

   oc cp topics.json <interactive_pod_name>:/tmp/topics.json

10. 대화형 pod 컨테이너에서 쉘 프로세스를 시작합니다.

    oc exec -n <namespace> -ti <interactive_pod_name> /bin/bash

    & lt;namespace& gt;를 Pod가 실행 중인 OpenShift 네임스페이스로 바꿉니다.

11. kafka-reassign-partitions.sh 명령을 사용하여 재할당 JSON을 생성합니다.

    my-topic 의 파티션을 지정된 브로커로 이동하는 명령의 예

    bin/kafka-reassign-partitions.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 \
      --command-config /tmp/config.properties \
      --topics-to-move-json-file /tmp/topics.json \
      --broker-list 0,1,2,3,4 \
      --generate