도메인 이벤트 기반 색인 파이프라인 Kafka 도입 #159
Conversation
- `spring-kafka` 추가하여 Kafka 기반 메시징 구현 가능 - 테스트를 위해 `spring-kafka-test` 의존성 추가 - 메시징 기능 확장 및 관련 테스트 작성 환경 준비
- Docker Compose 구성에 Kafka 및 Zookeeper 서비스 추가하여 메시징 기능 지원 - Kafka 설정에 내부 및 외부 리스너 설정(KAFKA_LISTENERS) 추가로 유연성 강화 - Kafka와 Zookeeper 서비스 간 healthcheck 및 종속성 설정으로 안정적 시작 보장 - 기존 서비스 구성 개선: - `restart: unless-stopped`를 설정하여 컨테이너 안정성 확보 - MySQL 및 Elasticsearch healthcheck 설정 최적화 - 필요하지 않은 주석 처리된 MySQL 테스트 서비스 제거로 구성 단순화
- `PostUpdated`와 `PostCreated` 이벤트 클래스에 `Serializable` 인터페이스 추가하여 이벤트 직렬화 가능 - 애플리케이션 계층에서 도메인 이벤트를 게시할 수 있도록 `PostEventPublisher` 인터페이스 정의 - 이벤트 직렬화 및 게시 기능 추가로 메시징 시스템 연동 및 확장성 강화 - 코드의 유지보수성을 고려하여 인터페이스 기반 설계를 채택
- PostEventProducer 클래스 추가하여 Kafka를 통해 도메인 이벤트 게시를 지원 - Spring Kafka를 사용해 POST 이벤트를 지정된 토픽(post-events)으로 발행 - 발행 성공 및 실패 로깅을 포함해 이벤트 전달 과정의 투명성 강화 - 메시징 시스템과의 통합으로 시스템 확장성 및 비동기 처리 효율성 향상
- KafkaProducerConfig 클래스 추가하여 Kafka 프로듀서 설정 정의 - Bootstrap 서버, Key/Value 직렬화 설정 및 JsonSerializer 옵션 적용 - KafkaTemplate 빈 등록으로 프로듀서 사용 간소화 - Kafka 기반 메시징 시스템 연동 준비 작업 완료
- `PostEventConsumer` 클래스 추가로 Kafka의 post-events 토픽 이벤트 처리 기능 구현 - `KafkaConsumerConfig` 클래스 추가하여 Kafka 소비자 설정 정의: - 기본 서버, 그룹 ID, 직렬화/역직렬화 옵션 설정 - ConsumerFactory와 KafkaListenerContainerFactory 빈 등록으로 간소화된 소비자 관리 - `PostSearchIndexer`와 통합하여 Post 생성 이벤트(PostCreated)를 Elasticsearch에 인덱싱 - 미처리 가능한 이벤트 유형은 로그 경고로 처리 - Kafka 메시징의 소비 기능 추가로 이벤트 기반 비동기 처리 완성
- `PostEventMessage` 및 `PostEventType` 추가로 Kafka 메시지 모델 정의 - `PostEventProcessor` 구현으로 Kafka 메시지 및 도메인 이벤트 처리: - Post 생성/업데이트 이벤트 메시지로 Elasticsearch 색인 생성/갱신 - Post 삭제 이벤트 메시지로 Elasticsearch 색인 제거 - 미처리 가능한 이벤트 및 메시지 타입을 로그로 기록하여 문제 식별 용이성 확보 - Kafka 통합의 이벤트 처리 역량 확장 및 비동기 처리 완성도 개선
…lity - KafkaProducerConfig 및 KafkaConsumerConfig에 @ConditionalOnProperty 추가로 기능 비활성화 가능 - KafkaProperties를 활용한 설정 간소화 및 대체 설정 지원 - PostEventMessage 기반 프로듀서/소비자 설정으로 메시징 체계 정교화 - JsonSerializer 설정 개선, 타입 정보 유지 및 신뢰된 패키지 설정 추가 - PostEventConsumer 리팩토링으로 PostEventProcessor와 메시지 처리 일원화 - 이벤트 타입별 처리 로직 단순화 및 테스트 용이성 확보 - 설정 및 처리 로직의 유연성 향상으로 Kafka 통합의 효율성과 확장성 증대
- Post 이벤트 발행 시 트랜잭션 동기화를 지원하여 커밋 후 이벤트를 발행 - @ConditionalOnProperty 추가로 Kafka 통합 비활성화 가능 - 이벤트 메시지로 `PostEventMessage`를 사용하며 PostCreated, PostUpdated, PostDeleted 이벤트 변환 처리 - 토픽명을 외부 설정에서 가져오도록 개선, 기본값으로 `"post-events"` 설정 - 발행 성공/실패 시 디버그 및 에러 로깅 추가로 문제 분석 용이성 향상 - Kafka 메시징의 유연성 및 안정성을 향상시켜 메시지 손실 가능성 최소화
- 기존 Elasticsearch 연동 `PostSearchIndexer` 대신 `PostEventPublisher`로 이벤트 처리 방식 변경 - 게시글 생성, 업데이트, 삭제 시 Post 도메인 이벤트를 외부 파이프라인으로 위임하도록 수정 - 중복 코드 제거를 위해 `publish` 메서드로 공통 로직 통합 - 도메인 이벤트 기반 아키텍처로 확장성 및 유지보수성 개선
- see.kafka.enabled=false 시 도메인 이벤트를 Kafka 대신 직접 처리하는 DirectPostEventPublisher 추가 - @ConditionalOnProperty를 사용해 설정 기반으로 컴포넌트 활성화 제어 - Kafka 비활성화 환경에서도 안정적으로 이벤트를 처리할 수 있도록 구현 - PostEventProcessor를 활용해 도메인 이벤트를 동기적으로 처리하며 디버그 로깅 추가
- application-test.yml 축소 및 불필요한 설정 제거로 단순화 - hibernate.ddl-auto를 `update`로 변경하여 테스트 환경 자동화 - Hibernate 관련 세부 설정 제거 - application.properties 파일에 `spring.profiles.active=test` 추가 - Kafka 설정 플래그(`see.kafka.enabled`) 제거로 간단한 테스트 구성 지원 - 설정 간소화와 효율성 개선으로 테스트 유지보수성 증가
- Kafka 활성화 및 비활성화 여부에 따라 이벤트 처리 방식 검증을 위한 통합 테스트 추가 - PostEventDirectPipelineTest: Kafka 비활성화 시 도메인 이벤트가 바로 처리되는 시나리오 테스트 - PostEventKafkaPipelineTest: Kafka 활성화 환경에서 이벤트가 적절히 색인되고 삭제되는 것을 검증 - 테스트 환경에서 재사용 가능한 RecordingPostSearchIndexer로 이벤트 처리 기록 및 검증 지원 - 도메인 이벤트 처리 방식 변화에 대한 신뢰성 및 안정성 확보
- PostEventHandlerTest 제거. 테스트에서 사용된 색인 로직은 Post 도메인 이벤트와 Kafka 처리로 대체 - 기존 테스트가 새로운 이벤트 처리 및 색인 파이프라인과 더 이상 호환되지 않음 - 유지보수성 향상을 위해 불필요한 테스트 코드 삭제
- Kafka와 Zookeeper를 CI 환경에 추가하여 테스트 시 메시징 연동 검증 가능 - 서비스 의존성 대기 로직 수정으로 Elasticsearch, Kafka, Zookeeper 준비 상태 확인 - SPRING_KAFKA_BOOTSTRAP_SERVERS 환경 변수 추가로 테스트 환경 설정 개선 - CI 파이프라인 확장으로 메시징 관련 통합 테스트 신뢰성 증대
WalkthroughKafka 기반 도메인 이벤트 파이프라인을 도입하고, see.kafka.enabled에 따라 직접 처리 또는 Kafka 발행/소비로 분기하도록 애플리케이션, 테스트, CI 및 Docker 구성을 변경했습니다. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant App as 애플리케이션 (PostManager)
participant Handler as PostEventHandler
participant Publisher as PostEventPublisher (추상)
participant Producer as PostEventProducer
participant Direct as DirectPostEventPublisher
participant Kafka as Kafka 브로커
participant Consumer as PostEventConsumer
participant Processor as PostEventProcessor
participant Indexer as PostSearchIndexer
App->>Handler: 포스트 생성/수정/삭제 이벤트 발생
Handler->>Publisher: publish(DomainEvent)
alt Kafka 활성 (see.kafka.enabled=true)
Publisher->>Producer: toMessage & send (afterCommit 또는 즉시)
Producer->>Kafka: post-events 토픽에 전송
Kafka->>Consumer: PostEventMessage 전달
Consumer->>Processor: process(PostEventMessage)
else Kafka 비활성 (see.kafka.enabled=false)
Publisher->>Direct: publish(DomainEvent)
Direct->>Processor: process(DomainEvent)
end
Processor->>Indexer: index(post) / delete(postId)
sequenceDiagram
autonumber
participant Producer as PostEventProducer
participant Tx as Transaction
participant Kafka as Kafka 브로커
Producer->>Tx: 현재 트랜잭션 존재 여부 확인
alt 트랜잭션 활성
Producer->>Tx: afterCommit에 전송 작업 등록
Tx->>Kafka: 커밋 후 전송 실행
else 트랜잭션 비활성
Producer->>Kafka: 즉시 전송
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related issues
Possibly related PRs
Poem
Pre-merge checks and finishing touches❌ Failed checks (2 warnings)
✅ Passed checks (1 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
- 각 CI 작업 단계에 적절한 이모지를 추가하여 가독성을 개선 - 서비스 준비 확인 로직 강화: 개별 서비스 상태(Zookeeper, Kafka, Elasticsearch) 확인으로 디버깅 용이성 향상 - 메시지 출력 개선: 서비스 준비 상태 및 재시도 상태를 직관적으로 전달 - 테스트 신뢰성과 개발 경험 개선을 위해 워크플로우 세부사항을 다듬음
- 새로운 CI 워크플로우 추가로 PR 생성 시 코드 컴파일 체크 수행 - Java 21 및 Gradle 설정 포함하여 최신 환경에서의 테스트 및 빌드를 지원 - secret 기반 application.yml 생성으로 안전한 설정 관리 구현 - 병렬 및 캐시 사용 설정으로 CI 속도 최적화 - 코드 품질 및 변경 사항 신뢰성 검증 강화
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/main/java/dooya/see/domain/post/event/PostUpdated.java (1)
6-19:PostDeleted이벤트에서Serializable미구현
PostCreated와PostUpdated는 모두Serializable을 구현하여 Kafka 직렬화를 지원하지만,PostDeleted는Serializable을 구현하지 않았습니다.일관된 직렬화 패턴을 유지하려면
PostDeleted도Serializable을 구현해야 합니다:public record PostDeleted( Long postId, Long memberId, PostStatus previousStatus ) implements DomainEvent, Serializable {}파일:
src/main/java/dooya/see/domain/post/event/PostDeleted.javasrc/main/java/dooya/see/application/post/PostEventHandler.java (1)
10-11: 트랜잭션 커밋 이후에만 이벤트를 퍼블리시하도록 전환 필요@eventlistener는 트랜잭션 롤백에도 이벤트가 발행될 수 있어 색인 불일치가 발생합니다. AFTER_COMMIT 단계의 @TransactionalEventListener로 바꿔 주세요.
권장 변경:
-import org.springframework.context.event.EventListener; +import org.springframework.transaction.event.TransactionalEventListener; +import org.springframework.transaction.event.TransactionPhase; @@ - @EventListener + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) public void handlePostCreated(PostCreated event) { publish(event); } @@ - @EventListener + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) public void handlePostUpdated(PostUpdated event) { publish(event); } @@ - @EventListener + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) public void handlePostDeleted(PostDeleted event) { publish(event); }Also applies to: 19-32
♻️ Duplicate comments (1)
src/main/java/dooya/see/domain/post/event/PostUpdated.java (1)
6-19: Serializable 구현 시 의존 객체 직렬화 검증 필요
PostUpdated도PostCreated와 동일하게Serializable을 구현했습니다.Post객체의 직렬화 가능 여부를 확인해야 합니다. 이는PostCreated에서 제기한 동일한 검증이 필요합니다.
PostCreated에 대한 검증 스크립트를 참조하여Post클래스와 관련 타입의 직렬화 구현을 확인해주세요.
🧹 Nitpick comments (10)
docker-compose.yml (2)
1-1: Docker Compose 버전 명시 확인Docker Compose 버전
3.8이 명시되었습니다. 이는 레거시 형식이며, Docker Compose V2에서는 버전 필드가 더 이상 필요하지 않습니다. 최신 Docker Compose를 사용하는 경우 이 필드를 제거할 수 있습니다.
55-68: ZooKeeper 의존성 제거 고려현재 구성은 ZooKeeper를 사용하는 전통적인 Kafka 설정입니다. Apache Kafka 3.3.0 이상에서는 KRaft 모드(ZooKeeper 없는 운영)가 프로덕션 지원되며, Kafka 4.0부터는 KRaft가 기본 모드입니다.
개발 환경에서 KRaft 모드로 전환하면 다음과 같은 이점이 있습니다:
- 더 간단한 인프라 (ZooKeeper 제거)
- 더 빠른 시작 시간
- 향후 프로덕션 환경과의 일관성
KRaft 모드 사용을 고려해보세요:
kafka: image: confluentinc/cp-kafka:7.6.1 container_name: see-kafka restart: unless-stopped ports: - "9092:9092" environment: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093 KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:29092,PLAINTEXT_HOST://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://see-kafka:29092,PLAINTEXT_HOST://localhost:9092 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_LOG_RETENTION_HOURS: 168 CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qksrc/main/java/dooya/see/domain/post/event/PostCreated.java (1)
7-15: serialVersionUID 명시 고려
Serializable을 구현할 때 명시적인serialVersionUID를 선언하는 것이 권장됩니다. 이를 통해 클래스 버전 관리와 역직렬화 호환성을 제어할 수 있습니다.레코드는 직접 필드를 선언할 수 없으므로, 필요한 경우 일반 클래스로 변환하거나 Kafka 직렬화를 JSON 기반으로 전환하는 것을 고려해보세요. Spring Kafka에서는
JsonSerializer/JsonDeserializer를 사용하여 Java 직렬화의 제약 없이 안전하게 메시지를 전송할 수 있습니다.src/main/java/dooya/see/adapter/integration/kafka/PostEventMessage.java (1)
6-16: 팩토리 메서드 public 노출 제안팩토리들이 package-private이라 외부 패키지에서 재사용이 어렵습니다. 퍼블리셔/프로듀서/테스트에서의 사용성을 위해 public을 권장합니다.
- static PostEventMessage created(Long postId) { + public static PostEventMessage created(Long postId) { return new PostEventMessage(PostEventType.CREATED, postId); } - static PostEventMessage updated(Long postId) { + public static PostEventMessage updated(Long postId) { return new PostEventMessage(PostEventType.UPDATED, postId); } - static PostEventMessage deleted(Long postId) { + public static PostEventMessage deleted(Long postId) { return new PostEventMessage(PostEventType.DELETED, postId); }src/test/java/dooya/see/application/post/PostEventKafkaPipelineTest.java (1)
50-53: Awaitility 타임아웃 완화 제안CI 환경에서 5초는 간헐적으로 부족할 수 있습니다. 10–15초로 늘려 플래키를 줄이세요.
-Awaitility.await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> +Awaitility.await().atMost(Duration.ofSeconds(15)).untilAsserted(() ->Also applies to: 58-61, 73-77, 90-92
.github/workflows/ci.yml (1)
59-63: Kafka 컨테이너 healthcheck를 보다 견고하게cp-kafka 이미지에 nc가 없을 수 있습니다. Confluent 유틸리티(cub)를 사용한 헬스체크가 더 안정적입니다.
- options: >- - --health-cmd "nc -z localhost 9092" - --health-interval 10s - --health-timeout 5s - --health-retries 10 + options: >- + --health-cmd "bash -c 'cub kafka-ready -b localhost:9092 1 20'" + --health-interval 10s + --health-timeout 5s + --health-retries 12src/test/java/dooya/see/application/post/PostEventDirectPipelineTest.java (1)
77-103: 코드 중복을 제거하세요.
RecordingPostSearchIndexer클래스가PostEventKafkaPipelineTest에도 동일하게 존재합니다. 테스트 유틸리티 클래스로 추출하여 중복을 제거하는 것이 좋습니다.다음과 같이 공통 테스트 픽스처로 추출하세요:
src/test/java/dooya/see/application/post/RecordingPostSearchIndexer.java파일을 생성:package dooya.see.application.post; import dooya.see.application.post.required.PostSearchIndexer; import dooya.see.domain.post.Post; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; public class RecordingPostSearchIndexer implements PostSearchIndexer { private final CopyOnWriteArrayList<Post> indexed = new CopyOnWriteArrayList<>(); private final CopyOnWriteArrayList<Long> deleted = new CopyOnWriteArrayList<>(); @Override public void index(Post post) { indexed.add(post); } @Override public void delete(Long postId) { deleted.add(postId); } public List<Post> indexedPosts() { return List.copyOf(indexed); } public List<Long> deletedPostIds() { return List.copyOf(deleted); } public void reset() { indexed.clear(); deleted.clear(); } }그런 다음 두 테스트 파일에서 내부 클래스를 제거하고 공통 클래스를 사용하세요.
src/main/java/dooya/see/adapter/integration/kafka/config/KafkaProducerConfig.java (1)
27-35: 프로듀서 신뢰성 설정을 검토하세요.현재 구성은 기본 프로듀서 설정을 사용하고 있습니다. 색인 파이프라인의 중요도에 따라 다음 설정을 추가하는 것을 고려하세요:
acks=all: 모든 복제본 확인enable.idempotence=true: 중복 메시지 방지 (정확히 한 번 전송)retries: 재시도 횟수 설정Line 32의
JsonSerializer.ADD_TYPE_INFO_HEADERS는 단순한 record 타입인PostEventMessage에는 불필요할 수 있습니다.필요하다면 다음과 같이 추가 설정을 적용할 수 있습니다:
public ProducerFactory<String, PostEventMessage> postEventProducerFactory() { Map<String, Object> config = new HashMap<>(kafkaProperties.buildProducerProperties(null)); config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); - config.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, true); + config.put(ProducerConfig.ACKS_CONFIG, "all"); + config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); return new DefaultKafkaProducerFactory<>(config); }src/main/java/dooya/see/adapter/integration/kafka/config/KafkaConsumerConfig.java (2)
29-43: 컨슈머 팩토리 설정이 대체로 적절합니다.보안을 위해 신뢰할 수 있는 패키지를 제한한 점과
putIfAbsent를 사용하여 설정 재정의를 허용한 점이 좋습니다.선택적 개선사항: 프로듀서에서
ADD_TYPE_INFO_HEADERS=true를 설정했지만,PostEventMessage가 상속이 없는 단순 record이므로 타입 헤더가 불필요할 수 있습니다. 프로듀서 설정에서 제거를 고려하세요.
45-54: 동시성 설정을 확인하세요.현재
concurrency=1로 설정되어 있습니다. 이는 다음 사항을 고려해야 합니다:
순서 보장이 필요한 경우: 프로듀서에서
postId를 메시지 키로 사용하고, 동시성을 높여 처리량을 개선할 수 있습니다 (동일 키는 동일 파티션/스레드로 라우팅됨).순서 보장이 불필요한 경우: 동시성을 높여 처리량을 개선할 수 있습니다.
현재 설정은 보수적이지만 확장성에 제한이 있을 수 있습니다. 프로듀서의 메시지 키 추가와 함께 이 설정을 재검토하세요.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (20)
.github/workflows/ci.yml(4 hunks)build.gradle.kts(2 hunks)docker-compose.yml(3 hunks)src/main/java/dooya/see/adapter/integration/kafka/DirectPostEventPublisher.java(1 hunks)src/main/java/dooya/see/adapter/integration/kafka/PostEventConsumer.java(1 hunks)src/main/java/dooya/see/adapter/integration/kafka/PostEventMessage.java(1 hunks)src/main/java/dooya/see/adapter/integration/kafka/PostEventProcessor.java(1 hunks)src/main/java/dooya/see/adapter/integration/kafka/PostEventProducer.java(1 hunks)src/main/java/dooya/see/adapter/integration/kafka/PostEventType.java(1 hunks)src/main/java/dooya/see/adapter/integration/kafka/config/KafkaConsumerConfig.java(1 hunks)src/main/java/dooya/see/adapter/integration/kafka/config/KafkaProducerConfig.java(1 hunks)src/main/java/dooya/see/application/post/PostEventHandler.java(2 hunks)src/main/java/dooya/see/application/post/required/PostEventPublisher.java(1 hunks)src/main/java/dooya/see/domain/post/event/PostCreated.java(1 hunks)src/main/java/dooya/see/domain/post/event/PostUpdated.java(2 hunks)src/test/java/dooya/see/application/post/PostEventDirectPipelineTest.java(1 hunks)src/test/java/dooya/see/application/post/PostEventHandlerTest.java(0 hunks)src/test/java/dooya/see/application/post/PostEventKafkaPipelineTest.java(1 hunks)src/test/resources/application-test.yml(1 hunks)src/test/resources/application.properties(1 hunks)
💤 Files with no reviewable changes (1)
- src/test/java/dooya/see/application/post/PostEventHandlerTest.java
🧰 Additional context used
🧬 Code graph analysis (3)
src/test/java/dooya/see/application/post/PostEventKafkaPipelineTest.java (2)
src/test/java/dooya/see/domain/post/PostFixture.java (1)
PostFixture(9-110)src/test/java/dooya/see/SeeTestConfiguration.java (1)
TestConfiguration(11-22)
src/main/java/dooya/see/adapter/integration/kafka/config/KafkaConsumerConfig.java (1)
src/main/java/dooya/see/adapter/integration/kafka/config/KafkaProducerConfig.java (1)
Configuration(18-41)
src/test/java/dooya/see/application/post/PostEventDirectPipelineTest.java (3)
src/test/java/dooya/see/application/post/PostEventKafkaPipelineTest.java (2)
SpringBootTest(25-134)RecordingPostSearchIndexer(98-124)src/test/java/dooya/see/domain/post/PostFixture.java (1)
PostFixture(9-110)src/test/java/dooya/see/SeeTestConfiguration.java (1)
TestConfiguration(11-22)
🪛 actionlint (1.7.8)
.github/workflows/ci.yml
64-64: unexpected key "depends_on" for "services" section. expected one of "credentials", "env", "image", "options", "ports", "volumes"
(syntax-check)
🔇 Additional comments (22)
src/test/resources/application.properties (1)
1-1: 테스트 프로파일 활성화 확인테스트 환경에서
test프로파일을 활성화하는 설정이 올바르게 추가되었습니다.application-test.yml과 함께 작동하여 테스트에서 Kafka를 비활성화하는 구성이 적용됩니다.src/test/resources/application-test.yml (1)
10-12: Kafka 비활성화 설정 확인테스트 환경에서 Kafka를 기본적으로 비활성화하는 설정이 올바르게 추가되었습니다. 이는
DirectPostEventPublisher를 사용하는 테스트 시나리오와 일치합니다.build.gradle.kts (2)
51-51: Kafka 의존성 추가 확인Spring Kafka 의존성이 올바르게 추가되었습니다. Spring Boot 3.5.4의 BOM을 통해 버전이 관리되어 호환성이 보장됩니다.
66-66: Kafka 테스트 의존성 추가 확인Spring Kafka 테스트 의존성이 올바르게 추가되었습니다. 이는 Kafka 통합 테스트(예:
PostEventKafkaPipelineTest)를 지원합니다.docker-compose.yml (2)
70-98: Kafka 리스너 구성 확인Kafka 리스너 구성이 올바르게 설정되어 있습니다:
PLAINTEXT://see-kafka:29092: 컨테이너 간 통신PLAINTEXT_HOST://localhost:9092: 호스트에서의 접근단일 브로커 환경에 적합한 복제 계수(1)와 적절한 로그 보관 기간(168시간)이 설정되었습니다.
20-23: MySQL 헬스체크 개선 확인MySQL 헬스체크에
-proot플래그와 명확한 interval/timeout이 추가되었습니다. 이는 서비스 시작 순서를 보장하는 데 도움이 됩니다.src/main/java/dooya/see/adapter/integration/kafka/PostEventType.java (1)
1-7: 이벤트 타입 enum 정의 확인Post 이벤트 타입을 명확하게 정의하는 enum이 올바르게 구현되었습니다. CREATED, UPDATED, DELETED의 세 가지 타입이 도메인 이벤트와 일치합니다.
src/main/java/dooya/see/application/post/required/PostEventPublisher.java (2)
1-7: 이벤트 발행 인터페이스 추상화 확인
PostEventPublisher인터페이스가 깔끔하게 정의되었습니다. 이를 통해 Kafka 활성화 여부에 따라DirectPostEventPublisher또는PostEventProducer구현체를 선택적으로 사용할 수 있습니다.단일 메서드 인터페이스로 명확한 책임을 가지며, 좋은 추상화 설계입니다.
1-7: 조건부 빈 구성이 올바르게 적용되어 있음 - 재검토 불필요검증 결과,
PostEventPublisher구현체들의 조건부 활성화가 다음과 같이 정확하게 구성되어 있습니다:
DirectPostEventPublisher:@ConditionalOnProperty(value = "see.kafka.enabled", havingValue = "false", matchIfMissing = true)→see.kafka.enabled가 "false"이거나 누락되었을 때 활성화PostEventProducer:@ConditionalOnProperty(value = "see.kafka.enabled", havingValue = "true")→see.kafka.enabled가 "true"일 때만 활성화두 조건이 상호배타적이므로, 어떤 상황에서도 정확히 하나의 구현체만 활성화되어
NoUniqueBeanDefinitionException은 발생하지 않습니다.Likely an incorrect or invalid review comment.
src/test/java/dooya/see/application/post/PostEventDirectPipelineTest.java (3)
20-34: 테스트 설정이 적절합니다.Kafka 비활성화 모드에서 동기적 색인 처리를 테스트하는 구성이 올바르게 작성되었습니다.
36-71: 테스트 로직이 정확합니다.동기적 이벤트 처리에 적합하게 즉시 단언(assertion)을 수행하고 있으며, 생성/수정/삭제 시나리오를 올바르게 커버하고 있습니다.
105-112: 테스트 구성이 올바릅니다.
@Primary어노테이션을 사용하여 프로덕션 빈을 테스트 구현체로 적절히 대체하고 있습니다.src/main/java/dooya/see/adapter/integration/kafka/PostEventProducer.java (3)
17-26: 클래스 설정이 적절합니다.조건부 활성화와 Kafka 템플릿 주입이 올바르게 구성되었습니다.
@Value로 주입되는topic필드는 final로 선언할 수 없지만, 이는 Spring의 제약사항입니다.
28-47: 트랜잭션 인식 메시지 발행이 올바르게 구현되었습니다.활성 트랜잭션이 있을 때 커밋 이후로 Kafka 전송을 연기하여 데이터 일관성을 보장하는 접근 방식이 적절합니다.
49-60: 이벤트 변환 로직이 명확합니다.패턴 매칭을 사용한 이벤트 타입 변환이 깔끔하게 구현되었습니다.
src/main/java/dooya/see/adapter/integration/kafka/config/KafkaProducerConfig.java (1)
37-40: KafkaTemplate 빈 설정이 적절합니다.src/main/java/dooya/see/adapter/integration/kafka/PostEventProcessor.java (5)
14-20: 클래스 설계가 적절합니다.패키지 프라이빗 접근 제어자를 사용하여 내부 구현 상세를 캡슐화한 점이 좋습니다.
21-31: 도메인 이벤트 처리 로직이 명확합니다.패턴 매칭을 통한 이벤트 타입 분기가 깔끔하게 구현되었습니다.
33-47: Kafka 메시지 처리 로직이 올바릅니다.
CREATED와UPDATED이벤트에서 최신 스냅샷을 조회하는 접근은 메시지 크기를 줄이고 결합도를 낮추는 좋은 설계 결정입니다. 추가 DB 조회가 발생하지만, 비동기 처리 맥락에서 허용 가능한 트레이드오프입니다.
49-71: 헬퍼 메서드의 오류 처리가 적절합니다.게시글을 찾을 수 없는 경우 경고 로그를 남기고, 색인 실패 시 오류를 기록하는 방식이 올바릅니다. 예외를 다시 던지지 않아 컨슈머가 중단되지 않도록 한 점이 좋습니다.
73-98: 도메인 이벤트 핸들러가 올바르게 구현되었습니다.직접 처리 모드에서 이벤트가 Post 엔티티를 포함하고 있어 추가 DB 조회 없이 효율적으로 색인할 수 있습니다.
src/main/java/dooya/see/adapter/integration/kafka/DirectPostEventPublisher.java (1)
10-22: 직접 발행자 구현이 적절합니다.
matchIfMissing=true설정으로 Kafka 속성이 없을 때 기본적으로 직접 처리 모드로 동작하여 하위 호환성을 보장하는 점이 좋습니다.
| @KafkaListener( | ||
| topics = "${see.kafka.topics.post-events:post-events}", | ||
| groupId = "${spring.kafka.consumer.group-id:post-indexer-group}" | ||
| ) | ||
| public void consume(PostEventMessage message) { | ||
| log.debug("Kafka 이벤트 수신: {}", message); | ||
| postEventProcessor.process(message); | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
🧩 Analysis chain
소비자 오류 처리와 역직렬화 설정을 명시하세요
현재 예외 시 처리 전략(DLT/재시도/백오프)이 보이지 않습니다. 또한 POJO 페이로드를 직접 받으려면 JsonDeserializer의 기본 타입/Trusted Packages 설정이 필요합니다. 전용 컨테이너 팩토리와 CommonErrorHandler를 사용하도록 리스너에 명시해 주세요.
예시(리스너에 컨테이너 팩토리 명시):
- @KafkaListener(
+ @KafkaListener(
topics = "${see.kafka.topics.post-events:post-events}",
- groupId = "${spring.kafka.consumer.group-id:post-indexer-group}"
- )
+ groupId = "${spring.kafka.consumer.group-id:post-indexer-group}",
+ containerFactory = "postEventKafkaListenerContainerFactory"
+ )
public void consume(PostEventMessage message) {다음 스캔으로 설정 유무를 확인해 주세요:
Based on learnings
🏁 Script executed:
#!/usr/bin/env bash
# JsonDeserializer/Trusted Packages/에러핸들러/컨테이너 팩토리 존재 여부 점검
rg -nP -C2 '(JsonDeserializer|CommonErrorHandler|DefaultErrorHandler|ConcurrentKafkaListenerContainerFactory|postEventKafkaListenerContainerFactory|spring\.kafka\.consumer\.properties\.spring\.json\.trusted\.packages|JsonDeserializer\.VALUE_DEFAULT_TYPE)'Length of output: 3779
---
🏁 Script executed:
```shell
#!/bin/bash
# Get full KafkaConsumerConfig.java to see error handler setup
cat src/main/java/dooya/see/adapter/integration/kafka/config/KafkaConsumerConfig.java
Length of output: 2622
🏁 Script executed:
#!/bin/bash
# Check if @KafkaListener in PostEventConsumer specifies containerFactory
rg -nP -A5 '@KafkaListener' src/main/java/dooya/see/adapter/integration/kafka/PostEventConsumer.javaLength of output: 288
---
🏁 Script executed:
```shell
#!/bin/bash
# Verify if error handler is configured anywhere for Kafka
rg -nP '(setCommonErrorHandler|setErrorHandler|DefaultErrorHandler|CommonErrorHandler)' src/main/java/dooya/see/adapter/integration/kafka/
Length of output: 0
@KafkaListener에 containerFactory를 지정하고 에러 핸들러를 설정하세요
현재 상태 검증 결과:
- ✓ JsonDeserializer 및 신뢰 패키지(trusted packages)는 이미 올바르게 구성됨 (KafkaConsumerConfig.java 라인 37-40)
- ✗ @KafkaListener가 containerFactory 파라미터를 지정하지 않음 (PostEventConsumer.java 라인 16-19)
- ✗ CommonErrorHandler 또는 DefaultErrorHandler가 설정되지 않음
필수 수정 사항:
- PostEventConsumer.java 라인 16:
containerFactory = "kafkaListenerContainerFactory"파라미터 추가 - KafkaConsumerConfig.java 라인 52:
kafkaListenerContainerFactory()메서드에 에러 핸들러 추가 (예:factory.setCommonErrorHandler(new DefaultErrorHandler())또는 사용자 정의 에러 핸들러 구현)
🤖 Prompt for AI Agents
In src/main/java/dooya/see/adapter/integration/kafka/PostEventConsumer.java
around lines 16 to 23, add the containerFactory parameter to the @KafkaListener
annotation (containerFactory = "kafkaListenerContainerFactory") so the listener
uses the configured factory; and in
src/main/java/dooya/see/config/KafkaConsumerConfig.java around line 52, update
the kafkaListenerContainerFactory() method to attach an error handler by calling
factory.setCommonErrorHandler(...) with a DefaultErrorHandler or your custom
CommonErrorHandler instance (ensure any required imports and bean visibility are
in place).
src/main/java/dooya/see/adapter/integration/kafka/PostEventProducer.java
Outdated
Show resolved
Hide resolved
| import java.io.Serializable; | ||
|
|
||
| public record PostCreated( | ||
| Long postId, | ||
| Long memberId, | ||
| PostCategory category, | ||
| boolean publishImmediately, | ||
| Post post | ||
| ) implements DomainEvent { | ||
| ) implements DomainEvent, Serializable { |
There was a problem hiding this comment.
🧩 Analysis chain
Serializable 구현 시 의존 객체 직렬화 검증 필요
PostCreated가 Serializable을 구현했지만, 레코드 내의 Post 객체와 PostCategory 객체도 직렬화 가능해야 합니다. 이들 타입이 Serializable을 구현하지 않으면 런타임에 NotSerializableException이 발생합니다.
다음 스크립트로 Post와 PostCategory 클래스의 직렬화 구현 여부를 확인해주세요:
🏁 Script executed:
#!/bin/bash
# Description: Post와 PostCategory의 Serializable 구현 확인
echo "=== Post class Serializable check ==="
rg -n "class Post" --type java -A5 | rg -e "implements.*Serializable" -e "class Post"
echo -e "\n=== PostCategory class/enum Serializable check ==="
rg -n "(class|enum) PostCategory" --type java -A5 | rg -e "implements.*Serializable" -e "(class|enum) PostCategory"
echo -e "\n=== Post 내부 필드 타입 확인 ==="
ast-grep --pattern $'class Post {
$$$
}'Length of output: 3747
🏁 Script executed:
#!/bin/bash
# Post.java 파일 직접 조회
echo "=== Post.java 클래스 선언부 확인 ==="
cat src/main/java/dooya/see/domain/post/Post.java | head -40
echo -e "\n=== PostCategory.java 파일 확인 ==="
cat src/main/java/dooya/see/domain/post/PostCategory.java
echo -e "\n=== AbstractAggregateRoot 위치 및 Serializable 구현 확인 ==="
find . -name "AbstractAggregateRoot.java" -type f | head -1 | xargs cat | head -30Length of output: 2079
🏁 Script executed:
#!/bin/bash
# AbstractEntity 확인
echo "=== AbstractEntity 파일 확인 ==="
find . -name "AbstractEntity.java" -type f | head -1 | xargs cat | head -30
echo -e "\n=== PostContent 클래스 확인 ==="
rg -n "class PostContent|record PostContent" --type java -A3 | head -20
echo -e "\n=== PostStatus enum 확인 ==="
rg -n "enum PostStatus" --type java -A8 | head -20
echo -e "\n=== PostMetaData 클래스 확인 ==="
rg -n "class PostMetaData|record PostMetaData" --type java -A3 | head -20
echo -e "\n=== Tag 클래스 확인 ==="
rg -n "class Tag|record Tag" --type java -B2 -A5 src/main/java/dooya/see/domain/post/ | head -30Length of output: 3592
🏁 Script executed:
#!/bin/bash
# PostCreated 레코드의 serialVersionUID 확인 및 직렬화 가능 여부 최종 확인
echo "=== PostCreated.java 전체 파일 ==="
cat src/main/java/dooya/see/domain/post/event/PostCreated.java
echo -e "\n=== Post 필드의 직렬화 가능 여부 종합 확인 ==="
echo "Post 클래스 선언부:"
rg -n "public class Post" --type java -A2 src/main/java/dooya/see/domain/post/Post.java
echo -e "\nPostContent record 전체:"
cat src/main/java/dooya/see/domain/post/PostContent.java
echo -e "\nPostMetaData record 전체:"
cat src/main/java/dooya/see/domain/post/PostMetaData.javaLength of output: 3214
Post 및 의존 객체들의 Serializable 구현 필수
PostCreated가 Serializable을 구현했지만, 레코드에 포함된 Post 객체와 그 내부 value objects(PostContent, PostMetaData, Tag)가 Serializable을 구현하지 않아 런타임에 NotSerializableException이 발생합니다.
Post:AbstractAggregateRoot상속,Serializable미구현PostContent: record,Serializable미구현PostMetaData: record,Serializable미구현Tag: record,Serializable미구현
PostCategory는 Java enum이므로 기본적으로 Serializable을 지원합니다.
src/test/java/dooya/see/application/post/PostEventKafkaPipelineTest.java
Show resolved
Hide resolved
| order_updates: true | ||
| order_inserts: true | ||
| show-sql: true # SQL 출력 원하면 추가 | ||
| ddl-auto: update |
There was a problem hiding this comment.
🧩 Analysis chain
ddl-auto를 update로 변경한 이유 확인 필요
테스트 환경에서 ddl-auto가 update로 변경되었습니다. 이는 테스트 실행 시마다 스키마를 자동으로 업데이트하는 설정입니다. 일반적으로 테스트에서는 create-drop을 사용하여 테스트 간 격리를 보장하거나, validate를 사용하여 스키마 일관성을 검증하는 것이 권장됩니다.
update 모드는 스키마 변경사항을 누적시킬 수 있어 테스트 간 간섭이나 예기치 않은 상태를 유발할 수 있습니다.
테스트 환경에서 H2 인메모리 데이터베이스를 사용 중인지, 그리고 create-drop이 더 적합하지 않은지 확인해주세요:
🏁 Script executed:
#!/bin/bash
# Description: 테스트에서 H2 사용 여부 확인 및 JPA 설정 검토
# H2 의존성 확인
echo "=== H2 dependency check ==="
rg -n "h2database" build.gradle.kts
# 테스트에서 DataSource 설정 확인
echo -e "\n=== Test DataSource configuration ==="
fd -e yml -e yaml -e properties . src/test/resources --exec cat {}
# 테스트 클래스에서 @DataJpaTest 또는 @SpringBootTest 사용 확인
echo -e "\n=== Test annotations for database tests ==="
rg -n "@(DataJpaTest|SpringBootTest)" src/test/java --type java -A2Length of output: 8437
추가 검증이 필요합니다. 메인 애플리케이션 설정과 테스트 격리 전략을 확인하겠습니다.
---
🏁 Script executed:
```shell
#!/bin/bash
# Main application configuration check and test database isolation analysis
echo "=== Main application configuration ==="
fd -e yml -e yaml . src/main/resources --exec cat {}
echo -e "\n=== test configuration files ==="
fd -e yml -e yaml . src/test/resources
echo -e "\n=== @DataJpaTest with custom configuration ==="
rg -n "@AutoConfigureTestDatabase" src/test/java --type java -B2 -A3
echo -e "\n=== Check if schema issues are documented ==="
rg -i "schema|ddl|update|create-drop" build.gradle.kts pom.xml 2>/dev/null | head -20
Length of output: 1113
최종 정보를 수집하기 위해 변경 이력과 주석을 확인하겠습니다.
---
🏁 Script executed:
```shell
#!/bin/bash
# Check for comments or documentation about the ddl-auto choice
echo "=== Search for comments around ddl-auto configuration ==="
rg -n "ddl-auto" . --type yml -B3 -A3
echo -e "\n=== Check git history for application-test.yml changes ==="
git log --oneline -n 20 src/test/resources/application-test.yml 2>/dev/null || echo "Git history not available"
echo -e "\n=== Look for TODO or FIXME related to schema/ddl ==="
rg -i "TODO|FIXME.*schema|TODO.*ddl" src/test/resources
echo -e "\n=== Verify if all SpringBootTest use @Transactional ==="
rg -n "@SpringBootTest" src/test/java --type java -A5 | grep -E "(SpringBootTest|Transactional)" | head -30
Length of output: 3318
테스트 환경의 ddl-auto: update 설정 수정 필요 — 트랜잭션 보호가 없는 테스트들이 스키마 누적 위험에 노출됨
검증 결과, 여러 @SpringBootTest 클래스가 @Transactional 없이 데이터베이스에 접근합니다:
SeeApplicationTestPostEventKafkaPipelineTestJwtTokenManagerTestPostEventDirectPipelineTestPostSearchElasticsearchRepositoryTestPostSearchElasticsearchReaderAdapterTestPostSearchElasticsearchAdapterTest
이들은 트랜잭션 롤백 보호 없이 update 모드에서 스키마 변경을 누적시킬 수 있습니다.
권장 해결책:
ddl-auto를create-drop으로 변경하여 테스트 간 격리 강화, 또는- 데이터베이스 접근이 있는 모든
@SpringBootTest에@Transactional추가
(PostRepositoryPerformanceTest의 @AutoConfigureTestDatabase(replace = NONE)는 의도적 성능 테스트 설정이므로 유지)
🤖 Prompt for AI Agents
In src/test/resources/application-test.yml around line 4, the test profile uses
"ddl-auto: update" which can cause schema changes to accumulate because several
@SpringBootTest classes access the DB without @Transactional; change the setting
to "ddl-auto: create-drop" to ensure each test run starts with a fresh schema,
or alternatively add @Transactional to each of the listed test classes
(SeeApplicationTest, PostEventKafkaPipelineTest, JwtTokenManagerTest,
PostEventDirectPipelineTest, PostSearchElasticsearchRepositoryTest,
PostSearchElasticsearchReaderAdapterTest, PostSearchElasticsearchAdapterTest) so
DB changes are rolled back after each test; keep PostRepositoryPerformanceTest's
@AutoConfigureTestDatabase(replace = NONE) unchanged as it is intentional.
- CI 파이프라인에서 Zookeeper를 제거하고 Kafka를 KRaft 모드로 전환 - KRaft 기반 설정 추가(KAFKA_PROCESS_ROLES, KAFKA_NODE_ID 등)로 Zookeeper 의존성 제거 - Kafka 서비스 구성 단순화 및 트랜잭션 관련 설정으로 테스트 신뢰성 강화 - 서비스 준비 대기 로직에서 Zookeeper 제거로 불필요한 대기 시간 감소 - 워크플로우 유지보수성을 개선하며 최신 Kafka 설정 적용
- CI 환경에서 Kafka 클러스터의 고유 식별자를 설정하도록 KAFKA_CLUSTER_ID 추가 - 고유 식별자를 통해 테스트 환경 간 충돌 가능성 방지 - Kafka 설정의 일관성을 유지하며 CI 안정성 개선
- Kafka 시작 명령어에 조건부 `kafka-storage format` 추가로 재포맷 방지 - CLUSTER_ID로 환경 변수 이름 변경하여 설정 명확성 향상 - 헬스체크 옵션 수정: 인터벌을 15초, 타임아웃을 10초로 증가시켜 안정성 개선 - CI 워크플로우에서 Kafka 구성의 신뢰성과 유지보수성을 강화
- Kafka 시작 스크립트를 `--entrypoint bash`로 수정하여 명령어 실행 방식 개선 - Kafka 시작 완료 후 'Kafka started successfully!' 메시지 추가로 진단 편의성 향상 - CI 실행 중 Kafka 구성 로직의 가독성과 유지보수성 강화
- Kafka 시작 스크립트의 `--entrypoint` 옵션을 JSON 배열 형식으로 변경하여 명령어 구문 개선 - 불필요한 로깅 메시지 및 bash 옵션 제거로 워크플로우 가독성 강화 - CI 환경의 명령어 실행 구조를 단순화하고 유지보수성을 향상시킴
- `--entrypoint` 옵션의 명령어를 단순화하여 CI 워크플로우 가독성 개선 - Kafka 시작 명령어를 다중 줄로 작성해 유지보수성 및 명령어 가독성 강화 - 불필요한 KAFKA_CLUSTER_ID 설정 제거로 환경 변수 구성 간소화
- KAFKA_CLUSTER_ID 환경 변수를 다시 추가하여 CI에서 클러스터 식별자를 명확히 설정 - 불필요한 `--entrypoint bash` 및 관련 명령어 제거로 워크플로우 단순화 - Kafka 시작 옵션 및 헬스체크 구성 가독성 강화 - 설정 정리를 통해 CI 환경 구성의 유지보수성 개선
- CLUSTER_ID 환경 변수를 Kafka 설정에 추가하여 CI에서 클러스터 식별자를 명확히 정의 - 설정 변경으로 CI 환경의 일관성 강화 및 유지보수성 개선
- CLUSTER_ID 및 KAFKA_CLUSTER_ID 환경 변수를 고유 값으로 업데이트하여 CI 환경에서의 명확성 강화 - 구성을 최신 상태로 유지하며 Kafka 클러스터 식별의 일관성 개선
- KafkaListenerContainerFactory에 `DefaultErrorHandler` 추가로 예외 처리 강화 - Kafka 리스너가 에러 발생 시 안정적 처리를 지원하도록 설정 수정 - `PostEventConsumer`에 컨테이너 팩토리 설정 적용으로 신뢰성 향상
- spring-retry 의존성 추가로 재시도 로직 구현 지원 - 특정 실패 시 복구 및 재시도 기능 강화 - 실패 내구성을 개선하고 신뢰성을 높이기 위한 사전 작업
- Kafka 프로듀서에 적용할 `RetryTemplate` 빈 추가 - 최대 시도 횟수 3회와 지수 백오프(초기 지연 200ms, 최대 2초) 설정 - 재시도 로직을 통해 메시지 전송 실패 시 안정성과 신뢰성 개선
- `RetryTemplate`을 활용한 Kafka 이벤트 발행 재시도 로직 추가 - 메시지 키 생성 및 재시도 실행 방식 수정으로 이벤트 전송 안정성 강화 - 재시도 초과 시 마지막 에러 로깅 및 예외 처리로 문제 식별성 개선
- `Tag`, `Post`, `PostContent`, and `PostMetaData`에 `Serializable` 구현하여 직렬화 가능하도록 수정 - 직렬화 지원을 위해 각 클래스에 `serialVersionUID` 필드 추가 - 엔티티 직렬화를 위한 사전 작업으로, 분산 환경에서의 데이터 전송 및 저장소 연계를 대비
- `spring.kafka.consumer.group-id`를 테스트 설정에 추가하여 Kafka 테스트 환경에서 유니크한 그룹 ID를 사용할 수 있도록 변경 - 테스트 병렬 실행 및 고유 컨슈머 그룹 간 충돌 방지를 위한 조치 - 랜덤 UUID 값을 환경 변수로 사용하여 매 실행마다 고유 그룹 ID 할당
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
src/main/java/dooya/see/adapter/integration/kafka/PostEventConsumer.java (1)
16-20: containerFactory 명시 확인 — listener id 추가와 에러 핸들러 설정 재확인 권장
- containerFactory 지정 완료, 좋습니다. 운영 가시성을 위해 listener id 추가를 권장합니다.
- 이전 코멘트에서 제안된 CommonErrorHandler/DefaultErrorHandler 설정은 컨테이너 팩토리(KafkaConsumerConfig)에서 반드시 붙여주세요(재시도/백오프/DLT 전략 포함). 이 부분은 기존 지적의 연장선입니다.
- @KafkaListener( + @KafkaListener( + id = "postEventConsumer", topics = "${see.kafka.topics.post-events:post-events}", groupId = "${spring.kafka.consumer.group-id:post-indexer-group}", containerFactory = "kafkaListenerContainerFactory" )
🧹 Nitpick comments (12)
src/main/java/dooya/see/domain/post/PostContent.java (1)
12-14: serialVersionUID에 @serial 추가 제안JDK 17+에서는 직렬화 관련 멤버를 @serial로 표기하는 것이 가독성과 도구 지원 측면에서 유리합니다. 기능 변화는 없고 선택 사항입니다.
+import java.io.Serial; ... ) implements java.io.Serializable { - private static final long serialVersionUID = 1L; + @Serial + private static final long serialVersionUID = 1L;src/main/java/dooya/see/domain/post/Post.java (1)
23-25: 직렬화 표기 보완(@serial) 및 이벤트 페이로드 범위 확인 요청
- serialVersionUID에 @serial 주석을 권장합니다.
- Post 자체가 Serializable이 되면서, 도메인 이벤트에 this(전체 Aggregate)를 포함한 상태로 Kafka에 실리지 않는지 확인 부탁드립니다. Kafka에는 최소 필드만 담은 DTO(PostEventMessage 등)로 매핑되어야 용량/민감정보/지연 측면에서 안전합니다. 그렇지 않다면 매핑 단에서 필드 축소가 필요합니다.
+import java.io.Serial; ... -public class Post extends AbstractAggregateRoot implements java.io.Serializable { +public class Post extends AbstractAggregateRoot implements java.io.Serializable { // ... - private static final long serialVersionUID = 1L; + @Serial + private static final long serialVersionUID = 1L;src/main/java/dooya/see/adapter/integration/kafka/PostEventConsumer.java (1)
22-24: 로그 페이로드 축소(민감정보/대용량 방지)message 전체를 찍기보다 키/타입/식별자 위주로 구조화 로그를 권장합니다(예: eventType, postId, version). 필요 시 샘플링 적용.
- log.debug("Kafka 이벤트 수신: {}", message); + log.debug("Kafka 이벤트 수신: type={}, postId={}, aggregateVersion={}", + message.eventType(), message.postId(), message.version());src/main/java/dooya/see/domain/post/Tag.java (1)
9-10: 직렬화 표기(@serial) 및 DB 제약 동기화(@column) 제안
- serialVersionUID에 @serial 주석을 권장합니다.
- 도메인 유효성(최대 20자, null 불가)에 맞춰 DB 제약도 일치시키면 데이터 무결성이 좋아집니다. ElementCollection에 매핑될 컬럼에 @column 적용을 권장합니다.
+import java.io.Serial; +import jakarta.persistence.Column; ... -@Embeddable -public record Tag(String name) implements Serializable { +@Embeddable +public record Tag(@Column(name = "name", length = 20, nullable = false) String name) implements Serializable { - private static final long serialVersionUID = 1L; + @Serial + private static final long serialVersionUID = 1L;src/main/java/dooya/see/domain/post/PostMetaData.java (1)
15-16: serialVersionUID에 @serial 추가 및 시간 필드 타입 검토(선택)
- @serial 주석 추가를 권장합니다.
- 선택: Kafka/외부 시스템과의 상호 운용성이 중요하다면 LocalDateTime 대신 UTC 기반 Instant 사용을 고려해 보세요(타임존 혼동 방지). 지금 변경 범위를 넘어선다면 추후 마이그레이션으로 충분합니다.
+import java.io.Serial; ... ) implements Serializable { - private static final long serialVersionUID = 1L; + @Serial + private static final long serialVersionUID = 1L;src/main/java/dooya/see/adapter/integration/kafka/PostEventProducer.java (4)
70-95: 동기 Future#get + 재시도는 커밋/처리 스레드를 블로킹합니다대기 시간이 커질 수 있습니다. 전용 Executor로 오프로드(비동기 전송 후 콜백에서 재시도/보고)하거나, 프로듀서 신뢰성(acks=all, idempotence 등)으로 전환을 검토하세요. 아래 설정 보강은 config에 반영하는 편이 낫습니다.
33-40: 무시되는 이벤트에 대한 관측성 보강 제안현재 DEBUG 로그만 있습니다. 향후 누락 타입 파악을 위해 WARN(저빈도) 또는 메트릭 카운터를 추가하는 것을 권장합니다.
Also applies to: 57-68
41-44: 키 null 경로 재검토(경미)toMessage에서 항상 postId를 채운다면 키가 null일 일이 없습니다. 가정이 맞다면 non-null을 전제(Objects.requireNonNull)하거나 null 분기 제거로 단순화 가능합니다.
10-16: 현재 상태: 빈 충돌 없음 — @qualifier 선택 사항현재 프로젝트에는 KafkaTemplate 빈 1개(
postEventKafkaTemplate), RetryTemplate 빈 1개(kafkaRetryTemplate)만 존재하므로 Spring이 명확히 의존성을 해석합니다. 따라서 @qualifier는 필수가 아니며, 향후 동일 타입의 빈이 추가될 경우 그때 추가하면 됩니다.src/main/java/dooya/see/adapter/integration/kafka/config/KafkaProducerConfig.java (3)
31-36: 프로듀서 신뢰성 설정 보강(acks/idempotence 등)브로커 레벨 보장을 강화하세요. 재시도 시 순서 보장이 중요하면 max.in.flight도 검토하세요.
Map<String, Object> config = new HashMap<>(kafkaProperties.buildProducerProperties(null)); config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); config.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, true); + // 신뢰성 강화 + config.put(ProducerConfig.ACKS_CONFIG, "all"); + config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); + // 순서 보장이 매우 중요하고 재시도가 발생할 수 있다면 주석 해제 고려 + // config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1); return new DefaultKafkaProducerFactory<>(config);
33-34: JsonSerializer 타입 헤더 사용 여부 확인소비자가 고정 타입(PostEventMessage)으로 역직렬화한다면 타입 헤더는 불필요합니다. 헤더 노출/오버헤드를 줄이려면 비활성화를 검토하세요. 단, 소비자가 헤더에 의존한다면 현재 값 유지.
- config.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, true); + // 소비자가 고정 타입이면 false 권장 + config.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
44-49: RetryTemplate 튜닝(비재시도 예외/지터/관측성)
- 재시도해도 의미 없는 예외(직렬화 오류, InvalidTopic 등)를 notRetryOn으로 분류하세요.
- 백오프에 지터를 추가해 스파이크를 완화하세요.
- 성공/실패 메트릭 발행을 고려하세요.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (10)
build.gradle.kts(2 hunks)src/main/java/dooya/see/adapter/integration/kafka/PostEventConsumer.java(1 hunks)src/main/java/dooya/see/adapter/integration/kafka/PostEventProducer.java(1 hunks)src/main/java/dooya/see/adapter/integration/kafka/config/KafkaConsumerConfig.java(1 hunks)src/main/java/dooya/see/adapter/integration/kafka/config/KafkaProducerConfig.java(1 hunks)src/main/java/dooya/see/domain/post/Post.java(1 hunks)src/main/java/dooya/see/domain/post/PostContent.java(1 hunks)src/main/java/dooya/see/domain/post/PostMetaData.java(2 hunks)src/main/java/dooya/see/domain/post/Tag.java(1 hunks)src/test/java/dooya/see/application/post/PostEventKafkaPipelineTest.java(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (2)
- src/main/java/dooya/see/adapter/integration/kafka/config/KafkaConsumerConfig.java
- src/test/java/dooya/see/application/post/PostEventKafkaPipelineTest.java
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: test
🔇 Additional comments (2)
build.gradle.kts (2)
51-52: Spring Boot BOM 관리 버전 확인 완료 — 명시적 버전 지정은 선택 사항.Spring Boot 3.4는 Spring Kafka 3.3으로 관리되며, Spring Boot 3.5.4도 동일한 Spring Kafka 3.3.x 라인을 사용합니다. 현재 코드는 BOM을 통해 호환 버전을 올바르게 받지만, 명시적 버전 지정은 빌드 재현성을 향상시키는 모범 사례입니다.
추천 버전 (선택):
spring-kafka:3.3.10spring-retry:2.0.12spring-kafka-test:3.3.10현재 코드는 작동하며, 명시적 버전 지정은 필수가 아닙니다.
51-52: 원본 리뷰 의견은 부정확합니다. 현재 구성이 올바릅니다.Spring Boot 애플리케이션에서 Spring Kafka를 사용할 때, Apache Kafka 의존성 버전은 Spring Boot의 의존성 관리에 의해 결정됩니다. 버전을 명시하지 않으면 Spring Boot BOM에서 관리하는 버전이 사용됩니다.
현재 코드는 Spring Boot 3.5.4의 BOM 의존성 관리를 활용하는 올바른 패턴입니다. 명시적 버전 지정은 BOM 관리를 무시하게 되므로 권장되지 않습니다.
spring-kafka-test의 버전은 자동으로spring-kafka와 일치합니다.Likely an incorrect or invalid review comment.
src/main/java/dooya/see/adapter/integration/kafka/PostEventProducer.java
Show resolved
Hide resolved
- Kafka 통합 및 도메인 이벤트 파이프라인 관련 문서 대폭 확장 - 이벤트 플로우 다이어그램 갱신: Kafka 및 DirectPublisher를 모두 포함하도록 변경 - Post 엔티티 및 값 객체 직렬화 지원에 대한 세부 내용 추가 - 메시지 재시도 정책, Fallback 전략, Embedded Kafka와 관련된 설정 팁 포함 - 개발 및 테스트 환경을 위한 Kafka 비활성화 옵션 설명 추가
- Kafka 이벤트 발행 중 발생할 수 있는 처리되지 않은 예외를 안전하게 처리하도록 수정 - `sendWithRetry` 실행 시 예외 로그를 기록하여 이슈 원인 파악 가능 - 안정성을 높이는 리팩토링 작업으로 서비스 신뢰성 강화 - DLQ 및 모니터링 알림 연동 고려를 위한 TODO 추가
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
src/main/java/dooya/see/adapter/integration/kafka/PostEventProducer.java (1)
76-102: 동기 전송 방식의 성능 영향을 고려하세요.현재 구현은
kafkaTemplate.send(...).get()(79라인)을 사용해 동기 방식으로 전송을 대기합니다. 이는 재시도 정책을 확실히 적용하고 전송 결과를 보장하지만, 트랜잭션 커밋 후 스레드를 블로킹합니다.처리량이 높은 환경에서는 다음을 고려하세요:
- CompletableFuture 체인으로 비동기 재시도 구현
- 또는 현재 방식 유지 시, 별도 스레드 풀로 afterCommit 작업 실행
99라인의 TODO 주석이 DLQ 연동의 필요성을 명시하고 있습니다.
처리량과 응답 시간 요구사항을 고려하여 현재 동기 방식이 적절한지 확인하세요. 필요하다면 비동기 재시도 구현이나 DLQ 연동 코드 생성을 도와드릴 수 있습니다.
README.md (1)
69-69: Markdown 리스트 들여쓰기를 일관되게 조정하세요.정적 분석 도구가 69, 72-75라인에서 중첩 리스트의 들여쓰기가 4칸이 아닌 2칸이어야 한다고 지적합니다. 기능적 문제는 아니지만, 린팅 규칙과의 일관성을 위해 수정을 권장합니다.
다음과 같이 수정하세요:
- **Application Layer**: 유스케이스 조율과 트랜잭션 관리 - - Primary Port: MemberManager, PostManager, CommentManager - - Secondary Port: MemberRepository, PostRepository, PostStatsRepository - - 이벤트 핸들러: PostEventHandler, PostStatsEventHandler + - Primary Port: MemberManager, PostManager, CommentManager + - Secondary Port: MemberRepository, PostRepository, PostStatsRepository + - 이벤트 핸들러: PostEventHandler, PostStatsEventHandler - **Adapter Layer**: 외부 시스템과의 연동 - - Web API: REST 엔드포인트 - - Security: JWT 인증/인가 - - Persistence: JPA 구현체 - - Integration: Kafka 프로듀서/컨슈머, Elasticsearch 색인기 + - Web API: REST 엔드포인트 + - Security: JWT 인증/인가 + - Persistence: JPA 구현체 + - Integration: Kafka 프로듀서/컨슈머, Elasticsearch 색인기Also applies to: 72-75
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
.docs/도메인 모델 문서.md(4 hunks).docs/용어 사전 문서.md(2 hunks)README.md(5 hunks)src/main/java/dooya/see/adapter/integration/kafka/PostEventProducer.java(1 hunks)
✅ Files skipped from review due to trivial changes (1)
- .docs/용어 사전 문서.md
🧰 Additional context used
🪛 markdownlint-cli2 (0.18.1)
README.md
69-69: Unordered list indentation
Expected: 2; Actual: 4
(MD007, ul-indent)
72-72: Unordered list indentation
Expected: 2; Actual: 4
(MD007, ul-indent)
73-73: Unordered list indentation
Expected: 2; Actual: 4
(MD007, ul-indent)
74-74: Unordered list indentation
Expected: 2; Actual: 4
(MD007, ul-indent)
75-75: Unordered list indentation
Expected: 2; Actual: 4
(MD007, ul-indent)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: test
🔇 Additional comments (9)
src/main/java/dooya/see/adapter/integration/kafka/PostEventProducer.java (3)
1-29: 클래스 선언 및 의존성 구성이 적절합니다.조건부 로딩(
@ConditionalOnProperty)과 필요한 의존성(KafkaTemplate, RetryTemplate)이 올바르게 설정되어 있습니다.
31-61: 트랜잭션 동기화 로직이 올바르게 구현되었습니다.트랜잭션 커밋 후 Kafka 전송을 지연시키는 로직이 적절하며, 예외 처리도 올바르게 되어 있습니다. 이전 리뷰에서 제기된
afterCommit내 예외 처리 문제가 44-48라인의 try-catch로 해결되었습니다.
63-74: 이벤트 변환 로직이 명확합니다.패턴 매칭을 사용한 이벤트 타입 변환이 깔끔하게 구현되어 있습니다.
README.md (2)
77-84: 이벤트 파이프라인 설명이 구현과 일치합니다.PostEventProducer 구현과 일관되게 트랜잭션 커밋 후 전송, postId 키 사용, RetryTemplate 재시도 정책이 정확히 문서화되어 있습니다.
223-231: 테스트 실행 가이드가 명확합니다.기본 테스트에서는 Kafka를 비활성화하고, 별도의 통합 테스트에서 Embedded Kafka를 사용하는 전략이 잘 문서화되어 있습니다.
.docs/도메인 모델 문서.md (4)
372-394: 이벤트 플로우 다이어그램이 아키텍처를 정확히 표현합니다.Kafka 기반 파이프라인과 Direct fallback 경로가 모두 명확히 표시되어 있으며, PostStatsEventHandler의 분리도 올바르게 표현되어 있습니다.
561-632: 코드 예제가 실제 구현과 일치합니다.PostEventHandler, PostEventProducer, PostEventProcessor의 예제 코드가 검토한 실제 구현 내용과 일치하며, 트랜잭션 동기화와 재시도 정책이 정확히 표현되어 있습니다.
636-642: 검증 결과,RetryTemplate은KafkaProducerConfig.java의kafkaRetryTemplate()메서드에서.maxAttempts(3)으로 명시적으로 설정되어 있습니다. 문서의 "최대 3회 재시도" 명시는 정확합니다.
254-254: 리뷰 의견이 부정확합니다.검증 결과, 문서에 명시된 모든 엔티티/값 객체가 실제로
Serializable을 구현하고 있습니다:
Post:java.io.Serializable구현 ✓PostContent:java.io.Serializable구현 ✓PostMetaData:Serializable구현 ✓Tag:Serializable구현 ✓문서의 설명이 정확하므로, 리뷰 의견의 우려 사항은 타당하지 않습니다.
Likely an incorrect or invalid review comment.
📌 Issue
🧐 현재 상황
🎯 목표
🛠 작업 내용
작업 할 내용을 입력해주세요.
도메인 이벤트 기반 색인 파이프라인 Kafka 도입
테스트 작성
🚀 기타 사항
Summary by CodeRabbit
새로운 기능
인프라 개선
테스트 & CI
문서