Day 2 - Lambda 이벤트 소스 매핑: SQS·Kinesis·DynamoDB Streams 내부 폴링 메커니즘

Lambda 함수는 "누군가 호출해야 실행된다"는 원칙에서 출발한다. 그 "누군가"가 누구인지에 따라 Lambda의 동작 방식 — 재시도 횟수, 에러 처리, 순서 보장 여부 — 이 완전히 달라진다. 이걸 이해하지 못하면 SQS 메시지가 무한 반복 처리되거나, Kinesis 샤드가 통째로 막히거나, S3 처리가 무한 루프에 빠지는 사고를 당한다.

Lambda의 호출 방식은 세 가지 범주로 나뉜다. 첫째, 동기 호출(Synchronous Invocation): 호출자가 직접 InvokeFunction API를 RequestResponse 모드로 부른다. API Gateway, ALB, Cognito가 여기 속한다. 둘째, 비동기 호출(Asynchronous Invocation): 이벤트가 Lambda 서비스의 내부 큐에 들어가고 Lambda가 0~2회 재시도한다. S3, SNS, EventBridge가 여기 속한다. 셋째, 이벤트 소스 매핑(Event Source Mapping, ESM): Lambda 서비스 자체가 소스를 능동적으로 폴링한다. SQS, Kinesis, DynamoDB Streams, MSK, Apache Kafka가 여기 속한다.

동기 호출의 내부 동작: 재시도는 호출자 책임

API Gateway → Lambda 경로를 따라가면 이 흐름이 보인다. 클라이언트가 REST API 요청을 보내면, API Gateway는 lambda:InvokeFunction을 InvocationType=RequestResponse로 호출한다. Lambda 서비스는 요청을 받아 실행 환경에 전달하고, 함수가 반환할 때까지 블록된 상태로 대기한다. 응답이 오면 API Gateway가 HTTP 응답으로 변환해 클라이언트에 돌려준다.

동기 호출에서 함수가 에러를 던지면 Lambda는 HTTP 200 OK 응답을 주지만, FunctionError 헤더를 붙인다. 응답 페이로드는 에러 메시지를 담은 JSON이다. 재시도는 호출자의 책임이다 — API Gateway는 기본적으로 재시도하지 않는다.

import boto3
import json
 
lambda_client = boto3.client('lambda')
 
response = lambda_client.invoke(
    FunctionName='my-function',
    InvocationType='RequestResponse',
    Payload=json.dumps({'key': 'value'}).encode()
)
 
# HTTP 200이어도 함수 내부 에러일 수 있음
if 'FunctionError' in response:
    error = json.loads(response['Payload'].read())
    print(f"함수 에러: {error['errorMessage']}")  # 재시도는 이 코드가 결정
else:
    result = json.loads(response['Payload'].read())

비동기 호출: Lambda 내부 큐와 재시도 메커니즘

S3 버킷에 파일이 업로드되면 S3가 InvocationType=Event로 Lambda를 호출한다. Lambda 서비스는 이벤트를 즉시 내부 이벤트 큐(SQS와 유사하지만 외부에서 보이지 않는 내부 큐)에 넣고 S3에 202 Accepted를 반환한다. S3는 함수가 실제로 실행됐는지, 성공했는지 알지 못한다.

Lambda 서비스는 큐에서 이벤트를 꺼내 함수를 실행하고, 실패하면 최대 2회 재시도한다. 재시도 간격은 1분, 2분으로 지수 백오프다. 그래도 실패하면 설정된 DLQ나 Destination으로 이벤트를 전송한다.

# Destination 설정 (성공 + 실패 모두)
aws lambda put-function-event-invoke-config \
  --function-name image-processor \
  --maximum-retry-attempts 2 \
  --maximum-event-age-in-seconds 3600 \
  --destination-config '{
    "OnSuccess": {
      "Destination": "arn:aws:sqs:ap-northeast-2:123:success-audit-queue"
    },
    "OnFailure": {
      "Destination": "arn:aws:events:ap-northeast-2:123:event-bus/ops-bus"
    }
  }'

