4단계: Amazon MSK 클러스터에서 주제 생성

https://console.aws.amazon.com/msk/에서 Amazon MSK 콘솔을 엽니다.

탐색 모음에서 MSK 클러스터를 생성한 리전을 선택합니다.

MSK 클러스터를 선택합니다.

클러스터에서 사용되는 Apache Kafka 버전을 확인합니다.

자습서에서 Amazon MSK 버전 번호의 인스턴스를 3단계에서 얻은 버전으로 변경합니다.

클라이언트 머신에 연결합니다.

1. https://console.aws.amazon.com/ec2/에서 Amazon EC2 콘솔을 엽니다.

2. 탐색 창에서 인스턴스를 선택합니다. 그런 다음 3단계: 클라이언트 머신 생성에서 생성한 클라이언트 머신 이름 옆의 확인란을 선택합니다.

3. 작업을 선택하고 연결을 선택합니다. 콘솔의 지침을 따라 클라이언트 머신에 연결합니다.

Java를 설치하고 Kafka 버전 환경 변수를 설정합니다.

1. 다음 명령을 실행하여 클라이언트 머신에 Java를 설치합니다.

   sudo yum -y install java-11

2. 다음 명령에서와 같이 환경 변수 KAFKA_VERSION에 MSK 클러스터의 Kafka 버전을 저장합니다. 이 정보는 설정 전반에 걸쳐 필요합니다.

   export KAFKA_VERSION={KAFKA VERSION}

   예를 들어 버전 3.6.0을 사용 중인 경우 다음 명령을 사용합니다.

   export KAFKA_VERSION=3.6.0

Apache Kafka를 다운로드하고 추출합니다.

1. Apache Kafka를 다운로드하려면 다음 명령을 실행합니다.

   wget https://archive.apache.org/dist/kafka/$KAFKA_VERSION/kafka_2.13-$KAFKA_VERSION.tgz

   참고

   다음 목록에는 문제가 발생할 경우 사용할 수 있는 몇 가지 대체 Kafka 다운로드 정보가 나와 있습니다.

   • 연결 문제가 발생하거나 미러 사이트를 사용하려는 경우 다음 명령에서와 같이 Apache 미러 선택기를 사용해 보세요.

     wget https://www.apache.org/dyn/closer.cgi?path=/kafka/$KAFKA_VERSION/kafka_2.13-$KAFKA_VERSION.tgz

   • Apache Kafka 웹 사이트에서 직접 적절한 버전을 다운로드합니다.

2. 이전 단계에 TAR 파일을 다운로드한 디렉토리에서 다음 명령을 실행합니다.

   tar -xzf kafka_2.13-$KAFKA_VERSION.tgz

3. 새로 생성된 디렉터리의 전체 경로를 KAFKA_ROOT 환경 변수 내에 저장합니다.

   export KAFKA_ROOT=$(pwd)/kafka_2.13-$KAFKA_VERSION

MSK 클러스터에 대한 인증을 설정합니다.

1. Amazon MSK IAM 클라이언트 라이브러리의 최신 버전을 찾습니다. 이 라이브러리를 사용하면 클라이언트 머신이 IAM 인증을 사용하여 MSK 클러스터에 액세스할 수 있습니다.

2. 다음 명령을 통해 $KAFKA_ROOT/libs 디렉터리로 이동하여 이전 단계에서 찾은 관련 Amazon MSK IAM JAR을 다운로드합니다. {LATEST VERSION}을 다운로드하는 실제 버전 번호로 바꿔야 합니다.

   cd $KAFKA_ROOT/libs

   wget https://github.com/aws/aws-msk-iam-auth/releases/latest/download/aws-msk-iam-auth-{LATEST VERSION}-all.jar

   참고

   MSK 클러스터와 상호 작용하는 Kafka 명령을 실행하기 전에 Java 클래스 경로에 Amazon MSK IAM JAR 파일을 추가해야 할 수 있습니다. 다음 예제에 표시된 대로 CLASSPATH 환경 변수를 설정합니다.

   export CLASSPATH=$KAFKA_ROOT/libs/aws-msk-iam-auth-{LATEST VERSION}-all.jar

   이를 통해 전체 세션에 대해 CLASSPATH가 설정되므로 모든 후속 Kafka 명령에서 JAR을 사용할 수 있습니다.

