MSK(Amazon Managed Streaming for Apache Kafka)에 대한 주제 등록

1. 소개

Amazon MSK(Amazon Managed Streaming for Apache Kafka)는 Apache Kafka의 관리형 서비스입니다. Amazon MSK 주제 등록 포함한 설정 변경은 Amazon CLI를 통해 이루어질 수 있습니다.
기본적으로 Kafka Broker에 대해 Kafka Producer는 주제를 생성할 수 없습니다. 왜냐하면, 토픽의 자동 생성을 ON으로 하는 파라미터인 auto.create.topics.enable ,가 true가 되어 있지 않기 때문입니다.
따라서 이 게시물에서는 먼저 Amazon CLI를 통해 주제 활성화 설정을 수행한 다음 Kafka Producer를 통해 주제를 등록하는 단계를 설명합니다.

2. 설정 방법

2-1. 보안 그룹 설정

Amazon CLI를 통해 Amazon MSK를 작동하려면 Kafka Cluster를 구축할 때 Amazon CLI에서 작동하는 단말기(Amazon CLI 단말기)를 액세스 가능한 보안 그룹에 속해야 합니다.
Kafka Cluster를 구축하려면 Amazon CLI 터미널이 포함된 보안 그룹을 액세스 대상으로 포함해야 합니다.

2-2. arn 확인

　
Amazon Management Consol에서 Amazon MSK의 arn을 확인합니다. 이 게시물을 위해서만 만들었으므로 arn은 이미 유효한 값이 아닙니다.



이 경우 arn은 다음 값입니다.

arn:aws:kafka:ap-northeast-1:196585472650:cluster/Amazon-MSK-Test/d5a6aebf-8858-43db-bfe4-4a7288d93775-2

2-3. bootstrap-brokers (Zookeeper)의 연결 대상 확인

describe-cluster 명령을 사용하여 bootstrap-brokers(Zookeeper)의 연결 대상을 확인합니다. 이 경우 "z-1.amazon-msk-test.sxm7ay.c2.kafka.ap-northeast-1.amazonaws.com:2181,z-3.amazon-msk-test.sxm7ay.c2.kafka.ap- northeast-1.amazonaws.com:2181,z-2.amazon-msk-test.sxm7ay.c2.kafka.ap-northeast-1.amazonaws.com:2181"입니다.

aws kafka describe-cluster --region ap-northeast-1 --cluster-arn arn:aws:kafka:ap-northeast-1:196585472650:cluster/Amazon-MSK-Test/d5a6aebf-8858-43db-bfe4-4a7288d93775-2
{
    "ClusterInfo": {
        "EncryptionInfo": {
            "EncryptionInTransit": {
                "ClientBroker": "TLS", 
                "InCluster": true
            }, 
            "EncryptionAtRest": {
                "DataVolumeKMSKeyId": "arn:aws:kms:ap-northeast-1:196585472650:key/d94c7909-3392-482a-908d-d25f9a31abb9"
            }
        }, 
        "BrokerNodeGroupInfo": {
            "BrokerAZDistribution": "DEFAULT", 
            "ClientSubnets": [
                "subnet-05e68568d43a7c6d3", 
                "subnet-091efbcc876bdefb1", 
                "subnet-05a73c6877da30604"
            ], 
            "StorageInfo": {
                "EbsStorageInfo": {
                    "VolumeSize": 1
                }
            }, 
            "SecurityGroups": [
                "sg-0204c59a2c6992390", 
                "sg-0fc8fe66209887218", 
                "sg-0c4b89523751d689c"
            ], 
            "InstanceType": "kafka.m5.large"
        }, 
        "ClusterName": "Amazon-MSK-Test", 
        "CurrentBrokerSoftwareInfo": {
            "KafkaVersion": "2.3.1", 
            "ConfigurationRevision": 1, 
            "ConfigurationArn": "arn:aws:kafka:ap-northeast-1:196585472650:configuration/SalesClusterConfiguration2/9c66b8dc-836e-4348-a2a3-b2bd19717b0b-2"
        }, 
        "Tags": {
            "Name": "Kafka-Cluster"
        }, 
        "CreationTime": "2019-12-19T02:09:06.63Z", 
        "NumberOfBrokerNodes": 3, 
        "ZookeeperConnectString": "z-1.amazon-msk-test.sxm7ay.c2.kafka.ap-northeast-1.amazonaws.com:2181,z-3.amazon-msk-test.sxm7ay.c2.kafka.ap-northeast-1.amazonaws.com:2181,z-2.amazon-msk-test.sxm7ay.c2.kafka.ap-northeast-1.amazonaws.com:2181", 
        "State": "ACTIVE", 
        "CurrentVersion": "K2EUQ1WTGCTBG2", 
        "ClusterArn": "arn:aws:kafka:ap-northeast-1:196585472650:cluster/Amazon-MSK-Test/d5a6aebf-8858-43db-bfe4-4a7288d93775-2", 
        "EnhancedMonitoring": "DEFAULT", 
        "OpenMonitoring": {
            "Prometheus": {
                "NodeExporter": {
                    "EnabledInBroker": false
                }, 
                "JmxExporter": {
                    "EnabledInBroker": false
                }
            }
        }
    }
}

