# 자체 관리형 Apache Kafka 이벤트 소스에 대한 Lambda 이벤트 소스 매핑 생성
이벤트 소스 매핑을 생성하려면 Lambda 콘솔, [AWS Command Line Interface(CLI)](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html) 또는 [AWS SDK](https://aws.amazon.com/getting-started/tools-sdks/)를 사용할 수 있습니다.
다음 콘솔 단계에서는 자체 관리형 Apache Kafka 클러스터를 Lambda 함수의 트리거로 추가합니다. 그러면 내부적으로 이벤트 소스 매핑 리소스가 생성됩니다.
## 사전 조건
+ 자체 관리형 Apache Kafka 클러스터. Lambda는 Apache Kafka 버전 0.10.1.0 이상을 지원합니다.
+ 자체 관리형 Kafka 클러스터에서 사용하는 AWS 리소스에 액세스할 수 있는 권한이 있는 [실행 역할](lambda-intro-execution-role.md)입니다.
## 자체 관리형 Kafka 클러스터 추가(콘솔)
다음 단계에 따라 자체 관리형 Apache Kafka 클러스터 및 Kafka 주제를 Lambda 함수의 트리거로 추가합니다.
**Lambda 함수에 Apache Kafka 트리거를 추가하려면(콘솔)**
1. Lambda 콘솔의 [함수 페이지](https://console.aws.amazon.com/lambda/home#/functions)를 엽니다.
1. Lambda 함수의 이름을 선택합니다.
1. **함수 개요(Function overview)**에서 **트리거 추가(Add trigger)**를 선택합니다.
1. **트리거 구성**에서 다음을 수행합니다.
1. **Apache Kafka** 트리거 유형을 선택합니다.
1. **Bootstrap 서버**에는 클러스터의 Kafka 브로커 호스트 및 포트 페어 주소를 입력한 다음 **추가**를 선택합니다. 클러스터의 각 Kafka 브로커에 대해 이를 반복합니다.
1. **주제 이름**에는 클러스터에 레코드를 저장하는 데 사용되는 Kafka 주제의 이름을 입력합니다.
1. 프로비저닝 모드를 구성하는 경우 **최소 이벤트 폴러**의 값, **최대 이벤트 폴러**의 값, PollerGroupName 값(선택 사항)을 입력하여 동일한 이벤트 소스 VPC 내에서 여러 ESM의 그룹화를 지정합니다.
1. (선택 사항) **배치 크기(Batch size)**에 단일 배치에서 검색할 최대 레코드 수를 입력합니다.
1. **Batch window**에서 Lambda가 함수를 호출하기 전에 레코드를 수집하는 데 걸리는 최대 시간(초)을 입력합니다.
1. (선택 사항) **Consumer group ID**에서 가입할 Kafka 소비자 그룹의 ID를 입력합니다.
1. (선택 사항) **시작 위치**의 경우 최신 레코드에서 스트림 읽기를 시작하려면 **최신**을 선택하고, 사용 가능한 가장 빠른 레코드에서 시작하려면 **수평 트리밍**을 선택하고, 읽기를 시작할 타임스탬프를 지정하려면 **타임스탬프**를 선택합니다.
1. (선택 사항) **VPC**에서 Kafka 클러스터용 Amazon VPC를 선택합니다. 그런 다음 **VPC 서브넷(VPC subnets)**과 **VPC 보안 그룹(VPC security groups)**을 선택합니다.
VPC 내의 사용자만 브로커에 액세스하는 경우 이 설정이 필요합니다.
1. (선택 사항) **인증(Authentication)**에서 **추가(Add)**를 선택한 후 다음을 수행합니다.
1. 클러스터에 있는 Kafka 브로커의 액세스 또는 인증 프로토콜을 선택합니다.
+ Kafka 브로커가 SASL/PLAIN 인증을 사용하는 경우 **BASIC\$1AUTH**를 선택합니다.
+ 브로커가 SASL/SCRAM 인증을 사용하는 경우 **SASL\$1SCRAM** 프로토콜 중 하나를 선택합니다.
+ mTLS 인증을 구성하는 경우 **CLIENT\$1CERTIFICATE\$1TLS\$1AUTH** 프로토콜을 선택합니다.
1. SASL/SCRAM 또는 mTLS 인증의 경우 Kafka 클러스터에 대한 자격 증명이 포함된 Secrets Manager 비밀 키를 선택합니다.
1. (선택 사항) **암호화(Encryption)**에서 Kafka 브로커가 프라이빗 CA에서 서명한 인증서를 사용하는 경우 Kafka 브로커가 TLS 암호화에 사용하는 루트 CA 인증서가 포함된 Secrets Manager 암호를 선택합니다.
이 설정은 SASL/SCRAM 또는 SASL/PLANE에 대한 TLS 암호화 및 mTLS 인증에 적용됩니다.
1. 테스트를 위해 트리거를 비활성화된 상태에서 생성하려면(권장됨) **트리거 사용**을 선택 해제합니다. 또는 트리거를 즉시 사용하려면 **트리거 사용**을 선택합니다.
1. 트리거를 생성하려면 **추가**를 선택합니다.
## 자체 관리형 Kafka 클러스터 추가(AWS CLI)
다음 예제 AWS CLI 명령을 사용하여 Lambda 함수에 대한 자체 관리형 Apache Kafka 트리거를 생성하고 확인합니다.
### SASL/SCRAM 사용
Kafka 사용자가 인터넷을 통해 Kafka 브로커에 액세스한다면 SASL/SCRAM 인증용으로 생성한 Secrets Manager 비밀 정보를 지정합니다. 다음 예제에서는 [create-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/create-event-source-mapping.html) AWS CLI 명령을 사용하여 `my-kafka-function`이라는 Lambda 함수를 `AWSKafkaTopic`이라는 Kafka 주제에 매핑합니다.
```
aws lambda create-event-source-mapping \
--topics AWSKafkaTopic \
--source-access-configuration Type=SASL_SCRAM_512_AUTH,URI=arn:aws:secretsmanager:us-east-1:111122223333:secret:MyBrokerSecretName \
--function-name arn:aws:lambda:us-east-1:111122223333:function:my-kafka-function \
--self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["abc3.xyz.com:9092", "abc2.xyz.com:9092"]}}'
```
### VPC 사용
VPC 내의 Kafka 사용자만 Kafka 브로커에 액세스한다면 VPC, 서브넷 및 VPC 보안 그룹을 지정해야 합니다. 다음 예제에서는 [create-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/create-event-source-mapping.html) AWS CLI 명령을 사용하여 `my-kafka-function`이라는 Lambda 함수를 `AWSKafkaTopic`이라는 Kafka 주제에 매핑합니다.
```
aws lambda create-event-source-mapping \
--topics AWSKafkaTopic \
--source-access-configuration '[{"Type": "VPC_SUBNET", "URI": "subnet:subnet-0011001100"}, {"Type": "VPC_SUBNET", "URI": "subnet:subnet-0022002200"}, {"Type": "VPC_SECURITY_GROUP", "URI": "security_group:sg-0123456789"}]' \
--function-name arn:aws:lambda:us-east-1:111122223333:function:my-kafka-function \
--self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["abc3.xyz.com:9092", "abc2.xyz.com:9092"]}}'
```
### AWS CLI를 사용하여 상태 확인
다음 예제에서는 [get-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/get-event-source-mapping.html) AWS CLI 명령을 사용하여 생성한 이벤트 소스 매핑의 상태를 설명합니다.
```
aws lambda get-event-source-mapping
--uuid dh38738e-992b-343a-1077-3478934hjkfd7
```