3. $KAFKA_ROOT/config 디렉터리로 이동하여 클라이언트 구성 파일을 생성합니다.

   cd $KAFKA_ROOT/config

4. 다음 속성 설정을 복사하여 새 파일에 붙여넣습니다. 파일을 client.properties(으)로 저장합니다.

   security.protocol=SASL_SSL
   sasl.mechanism=AWS_MSK_IAM
   sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;
   sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler

(선택 사항) Kafka 도구의 Java 힙 크기를 조정합니다.

메모리 관련 문제가 발생하거나 많은 수의 토픽 또는 파티션으로 작업하는 경우 Java 힙 크기를 조정할 수 있습니다. 이를 수행하려면 Kafka 명령을 실행하기 전에 KAFKA_HEAP_OPTS 환경 변수를 설정합니다.

다음 예제에서는 최대 힙 크기와 초기 힙 크기를 모두 512MB로 설정합니다. 특정 요구 사항 및 사용 가능한 시스템 리소스에 따라 이러한 값을 조정합니다.

export KAFKA_HEAP_OPTS="-Xmx512M -Xms512M"

클러스터 연결 정보를 가져옵니다.

1. https://console.aws.amazon.com/msk/에서 Amazon MSK 콘솔을 엽니다.

2. 클러스터 상태가 활성이 될 때까지 기다립니다. 몇 분 정도 걸릴 수 있습니다. 상태가 활성이 되면 클러스터 이름을 선택합니다. 클러스터 요약이 포함된 페이지로 이동합니다.

3. 클라이언트 정보 보기를 선택합니다.

4. 프라이빗 엔드포인트에 대한 연결 문자열을 복사합니다.

   각 브로커에 대해 3개의 엔드포인트가 제공됩니다. 다음 명령에서와 같이 이러한 연결 문자열 중 하나를 BOOTSTRAP_SERVER 환경 변수에 저장합니다. <bootstrap-server-string>을 연결 문자열의 실제 값으로 바꿉니다.

   export BOOTSTRAP_SERVER=<bootstrap-server-string>

다음 명령을 실행하여 토픽을 생성합니다.

$KAFKA_ROOT/bin/kafka-topics.sh --create --bootstrap-server $BOOTSTRAP_SERVER --command-config $KAFKA_ROOT/config/client.properties --replication-factor 3 --partitions 1 --topic MSKTutorialTopic

client.properties 파일에 대해 NoSuchFileException을 가져오는 경우 이 파일이 Kafka bin 디렉터리 내의 현재 작업 디렉터리에 있는지 확인합니다.

CLASSPATH=$KAFKA_ROOT/libs/aws-msk-iam-auth-{LATEST VERSION}-all.jar \
$KAFKA_ROOT/bin/kafka-topics.sh --create \
--bootstrap-server $BOOTSTRAP_SERVER \
--command-config $KAFKA_ROOT/config/client.properties \
--replication-factor 3 \
--partitions 1 \
--topic MSKTutorialTopic

(선택 사항) 토픽이 성공적으로 생성되었는지 확인합니다.

1. 명령이 성공하면 Created topic MSKTutorialTopic. 메시지가 표시됩니다.

2. 모든 토픽을 나열하여 토픽이 존재하는지 확인합니다.

   $KAFKA_ROOT/bin/kafka-topics.sh --list --bootstrap-server $BOOTSTRAP_SERVER --command-config $KAFKA_ROOT/config/client.properties

명령이 실패하거나 오류가 발생하면 Amazon MSK 클러스터 문제 해결 섹션에서 문제 해결 정보를 참조하세요.

(선택 사항) 이 자습서에서 사용한 환경 변수를 삭제합니다.

이 자습서의 다음 단계를 위해 환경 변수를 유지하려면 이 단계를 건너뜁니다. 아니면 다음 예제에서와 같이 이러한 변수를 설정 해제할 수 있습니다.

unset KAFKA_VERSION KAFKA_ROOT BOOTSTRAP_SERVER CLASSPATH KAFKA_HEAP_OPTS