⚠️ 함정: 비동기 호출에서 Lambda는 같은 이벤트를 최대 3번(원본 + 2회 재시도) 실행할 수 있다. 이는 at-least-once delivery 보장이다. 함수가 멱등적(idempotent)이지 않으면 중복 처리 문제가 생긴다. 예를 들어 "사용자에게 이메일 보내기" 함수가 비동기로 호출되고 두 번 성공하면 이메일이 두 통 간다.

S3 이벤트: 비동기 호출과 무한 루프 방지

S3 이벤트 알림은 두 가지 방식이 있다.

S3 Event Notifications(레거시): 버킷 설정에서 직접 Lambda, SQS, SNS로 이벤트를 발송한다. 간단하지만 필터링이 prefix/suffix 정도에 그치고, 하나의 이벤트 타입에 하나의 대상만 설정 가능하다.

S3 → EventBridge(현재 권장): EventBridge를 경유해 더 세밀한 패턴 매칭, 여러 대상으로 팬아웃, 이벤트 아카이브와 리플레이가 가능하다.

S3 이벤트 처리에서 대표적인 사고는 무한 루프다.

[원본 버킷] 파일 업로드 → Lambda 실행 → 처리 결과를 [같은 버킷]에 저장
    ↑                                                              ↓
    └──────────────── 새 업로드 이벤트 발생 → Lambda 재실행 ────────┘

해결책 3가지:

1. 입출력 버킷 분리: 처리 결과를 다른 버킷에 저장한다.
2. Prefix/Suffix 필터: input/ 폴더 업로드만 트리거하도록 설정하고, 결과는 output/에 저장.
3. 객체 태그 검사: 처리된 객체에 processed=true 태그를 달고, 함수가 이 태그를 먼저 확인한 뒤 건너뛴다.

def lambda_handler(event, context):
    s3 = boto3.client('s3')
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']
        
        # 이미 처리된 객체인지 확인 (방어 로직)
        try:
            tags = s3.get_object_tagging(Bucket=bucket, Key=key)
            for tag in tags['TagSet']:
                if tag['Key'] == 'processed' and tag['Value'] == 'true':
                    print(f"이미 처리된 객체 건너뜀: {key}")
                    return
        except Exception:
            pass
        
        # 처리 로직
        process_object(bucket, key)
        
        # 처리 완료 태그 추가
        s3.put_object_tagging(
            Bucket=bucket, Key=key,
            Tagging={'TagSet': [{'Key': 'processed', 'Value': 'true'}]}
        )

SQS Event Source Mapping: Lambda가 능동적으로 폴링

SQS와 Lambda의 통합은 처음 보면 직관에 반한다. SQS가 Lambda를 호출하는 게 아니라, Lambda 서비스가 SQS를 폴링한다. 이게 이벤트 소스 매핑(ESM)의 핵심이다.

Lambda 서비스는 SQS 큐를 롱 폴링(Long Polling, 최대 20초 대기)으로 메시지를 가져온다. 배치로 묶어서 함수를 호출하며, 함수가 성공하면 메시지를 삭제한다. 실패하면 가시성 타임아웃(Visibility Timeout)이 끝난 후 메시지가 큐에 다시 나타난다. maxReceiveCount(기본 3)를 초과하면 SQS DLQ로 이동한다.

핵심 ESM 파라미터:

SQS ESM의 폴링은 Lambda 서비스가 SQS API ReceiveMessage를 주기적으로 호출하는 방식이다. 트래픽이 없을 때는 폴러 수가 줄어들고(최소 1개), 메시지가 쌓이면 폴러가 자동으로 늘어난다. 표준 큐에서는 최대 1,000개의 폴러가 동시에 동작할 수 있어 높은 처리량을 낼 수 있다. FIFO 큐에서는 메시지 그룹 ID당 하나의 Lambda 인스턴스가 처리해 순서를 보장한다.

Partial Batch Response: 성공한 메시지는 보존하기

SQS ESM에서 배치 중 일부 메시지만 실패하면 기본 동작은 배치 전체가 실패로 간주된다. 성공한 9개 메시지가 1개 실패 때문에 전부 다시 처리되는 것이다.

ReportBatchItemFailures를 활성화하면 Lambda가 실패한 메시지 ID만 반환해 그것만 재처리한다.

