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

해결 방법

VPC 피어링 설정

Amazon MSK 클러스터는 서로 다른 Virtual Private Cloud(VPC)에 있으므로 VPC 피어링 연결을 만들어야 합니다. 같은 AWS 리전 또는 다른 리전에 있는 다른 AWS 계정의 VPC와 이 VPC 피어링 연결을 생성합니다. 자세한 내용은 VPC 피어링 연결 생성을 참조하십시오.

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

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

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/

Apache Kafka와 MSK Connect에는 MirrorMaker 라이브러리가 기본 제공되므로 이 기능을 위해 .jar 파일을 추가할 필요는 없습니다. MSK Connect에는 커넥터 생성 시 사용자 지정 플러그인이 있어야 한다는 전제 조건이 있습니다. 따라서 참조용으로 빈 커넥터를 생성해야 합니다.

3.    타겟 계정에서 .zip 파일을 사용하여 사용자 정의 플러그인을 생성합니다. 사용자 지정 플러그인의 이름으로 mm2-connect-plugin을 사용합니다.

MSK Connect 커넥터 생성

다음 단계를 완료하여 타겟 계정에 커넥터를 생성합니다.

1.    Amazon MSK 콘솔을 엽니다.

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

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

4.    사용자 지정 플러그인 목록에서 생성한 사용자 지정 플러그인 옆에 있는 플러그인을 선택하고 다음을 선택합니다.

5.    커넥터 이름과 선택적으로 커넥터 이름을 입력합니다.

6.    클러스터 목록에서 타겟 클러스터를 선택합니다.

7.    다음 구성을 복사한 다음 커넥터 구성 필드에 붙여 넣습니다. 사용 사례에 맞게 예시를 수정하십시오.

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_MSK_Bootstrap_Server_PLAINTEXT>  
source.cluster.security.protocol=PLAINTEXT  

# Destination cluster options  
target.cluster.bootstrap.servers=<Target_MSK_Bootstrap_Server_PLAINTEXT>  
target.cluster.security.protocol=PLAINTEXT

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

9.    워커 구성에서 MSK 기본 구성 사용을 선택합니다.

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

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

12.    선택적으로 로그 페이지에서 로그 전달을 설정합니다. 그런 후 다음을 선택합니다.

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

참고: 이 구성을 사용하면 소스 클러스터에서 각 토픽을 복제하기 위해 MM2가 타겟 클러스터에 두 개의 토픽을 생성합니다. 예를 들어, 소스 클러스터에 exampleTopic1 토픽이 있는 경우 MM2는 타겟 클러스터에 primary.exampleTopic1 및 exampleTopic1 토픽을 생성합니다. 메시지는 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 토픽을 생성합니다.

<path-to-your-kafka-installation>/bin/kafka-topics.sh --create --bootstrap-server <Source MSK cluster BootstrapServerString> --replication-factor 3 --partitions 1 --topic exampletopic1

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

<path-to-your-kafka-installation>/bin/kafka-console-producer.sh --broker-list <Source MSK cluster BootstrapServerString> --topic exampletopic1

>message 1  
>message 2

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

<path-to-your-kafka-installation>/bin/kafka-topics.sh --bootstrap-server <Target MSK cluster BootstrapServerString> --list

출력은 다음과 비슷합니다.

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

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

<path-to-your-kafka-installation>/bin/kafka-console-consumer.sh --bootstrap-server <Target MSK cluster BootstrapServerString> --topic exampletopic1 --from-beginning

message 1  
message 2

관련 정보

Apache Kafka의 MirrorMaker를 사용하여 클러스터 마이그레이션