MSK Connect에서 실행되는 MirrorMaker 2로 서로 다른 계정의 Amazon MSK 클러스터 간에 데이터를 전송하려면 어떻게 해야 합니까?

해결 방법

VPC 피어링 설정

Amazon MSK 클러스터가 서로 다른 Amazon Virtual Private Cloud(Amazon VPC)에 있는 경우 VPC 피어링 연결을 만들어야 합니다.

소스 Amazon MSK 클러스터와 연결된 보안 그룹은 대상 클러스터의 보안 그룹에서 오는 모든 트래픽을 허용해야 합니다. 또한 대상 클러스터의 보안 그룹은 소스 Amazon MSK 클러스터의 보안 그룹에서 오는 모든 트래픽을 허용해야 합니다. 자세한 내용은 피어 보안 그룹을 참조하도록 보안 그룹 업데이트를 참조하십시오.

참고: 다른 계정의 보안 그룹을 참조하려면 소스 또는 대상 필드에 계정 번호를 입력합니다.

MM2 플러그인 정보로 플러그인 만들기

참고: MSK Connect 사용자 지정 플러그인은 .jar 또는 .zip 형식의 파일 또는 폴더를 허용합니다.

다음 단계를 완료하십시오.

1. 테스트 폴더를 만든 다음 폴더를 압축하려면 다음 명령을 실행합니다.

   mkdir mm2
   zip mm2.zip mm2

2. .zip 개체를 대상 계정의 Amazon Simple Storage Service(Amazon S3) 버킷으로 업로드합니다.

   aws s3 cp mm2.zip s3://mytestbucket/

   참고: **s3://mytestbucket/**을 Amazon S3 버킷 URL로 바꾸십시오.
   Apache Kafka와 MSK Connect에는 MirrorMaker 라이브러리가 기본 제공되므로 이 기능을 위해 .jar 파일을 추가할 필요는 없습니다.

3. 대상 계정에서.zip 파일을 사용하여 사용자 지정 플러그인을 만들고 파일 이름을 mm2-connect-plugin으로 지정합니다.
   참고: MSK Connect를 사용하는 경우 커넥터 생성 시 사용자 지정 플러그인이 있어야 합니다.

대상 계정에 MSK Connect 커넥터 만들기

다음 단계를 완료하십시오.

1. Amazon MSK 콘솔을 엽니다.

2. 탐색 창의 MSK Connect에서 커넥터를 선택합니다.

3. 커넥터 생성을 선택합니다.

4. 기존 사용자 지정 플러그인 사용을 선택합니다.

5. 사용자 지정 플러그인에서 자신이 만든 사용자 지정 플러그인을 선택한 후 다음을 선택합니다.

6. 커넥터 이름을 입력합니다.
   (선택 사항) 설명을 입력합니다.

7. 클러스터 목록에서 대상 클러스터를 선택합니다.

8. 다음 예시와 유사한 구성을 커넥터 구성 필드에 추가합니다.

   connector.class=org.apache.kafka.connect.mirror.MirrorSourceConnector  
   tasks.max=1  
   
   clusters=primary,replica  
   source.cluster.alias=primary  
   target.cluster.alias=replica  
   
   topics=example.*  
   replication.factor=2  
   topic.creation.default.replication.factor=2  
   topic.creation.default.partitions=2  
   consumer.group.id=mm2-connector  
   
   refresh.groups.interval.seconds=20  
   refresh.topics.interval.seconds=20  
   
   sync.topic.configs.interval.seconds=20  
   sync.topic.acls.interval.seconds=20  
   
   producer.enable.idempotence=true  
   
   transforms=renameTopic  
   transforms.renameTopic.type=org.apache.kafka.connect.transforms.RegexRouter  
   transforms.renameTopic.regex=primary.(.*)  
   transforms.renameTopic.replacement=$1  
   
   key.converter=org.apache.kafka.connect.converters.ByteArrayConverter  
   value.converter=org.apache.kafka.connect.converters.ByteArrayConverter  
   
   # Source cluster options  
   source.cluster.bootstrap.servers=  
   source.cluster.security.protocol=PLAINTEXT  
   
   # Target cluster options  
   target.cluster.bootstrap.servers=  
   target.cluster.security.protocol=PLAINTEXT