def lambda_handler(event, context):
    batch_item_failures = []
    
    for record in event['Records']:
        try:
            process_message(record)
        except Exception as e:
            # 이 메시지 ID만 실패로 보고
            batch_item_failures.append({
                'itemIdentifier': record['messageId']
            })
            print(f"실패: {record['messageId']}, 에러: {e}")
    
    # 실패 목록 반환 — Lambda가 이것만 큐에 되돌림
    return {'batchItemFailures': batch_item_failures}

# SQS ESM 생성 (Partial Batch Response 활성화)
aws lambda create-event-source-mapping \
  --function-name order-processor \
  --event-source-arn arn:aws:sqs:ap-northeast-2:123:orders-queue \
  --batch-size 100 \
  --maximum-batching-window-in-seconds 5 \
  --function-response-types ReportBatchItemFailures

⚠️ 함정: Partial Batch Response를 활성화해도 함수가 예외를 던지면(return이 아닌 raise) 전체 배치가 실패로 처리된다. 반드시 try/except로 감싸고 batchItemFailures를 반환해야 한다. 또한 SQS FIFO 큐에서는 실패한 메시지 뒤의 메시지들이 모두 블록된다는 점에 주의해야 한다.

Kinesis Data Streams ESM: 샤드 블록 문제

Kinesis와 Lambda의 통합에서 대표적인 함정은 샤드 블록이다.

Kinesis는 샤드(shard) 단위로 데이터를 처리한다. 각 샤드에서 레코드 순서가 보장된다. Lambda ESM은 샤드당 하나의 Lambda 인스턴스를 할당해 순서대로 처리한다.

문제: 하나의 레코드가 처리에 실패하면 Lambda는 그 레코드를 재시도한다. 재시도가 계속 실패하면 새 레코드를 처리할 수 없다 — 샤드 전체가 막힌다. CloudWatch IteratorAge 메트릭이 계속 증가하는 것으로 감지할 수 있다.

해결책:

# Kinesis ESM 생성 (샤드 블록 방지 설정)
aws lambda create-event-source-mapping \
  --function-name kinesis-processor \
  --event-source-arn arn:aws:kinesis:ap-northeast-2:123:stream/events \
  --starting-position LATEST \
  --batch-size 100 \
  --maximum-retry-attempts 3 \
  --maximum-record-age-in-seconds 300 \
  --bisect-batch-on-function-error \
  --destination-config '{"OnFailure": {"Destination": "arn:aws:sqs:ap-northeast-2:123:failed-records"}}'

• MaximumRetryAttempts: 무한 재시도 대신 최대 횟수 설정 (0이면 재시도 없음)
• MaximumRecordAgeInSeconds: 이 시간이 지난 레코드는 폐기
• BisectBatchOnFunctionError: 실패 배치를 절반으로 나눠 이진 탐색으로 문제 레코드 격리
• OnFailure Destination: 최종 실패 레코드를 SQS나 SNS로 전송

ParallelizationFactor는 처리량을 높이는 고급 기능이다. 같은 샤드에서 최대 10개의 Lambda를 동시에 실행할 수 있다. 순서는 파티션 키 단위로 유지된다.

aws lambda create-event-source-mapping \
  --function-name kinesis-processor \
  --event-source-arn arn:aws:kinesis:ap-northeast-2:123:stream/events \
  --starting-position LATEST \
  --parallelization-factor 5

Kinesis ESM의 파티션 키 단위 순서 보장은 분산 스트림 처리의 partition-based ordering 원칙과 같다. 순서 보장과 병렬 처리는 트레이드오프 관계다 — 완전한 순서 보장을 원하면 샤드당 1개 Lambda로 직렬 처리해야 하고, 높은 처리량을 원하면 순서를 키 단위로만 보장하며 병렬화한다.

DynamoDB Streams ESM: 변경 이벤트 처리

DynamoDB Streams는 테이블의 INSERT, MODIFY, REMOVE 이벤트를 24시간 동안 보존하는 스트림이다. Lambda ESM이 이 스트림을 폴링해 변경 이벤트를 처리한다.