2-4. 설정 파일 만들기

Amazon CLI 단말기의 모든 디렉토리에 구성 변경을 위한 파일을 생성합니다. 구성 파일의 형식은 Amazon MSK에서 지정한 형식이어야 합니다. 여기을 참조하십시오.

/home/ec2-user/test/configuration.txt

auto.create.topics.enable = true
message.max.bytes = 73400320

2-5. Amazon MSK 공식 파이썬 도구 준비

구성 파일과 동일한 디렉토리 계층 구조에 다음 python 파일을 만듭니다. 이 경우에는 configuration.txt 와 같은 디렉토리인 home/ec2-user/ 에 배치합니다.
파이썬 파일의 병아리에 대해서는 여기을 참조하십시오.

/home/ec2-user/unction.py

import boto3

client = boto3.client('kafka')

config_file = open('/home/ec2-user/test/configuration.txt', 'r')

server_properties = config_file.read()

response = client.create_configuration(
            #この値はAWS アカウントでユニークにする必要があります。
            Name='SalesClusterConfiguration3',
                Description='The configuration to use on all sales clusters.',
                    KafkaVersions=['1.1.1', '2.1.0','2.3.1'],
                        ServerProperties=server_properties
                        )
print(response)

boto3 라이브러리가 없으면 pip 명령으로 설치하십시오.

pip install boto3

2-6. 설정 반영

python 명령을 실행하여 설정을 반영할 수 있습니다. 다음과 같이 'HTTPStatusCode': 200가 회신되면 설정 변경이 가능합니다.

python function.py 
{u'LatestRevision': {u'CreationTime': datetime.datetime(2019, 12, 19, 2, 42, 53, 2000, tzinfo=tzlocal()), u'Description': u'The configuration to use on all sales clusters.', u'Revision': 1}, u'Arn': u'arn:aws:kafka:ap-northeast-1:196585472650:configuration/SalesClusterConfiguration3/30ffe529-2733-486d-bfed-1f1641db9edf-2', u'CreationTime': datetime.datetime(2019, 12, 19, 2, 42, 53, 2000, tzinfo=tzlocal()), u'Name': u'SalesClusterConfiguration3', 'ResponseMetadata': {'RetryAttempts': 0, 'HTTPStatusCode': 200, 'RequestId': '939d9052-a6f7-4fa4-b66f-91fe737a8057', 'HTTPHeaders': {'x-amzn-requestid': '939d9052-a6f7-4fa4-b66f-91fe737a8057', 'x-amz-cf-pop': 'NRT12-C2', 'content-length': '347', 'via': '1.1 4cb3df5349fbb69c930b315b7d0a5272.cloudfront.net (CloudFront)', 'x-cache': 'Miss from cloudfront', 'x-amz-apigw-id': 'E7iC8G5xNjMFeFw=', 'x-amzn-trace-id': 'Root=1-5dfae3ac-0cd2e93c1fe9319eb9166bc2;Sampled=0', 'connection': 'keep-alive', 'x-amz-cf-id': 'zSz8M9gaSNl4bIuNUdbDfY_Sm4cEWCoZF3RfuAX22aQBZxHJOUI4ug==', 'date': 'Thu, 19 Dec 2019 02:42:53 GMT', 'content-type': 'application/json'}}}

2-7. 주제 등록

Kafka Producer에서 주제를 등록합니다. Created topic "XXXXXXX". 및 결과가 반환되면 주제 생성 완료입니다.

kafka-topics --zookeeper "z-1.amazon-msk-test.sxm7ay.c2.kafka.ap-northeast-1.amazonaws.com:2181,z-3.amazon-msk-test.sxm7ay.c2.kafka.ap-northeast-1.amazonaws.com:2181,z-2.amazon-msk-test.sxm7ay.c2.kafka.ap-northeast-1.amazonaws.com:2181" --create --topic first-test --partitions 3 --replication-factor 3
Created topic "first-test".

이상