9. 커넥터의 용량을 설정합니다.

10. 작업자 구성에서 MSK 기본 구성 사용을 선택합니다.

11. 액세스 권한에서 MSK Connect에 필요한 권한을 제공하는 AWS Identity and Access Management(AWS IAM) 역할을 선택합니다. 그 후 다음을 선택합니다.

12. 보안 페이지의 암호화 - 전송 중에서 일반 텍스트 트래픽을 선택합니다. 그 후 다음을 선택합니다.

13. (선택 사항) 로그 페이지에서 로그 전달을 설정합니다. 그 후 다음을 선택합니다.

14. 검토 및 생성에서 커넥터 생성을 선택합니다.

참고: 이전 구성을 사용하면 소스 클러스터에서 각 토픽을 복제하기 위해 MM2가 대상 클러스터에 두 개의 토픽을 만듭니다. 예를 들어, 소스 클러스터에 exampleTopic1 토픽이 있는 경우 MM2는 대상 클러스터에 primary.exampleTopic1 및 exampleTopic1 토픽을 만듭니다. 그런 다음 MM2는 메시지를 exampleTopic1 토픽으로 라우팅합니다.

클라이언트 인스턴스 만들기

토픽을 만들고 토픽에서 데이터를 생성하거나 사용하려면 클라이언트 인스턴스를 만들어야 합니다.

다음 단계를 완료하십시오.

1. Amazon Elastic Compute Cloud(Amazon EC2) 인스턴스를 시작한 다음 인스턴스에 연결합니다.

2. 클라이언트 컴퓨터에 Java를 설치하려면 다음 명령을 실행합니다.

   sudo yum -y install java-11

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

   wget https://archive.apache.org/dist/kafka/2.8.1/kafka_2.12-2.8.1.tgz
   tar -xzf kafka_2.12-2.8.1.tgz

4. 소스 계정의 Amazon MSK 클러스터에 exampletopic1 토픽을 만듭니다.

   kafka-installation-path/bin/kafka-topics.sh --create --bootstrap-server SourceMSKclusterBootstrapServerString --replication-factor 3 --partitions 1 --topic exampletopic1

   참고: kafka-installation-path를 Kafka가 설치된 시스템 경로로 바꾸십시오. SourceMSKclusterBootstrapServerString을 소스 Amazon MSK 클러스터의 부트스트랩 서버 문자열로 바꾸십시오.

5. 소스 계정의 클러스터에 데이터를 생성합니다.

   Kafka-installation-path/bin/kafka-console-producer.sh --broker-list SourceMSKclusterBootstrapServerString --topic exampletopic1

   참고: kafka-installation-path를 Kafka가 설치된 시스템 경로로 바꾸십시오. SourceMSKclusterBootstrapServerString을 소스 Amazon MSK 클러스터의 부트스트랩 서버 문자열로 바꾸십시오.
   예상 출력:

   >message 1
   >message 2

6. 대상 계정의 클러스터에 있는 토픽을 나열합니다.

   Kafka-installation-path/bin/kafka-topics.sh --bootstrap-server TargetMSKclusterBootstrapServerString --list

   참고: kafka-installation-path를 Kafka가 설치된 시스템 경로로 바꾸십시오. TargetMSKclusterBootstrapServerString을 대상 Amazon MSK 클러스터의 부트스트랩 서버 문자열로 바꾸십시오.
   출력 예시:

   __amazon_msk_canary  __amazon_msk_connect_configs_mm2-*****  
   __amazon_msk_connect_offsets_mm2-*****  
   __amazon_msk_connect_status_mm2-*****  
   __consumer_offsets  
   exampleTopic1  
   primary.exampleTopic1

7. 대상 클러스터의 데이터를 사용합니다.

   Kafka-installation-path/bin/kafka-console-consumer.sh --bootstrap-server TargetMSKclusterBootstrapServerString --topic exampletopic1 --from-beginning

   참고: kafka-installation-path를 Kafka가 설치된 시스템 경로로 바꾸십시오. TargetMSKclusterBootstrapServerString을 대상 Amazon MSK 클러스터의 부트스트랩 서버 문자열로 바꾸십시오.
   예상 출력:

   >message 1
   >message 2

관련 정보

Amazon MSK 클러스터로 마이그레이션