DynamoDB Streams ESM은 Kinesis와 유사하게 동작하지만 차이점이 있다.

스트림 뷰 타입에 따라 함수가 받는 데이터가 달라진다.

def lambda_handler(event, context):
    for record in event['Records']:
        event_type = record['eventName']  # INSERT, MODIFY, REMOVE
        
        if event_type == 'INSERT':
            new_item = record['dynamodb']['NewImage']
            print(f"새 항목: {new_item}")
        
        elif event_type == 'MODIFY':
            old_item = record['dynamodb']['OldImage']
            new_item = record['dynamodb']['NewImage']
            # 변경된 필드 감지
            changed_fields = {
                k for k in new_item 
                if old_item.get(k) != new_item.get(k)
            }
            print(f"변경된 필드: {changed_fields}")
        
        elif event_type == 'REMOVE':
            old_item = record['dynamodb']['OldImage']
            print(f"삭제된 항목: {old_item}")

EventBridge 스케줄 + Lambda

EventBridge(구 CloudWatch Events)는 두 가지 방식으로 Lambda를 트리거한다.

Rate 표현식: rate(5 minutes), rate(1 hour) — 일정 간격으로 실행.

Cron 표현식: AWS의 cron은 표준 Linux cron과 다르다. 6개 필드를 사용하며 초 대신 연도를 포함한다. cron(분 시간 일 월 요일 연도).

# 매일 오전 9시(UTC) 실행
aws events put-rule \
  --name daily-report \
  --schedule-expression "cron(0 9 * * ? *)"
 
# 5분마다 실행
aws events put-rule \
  --name health-check \
  --schedule-expression "rate(5 minutes)"

AWS EventBridge cron에서 일(day-of-month)과 요일(day-of-week)은 동시에 지정할 수 없다. 하나를 지정하면 다른 하나는 반드시 ?(무관)로 설정해야 한다. cron(0 9 15 * ? *) (매월 15일), cron(0 9 ? * MON-FRI *) (평일)처럼 사용한다.

멱등성: at-least-once를 safely-exactly-once로

비동기 호출과 ESM 폴링은 모두 at-least-once 전달을 보장한다. 동일 이벤트가 두 번 처리될 수 있다. 함수를 멱등적으로 만들어야 중복 처리를 방어할 수 있다.

Lambda Powertools Idempotency 모듈은 DynamoDB를 이용해 이벤트 ID를 관리한다.

from aws_lambda_powertools.utilities.idempotency import (
    idempotent, DynamoDBPersistenceLayer
)
 
persistence_layer = DynamoDBPersistenceLayer(
    table_name="IdempotencyTable"
)
 
@idempotent(persistence_store=persistence_layer)
def lambda_handler(event, context):
    # 이 코드는 같은 이벤트 ID로 다시 불려도 한 번만 실행됨
    charge_customer(event['customerId'], event['amount'])
    return {'charged': True}

내부 동작: 첫 호출 시 이벤트 ID(기본은 이벤트 전체의 해시)를 DynamoDB에 "IN_PROGRESS" 상태로 저장. 성공하면 "COMPLETED"와 결과를 저장. 동일 이벤트 ID로 다시 오면 DynamoDB에서 이전 결과를 꺼내 그대로 반환 — Lambda 코드는 실행되지 않는다.

호출 유형 종합 비교

정리

Lambda의 세 가지 호출 방식 — 동기, 비동기, ESM — 은 각각 다른 신뢰성 모델을 가진다. 동기는 단순하지만 호출자가 모든 에러 처리를 책임진다. 비동기는 내구성이 높지만 멱등성 구현이 필수다. ESM은 관리된 폴링으로 편리하지만 Kinesis의 샤드 블록, SQS의 가시성 타임아웃 같은 메커니즘을 이해해야 한다. 이 차이를 명확히 구분하면 "SQS 메시지가 왜 반복 처리되지?", "Kinesis 레코드가 왜 쌓이지?" 같은 운영 질문에 즉시 답할 수 있다.

다음 글에서는 Lambda 배포 라이프사이클 — 환경 변수 관리, 레이어, 버전과 별칭, CodeDeploy 트래픽 시프트 — 를 다룬다.