Repository navigation
Volume 9 - #395
Merged
Merged
Volume 9#395
Conversation
생성자에서 시스템 프로퍼티를 세팅하면 RedisConfig 의 프로퍼티 바인딩이 먼저 일어나는 앱(commerce-streamer)에서 localhost 기본값으로 붙는다. 클래스 로딩 시점(static)으로 옮겨 빈 생성 순서와 무관하게 동작하도록 수정. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
- RankingKey: occurredAt(Asia/Seoul) 기준 ranking:all:yyyyMMdd 키 계산 - RankingScorePolicy: view 0.1 / like ±0.2 / order 0.7×qty (주문 1건 > 좋아요 3건) - RankingEventHandler: 배치 내 (판, 상품)별 사전 합산 후 ZINCRBY, 실패는 best-effort 스킵 - RedisRankingRepository: 키 최초 생성 시에만 TTL 2일 - MetricsEventHandler 가 "새로 처리 여부" 반환 → 중복 이벤트는 랭킹 미반영 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
- GET /api/v1/rankings?date&page&size — ZREVRANGE 후 상품·브랜드 IN 쿼리 Aggregation - 랭킹 조회는 replica 읽기 (복제 지연 허용), 없는 날짜는 빈 목록 - 상품 상세에 오늘 순위(rank) 추가 — @Cacheable 밖에서 조합, 순위 없으면 null, Redis 장애 격리 - ProductRepository.findAllByIds 추가 (N+1 방지) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
📝 Walkthrough변경 목적: 이벤트 기반 일일 상품 랭킹을 도입하고 랭킹 조회 및 상품 상세 응답에 순위 정보를 제공한다. Walkthrough이 변경은 이벤트 기반 일간 랭킹 점수 집계와 Redis 저장을 추가하고, Commerce API에서 날짜별 랭킹 페이지와 상품 상세의 오늘 순위를 제공한다. 랭킹 데이터는 상품·브랜드 정보와 조합되며, 입력 검증과 E2E 테스트가 추가된다. Changes일간 랭킹 기능
Sequence Diagram(s)이벤트 기반 랭킹 반영sequenceDiagram
participant CatalogEventConsumer
participant MetricsEventHandler
participant RankingEventHandler
participant RedisRankingRepository
CatalogEventConsumer->>MetricsEventHandler: 이벤트 멱등 처리
MetricsEventHandler-->>CatalogEventConsumer: 적용 여부 반환
CatalogEventConsumer->>RankingEventHandler: 적용 이벤트 배치 전달
RankingEventHandler->>RedisRankingRepository: 점수 증분 및 TTL 반영
랭킹 페이지 조회sequenceDiagram
participant Client
participant RankingV1Controller
participant RankingFacade
participant RedisRankingRepository
participant ProductRepository
participant BrandRepository
Client->>RankingV1Controller: 랭킹 조회 요청
RankingV1Controller->>RankingFacade: 날짜·페이지·크기 전달
RankingFacade->>RedisRankingRepository: ZSET 랭킹 조회
RankingFacade->>ProductRepository: 상품 정보 조회
RankingFacade->>BrandRepository: 브랜드 정보 조회
RankingV1Controller-->>Client: 랭킹 페이지 응답
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (3 passed)
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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (3)
apps/commerce-streamer/src/main/java/com/loopers/application/ranking/RankingEventHandler.java (1)
36-38: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win빈 리스트 처리 시 불필요한 자원 낭비를 방지한다.
운영 환경에서는 이벤트가 모두 중복으로 판정되어
appliedMessages가 빈 리스트로 넘어올 가능성이 존재한다. 빈 리스트일 경우 불필요한 Map 객체 생성 및 순회를 막기 위해 빠른 종료(Early return)를 적용해야 한다. 빈 리스트 입력 시 예외 없이 정상 종료되는지 확인하는 테스트를 추가한다.
apps/commerce-streamer/src/main/java/com/loopers/application/ranking/RankingEventHandler.java#L36-L38:handleCatalog메서드 도입부에if (appliedMessages.isEmpty()) return;을 추가한다.apps/commerce-streamer/src/main/java/com/loopers/application/ranking/RankingEventHandler.java#L53-L55:handleOrder메서드 도입부에도 동일하게 빠른 종료 조건을 추가한다.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/commerce-streamer/src/main/java/com/loopers/application/ranking/RankingEventHandler.java` around lines 36 - 38, Update RankingEventHandler.handleCatalog and handleOrder to return immediately when appliedMessages is empty, before creating maps or iterating messages. Preserve existing processing for non-empty inputs, and add tests confirming empty lists complete normally without errors.apps/commerce-api/src/test/java/com/loopers/interfaces/api/RankingV1ApiE2ETest.java (2)
151-171: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick win
page와size의 경계값을 E2E 테스트에 추가해야 한다.현재는 잘못된 날짜만 검증하므로 요청 제한이 회귀해도 탐지되지 않는다.
page=0,size=0,size=101은 400을,page=1,size=1,size=100은 정상 응답을 검증해야 한다. 최대 크기 제한 회귀는 운영 환경의 Redis·DB 과다 조회로 이어질 수 있다.As per path instructions, 단위 테스트는 경계값/실패 케이스/예외 흐름을 포함하는지 점검한다. <path_instructions>
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/commerce-api/src/test/java/com/loopers/interfaces/api/RankingV1ApiE2ETest.java` around lines 151 - 171, Extend the RankingV1ApiE2ETest coverage around the existing test methods to validate pagination boundaries: assert HTTP 400 for page=0, size=0, and size=101, and assert successful responses for page=1, size=1, and size=100. Reuse the existing endpoint and response setup, keeping the assertions focused on the request validation outcomes.Source: Path instructions
222-238: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winRedis 장애 격리 시나리오를 별도로 검증해야 한다.
빈 랭킹에서
null을 반환하는 테스트는 Redis 예외 흐름을 통과하지 않는다. 랭킹 저장소가 예외를 던져도 상품 상세가 200과rank=null을 반환하는 테스트를 추가해야 한다. 그렇지 않으면 Redis 장애가 상품 상세 API 장애로 확산되는 회귀를 탐지할 수 없다.As per path instructions, 단위 테스트는 경계값/실패 케이스/예외 흐름을 포함하는지 점검한다. <path_instructions>
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/commerce-api/src/test/java/com/loopers/interfaces/api/RankingV1ApiE2ETest.java` around lines 222 - 238, RankingV1ApiE2ETest의 returnsNullRank_whenProductIsNotRanked와 별도로 Redis 랭킹 저장소가 예외를 던지는 장애 격리 시나리오를 추가하라. 랭킹 조회 의존성을 예외 발생하도록 설정한 뒤 상품 상세 API를 호출하고, 응답이 200 OK이며 ProductDetailResponse의 rank가 null인지 검증하라.Source: Path instructions
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@apps/commerce-api/src/main/java/com/loopers/application/ranking/RankingFacade.java`:
- Around line 47-54: Replace the per-ID brandRepository.findById loop in
RankingFacade with a single bulk IN-query using the collected brandIds, then
construct brandMap from the returned brands keyed by ID. Add an integration test
for a ranking containing multiple brands that verifies only one brand lookup
query is executed.
In
`@apps/commerce-api/src/main/java/com/loopers/interfaces/api/ranking/RankingV1Controller.java`:
- Around line 20-39: RankingV1Controller에 `@Validated를` 추가하고 getRankings의 page와
size 파라미터에 각각 `@Min`(1), `@Max`(100) 검증 애너테이션을 적용하세요. 컨트롤러 내부의 page/size 하드코딩 조건문과
직접 CoreException을 발생시키는 검증 로직은 제거하고, 검증 실패가 기존 `@ControllerAdvice의` 표준 예외 응답으로
처리되도록 관련 테스트를 추가하세요.
In
`@apps/commerce-streamer/src/main/java/com/loopers/infrastructure/ranking/RedisRankingRepository.java`:
- Around line 28-42: Update RedisRankingRepository.incrementScores to execute
all ZINCRBY operations through redisTemplate.executePipelined, preserving the
empty-input early return and existing key/TTL behavior. Apply the expiry
handling after the pipeline completes, and add an integration test covering
multiple product score updates through the pipeline.
In
`@apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/CatalogEventConsumer.java`:
- Around line 38-49: Update the record-processing loop in CatalogEventConsumer
to skip records whose record.value() is null before deserialization, without
logging an error. For deserialization or handling failures, extend the existing
log.error call to include record.topic(), record.partition(), and
record.offset() so the failed Kafka message is fully identifiable. Add unit
tests covering silent tombstone skipping and poison-message logs containing
topic and partition.
In
`@apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/OrderEventConsumer.java`:
- Around line 40-47: Update the exception handling in the OrderEventConsumer
processing loop to catch and skip only non-recoverable deserialization failures
from objectMapper.readValue, while allowing transaction or infrastructure
exceptions thrown by metricsEventHandler.handleOrder to propagate to the Kafka
container for retry and prevent acknowledge from committing the batch. Add an
integration test covering a DB/infrastructure failure and verifying the offset
is not committed.
---
Nitpick comments:
In
`@apps/commerce-api/src/test/java/com/loopers/interfaces/api/RankingV1ApiE2ETest.java`:
- Around line 151-171: Extend the RankingV1ApiE2ETest coverage around the
existing test methods to validate pagination boundaries: assert HTTP 400 for
page=0, size=0, and size=101, and assert successful responses for page=1,
size=1, and size=100. Reuse the existing endpoint and response setup, keeping
the assertions focused on the request validation outcomes.
- Around line 222-238: RankingV1ApiE2ETest의
returnsNullRank_whenProductIsNotRanked와 별도로 Redis 랭킹 저장소가 예외를 던지는 장애 격리 시나리오를
추가하라. 랭킹 조회 의존성을 예외 발생하도록 설정한 뒤 상품 상세 API를 호출하고, 응답이 200 OK이며
ProductDetailResponse의 rank가 null인지 검증하라.
In
`@apps/commerce-streamer/src/main/java/com/loopers/application/ranking/RankingEventHandler.java`:
- Around line 36-38: Update RankingEventHandler.handleCatalog and handleOrder to
return immediately when appliedMessages is empty, before creating maps or
iterating messages. Preserve existing processing for non-empty inputs, and add
tests confirming empty lists complete normally without errors.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 10f7c573-c272-411c-9bd2-100330a8fb25
📒 Files selected for processing (29)
apps/commerce-api/src/main/java/com/loopers/application/ranking/RankingFacade.javaapps/commerce-api/src/main/java/com/loopers/application/ranking/RankingInfo.javaapps/commerce-api/src/main/java/com/loopers/domain/product/ProductRepository.javaapps/commerce-api/src/main/java/com/loopers/domain/ranking/RankedProduct.javaapps/commerce-api/src/main/java/com/loopers/domain/ranking/RankingKey.javaapps/commerce-api/src/main/java/com/loopers/domain/ranking/RankingRepository.javaapps/commerce-api/src/main/java/com/loopers/infrastructure/product/ProductRepositoryImpl.javaapps/commerce-api/src/main/java/com/loopers/infrastructure/ranking/RedisRankingRepository.javaapps/commerce-api/src/main/java/com/loopers/interfaces/api/product/ProductV1Controller.javaapps/commerce-api/src/main/java/com/loopers/interfaces/api/product/ProductV1Dto.javaapps/commerce-api/src/main/java/com/loopers/interfaces/api/ranking/RankingV1ApiSpec.javaapps/commerce-api/src/main/java/com/loopers/interfaces/api/ranking/RankingV1Controller.javaapps/commerce-api/src/main/java/com/loopers/interfaces/api/ranking/RankingV1Dto.javaapps/commerce-api/src/test/java/com/loopers/interfaces/api/RankingV1ApiE2ETest.javaapps/commerce-api/src/test/resources/docker-java.propertiesapps/commerce-streamer/src/main/java/com/loopers/application/metrics/MetricsEventHandler.javaapps/commerce-streamer/src/main/java/com/loopers/application/ranking/RankingEventHandler.javaapps/commerce-streamer/src/main/java/com/loopers/domain/ranking/RankingKey.javaapps/commerce-streamer/src/main/java/com/loopers/domain/ranking/RankingRepository.javaapps/commerce-streamer/src/main/java/com/loopers/domain/ranking/RankingScorePolicy.javaapps/commerce-streamer/src/main/java/com/loopers/infrastructure/ranking/RedisRankingRepository.javaapps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/CatalogEventConsumer.javaapps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/OrderEventConsumer.javaapps/commerce-streamer/src/test/java/com/loopers/application/metrics/MetricsEventHandlerTest.javaapps/commerce-streamer/src/test/java/com/loopers/application/ranking/RankingEventHandlerTest.javaapps/commerce-streamer/src/test/java/com/loopers/domain/ranking/RankingKeyTest.javaapps/commerce-streamer/src/test/java/com/loopers/domain/ranking/RankingScorePolicyTest.javahttp/commerce-api/rankings.httpmodules/redis/src/testFixtures/java/com/loopers/testcontainers/RedisTestContainersConfig.java
Comment on lines
+47
to
+54
| Set<Long> brandIds = productMap.values().stream() | ||
| .map(ProductModel::getBrandId) | ||
| .collect(Collectors.toSet()); | ||
| Map<Long, BrandModel> brandMap = new HashMap<>(); | ||
| for (Long id : brandIds) { | ||
| brandRepository.findById(id).ifPresent(b -> brandMap.put(id, b)); | ||
| } | ||
|
|
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
루프 내 개별 DB 조회(N+1)로 인한 성능 저하를 방지해야 한다.
여러 상품의 브랜드 정보를 조회할 때 findById를 반복 호출하면 N+1 문제가 발생한다. 트래픽이 몰릴 경우 DB 커넥션 풀을 고갈시키고 전체 시스템의 지연(Latency)과 장애를 유발할 수 있다.
IN 쿼리를 사용하여 한 번에 조회하도록 수정한다. 또한 여러 브랜드가 포함된 랭킹 조회 시 쿼리가 1회만 발생하는지 통합 테스트를 통해 검증해야 한다. As per path instructions, N+1 가능성을 점검해야 한다.
🛠️ 제안하는 수정안
Set<Long> brandIds = productMap.values().stream()
.map(ProductModel::getBrandId)
.collect(Collectors.toSet());
Map<Long, BrandModel> brandMap = new HashMap<>();
- for (Long id : brandIds) {
- brandRepository.findById(id).ifPresent(b -> brandMap.put(id, b));
- }
+ brandRepository.findAllById(brandIds).forEach(b -> brandMap.put(b.getId(), b));📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| Set<Long> brandIds = productMap.values().stream() | |
| .map(ProductModel::getBrandId) | |
| .collect(Collectors.toSet()); | |
| Map<Long, BrandModel> brandMap = new HashMap<>(); | |
| for (Long id : brandIds) { | |
| brandRepository.findById(id).ifPresent(b -> brandMap.put(id, b)); | |
| } | |
| Set<Long> brandIds = productMap.values().stream() | |
| .map(ProductModel::getBrandId) | |
| .collect(Collectors.toSet()); | |
| Map<Long, BrandModel> brandMap = new HashMap<>(); | |
| brandRepository.findAllById(brandIds).forEach(b -> brandMap.put(b.getId(), b)); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@apps/commerce-api/src/main/java/com/loopers/application/ranking/RankingFacade.java`
around lines 47 - 54, Replace the per-ID brandRepository.findById loop in
RankingFacade with a single bulk IN-query using the collected brandIds, then
construct brandMap from the returned brands keyed by ID. Add an integration test
for a ranking containing multiple brands that verifies only one brand lookup
query is executed.
Source: Path instructions
Comment on lines
+20
to
+39
| @RequiredArgsConstructor | ||
| @RestController | ||
| @RequestMapping("/api/v1/rankings") | ||
| public class RankingV1Controller implements RankingV1ApiSpec { | ||
|
|
||
| private static final DateTimeFormatter DATE_FORMAT = DateTimeFormatter.BASIC_ISO_DATE; | ||
|
|
||
| private final RankingFacade rankingFacade; | ||
|
|
||
| @GetMapping | ||
| @Override | ||
| public ApiResponse<RankingV1Dto.RankingPageResponse> getRankings( | ||
| @RequestParam(required = false) String date, | ||
| @RequestParam(required = false, defaultValue = "1") Integer page, | ||
| @RequestParam(required = false, defaultValue = "20") Integer size | ||
| ) { | ||
| LocalDate targetDate = parseDateOrToday(date); | ||
| if (page < 1 || size < 1 || size > 100) { | ||
| throw new CoreException(ErrorType.BAD_REQUEST, "page 는 1 이상, size 는 1~100 이어야 합니다."); | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Bean Validation을 사용하여 요청 파라미터 검증을 표준화해야 한다.
컨트롤러 내부에 하드코딩된 조건문으로 파라미터를 검증하면, 에러 응답 포맷이 파편화되고 @ControllerAdvice를 통한 중앙 집중식 예외 처리를 활용하기 어렵다. 이는 운영 환경에서 API 에러 모니터링 및 클라이언트 연동에 혼선을 초래한다.
클래스 레벨에 @Validated를 추가하고 파라미터에 @Min, @Max 애너테이션을 적용하여 검증 로직을 위임한다. 검증 실패 시 전역 예외 처리기에서 동일한 형태의 표준 에러 응답이 내려가는지 검증하는 테스트를 추가해야 한다. As per path instructions, Controller는 요청 검증(Bean Validation)에 집중하고 @ControllerAdvice 기반 표준 처리로 유도해야 한다.
🛠️ 제안하는 수정안 (Validation 의존성 추가 필요)
+import org.springframework.validation.annotation.Validated;
+import jakarta.validation.constraints.Min;
+import jakarta.validation.constraints.Max;
+
+@Validated
`@RequiredArgsConstructor`
`@RestController`
`@RequestMapping`("/api/v1/rankings")
public class RankingV1Controller implements RankingV1ApiSpec {
// ...
`@GetMapping`
`@Override`
public ApiResponse<RankingV1Dto.RankingPageResponse> getRankings(
`@RequestParam`(required = false) String date,
- `@RequestParam`(required = false, defaultValue = "1") Integer page,
- `@RequestParam`(required = false, defaultValue = "20") Integer size
+ `@RequestParam`(required = false, defaultValue = "1") `@Min`(1) Integer page,
+ `@RequestParam`(required = false, defaultValue = "20") `@Min`(1) `@Max`(100) Integer size
) {
LocalDate targetDate = parseDateOrToday(date);
- if (page < 1 || size < 1 || size > 100) {
- throw new CoreException(ErrorType.BAD_REQUEST, "page 는 1 이상, size 는 1~100 이어야 합니다.");
- }
List<RankingInfo> infos = rankingFacade.getRankingPage(targetDate, page, size);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| @RequiredArgsConstructor | |
| @RestController | |
| @RequestMapping("/api/v1/rankings") | |
| public class RankingV1Controller implements RankingV1ApiSpec { | |
| private static final DateTimeFormatter DATE_FORMAT = DateTimeFormatter.BASIC_ISO_DATE; | |
| private final RankingFacade rankingFacade; | |
| @GetMapping | |
| @Override | |
| public ApiResponse<RankingV1Dto.RankingPageResponse> getRankings( | |
| @RequestParam(required = false) String date, | |
| @RequestParam(required = false, defaultValue = "1") Integer page, | |
| @RequestParam(required = false, defaultValue = "20") Integer size | |
| ) { | |
| LocalDate targetDate = parseDateOrToday(date); | |
| if (page < 1 || size < 1 || size > 100) { | |
| throw new CoreException(ErrorType.BAD_REQUEST, "page 는 1 이상, size 는 1~100 이어야 합니다."); | |
| } | |
| `@RequiredArgsConstructor` | |
| `@RestController` | |
| `@RequestMapping`("/api/v1/rankings") | |
| `@Validated` | |
| public class RankingV1Controller implements RankingV1ApiSpec { | |
| private static final DateTimeFormatter DATE_FORMAT = DateTimeFormatter.BASIC_ISO_DATE; | |
| private final RankingFacade rankingFacade; | |
| `@GetMapping` | |
| `@Override` | |
| public ApiResponse<RankingV1Dto.RankingPageResponse> getRankings( | |
| `@RequestParam`(required = false) String date, | |
| `@RequestParam`(required = false, defaultValue = "1") `@Min`(1) Integer page, | |
| `@RequestParam`(required = false, defaultValue = "20") `@Min`(1) `@Max`(100) Integer size | |
| ) { | |
| LocalDate targetDate = parseDateOrToday(date); | |
| List<RankingInfo> infos = rankingFacade.getRankingPage(targetDate, page, size); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@apps/commerce-api/src/main/java/com/loopers/interfaces/api/ranking/RankingV1Controller.java`
around lines 20 - 39, RankingV1Controller에 `@Validated를` 추가하고 getRankings의 page와
size 파라미터에 각각 `@Min`(1), `@Max`(100) 검증 애너테이션을 적용하세요. 컨트롤러 내부의 page/size 하드코딩 조건문과
직접 CoreException을 발생시키는 검증 로직은 제거하고, 검증 실패가 기존 `@ControllerAdvice의` 표준 예외 응답으로
처리되도록 관련 테스트를 추가하세요.
Source: Path instructions
Comment on lines
+28
to
+42
| @Override | ||
| public void incrementScores(RankingKey key, Map<Long, Double> scoreDeltaByProductId) { | ||
| if (scoreDeltaByProductId.isEmpty()) { | ||
| return; | ||
| } | ||
| String keyValue = key.value(); | ||
| scoreDeltaByProductId.forEach((productId, delta) -> | ||
| redisTemplate.opsForZSet().incrementScore(keyValue, String.valueOf(productId), delta)); | ||
|
|
||
| // ZINCRBY 로 키가 처음 생겼을 때만 TTL 을 건다. (매번 걸면 쓰기가 있을 때마다 만료가 밀린다) | ||
| Long ttlSeconds = redisTemplate.getExpire(keyValue); | ||
| if (ttlSeconds != null && ttlSeconds == -1) { | ||
| redisTemplate.expire(keyValue, TTL); | ||
| } | ||
| } |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
Redis 파이프라인을 사용하여 네트워크 왕복(Round-trip) 횟수를 최소화한다.
운영 관점에서 이벤트 스트림을 처리할 때 상품 수만큼 Redis에 개별적으로 incrementScore를 호출하면 N번의 네트워크 왕복이 발생하여 애플리케이션 지연 및 병목의 원인이 된다. redisTemplate.executePipelined를 사용하여 명령을 한 번에 전송하도록 변경해야 한다. 추가로 대량의 상품 점수 업데이트 시 파이프라인 처리가 정상 작동하는지 검증하는 통합 테스트를 작성한다.
⚡ 제안하는 파이프라인 적용 수정안
`@Override`
public void incrementScores(RankingKey key, Map<Long, Double> scoreDeltaByProductId) {
if (scoreDeltaByProductId.isEmpty()) {
return;
}
String keyValue = key.value();
+
+ Long currentTtl = redisTemplate.getExpire(keyValue);
+ boolean needsTtl = currentTtl != null && currentTtl < 0;
+
+ redisTemplate.executePipelined(new org.springframework.data.redis.core.SessionCallback<Object>() {
+ `@Override`
+ public <K, V> Object execute(org.springframework.data.redis.core.RedisOperations<K, V> operations) throws org.springframework.dao.DataAccessException {
+ `@SuppressWarnings`("unchecked")
+ org.springframework.data.redis.core.RedisOperations<String, String> stringOps = (org.springframework.data.redis.core.RedisOperations<String, String>) operations;
+ scoreDeltaByProductId.forEach((productId, delta) ->
+ stringOps.opsForZSet().incrementScore(keyValue, String.valueOf(productId), delta));
+
+ if (needsTtl) {
+ stringOps.expire(keyValue, TTL);
+ }
+ return null;
+ }
+ });
- scoreDeltaByProductId.forEach((productId, delta) ->
- redisTemplate.opsForZSet().incrementScore(keyValue, String.valueOf(productId), delta));
-
- // ZINCRBY 로 키가 처음 생겼을 때만 TTL 을 건다. (매번 걸면 쓰기가 있을 때마다 만료가 밀린다)
- Long ttlSeconds = redisTemplate.getExpire(keyValue);
- if (ttlSeconds != null && ttlSeconds == -1) {
- redisTemplate.expire(keyValue, TTL);
- }
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| @Override | |
| public void incrementScores(RankingKey key, Map<Long, Double> scoreDeltaByProductId) { | |
| if (scoreDeltaByProductId.isEmpty()) { | |
| return; | |
| } | |
| String keyValue = key.value(); | |
| scoreDeltaByProductId.forEach((productId, delta) -> | |
| redisTemplate.opsForZSet().incrementScore(keyValue, String.valueOf(productId), delta)); | |
| // ZINCRBY 로 키가 처음 생겼을 때만 TTL 을 건다. (매번 걸면 쓰기가 있을 때마다 만료가 밀린다) | |
| Long ttlSeconds = redisTemplate.getExpire(keyValue); | |
| if (ttlSeconds != null && ttlSeconds == -1) { | |
| redisTemplate.expire(keyValue, TTL); | |
| } | |
| } | |
| `@Override` | |
| public void incrementScores(RankingKey key, Map<Long, Double> scoreDeltaByProductId) { | |
| if (scoreDeltaByProductId.isEmpty()) { | |
| return; | |
| } | |
| String keyValue = key.value(); | |
| Long currentTtl = redisTemplate.getExpire(keyValue); | |
| boolean needsTtl = currentTtl != null && currentTtl < 0; | |
| redisTemplate.executePipelined(new org.springframework.data.redis.core.SessionCallback<Object>() { | |
| `@Override` | |
| public <K, V> Object execute(org.springframework.data.redis.core.RedisOperations<K, V> operations) throws org.springframework.dao.DataAccessException { | |
| `@SuppressWarnings`("unchecked") | |
| org.springframework.data.redis.core.RedisOperations<String, String> stringOps = (org.springframework.data.redis.core.RedisOperations<String, String>) operations; | |
| scoreDeltaByProductId.forEach((productId, delta) -> | |
| stringOps.opsForZSet().incrementScore(keyValue, String.valueOf(productId), delta)); | |
| if (needsTtl) { | |
| stringOps.expire(keyValue, TTL); | |
| } | |
| return null; | |
| } | |
| }); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@apps/commerce-streamer/src/main/java/com/loopers/infrastructure/ranking/RedisRankingRepository.java`
around lines 28 - 42, Update RedisRankingRepository.incrementScores to execute
all ZINCRBY operations through redisTemplate.executePipelined, preserving the
empty-input early return and existing key/TTL behavior. Apply the expiry
handling after the pipeline completes, and add an integration test covering
multiple product score updates through the pipeline.
Comment on lines
+38
to
49
| List<CatalogEventMessage> applied = new ArrayList<>(); | ||
| for (ConsumerRecord<String, byte[]> record : records) { | ||
| try { | ||
| CatalogEventMessage message = objectMapper.readValue(record.value(), CatalogEventMessage.class); | ||
| metricsEventHandler.handleCatalog(message); | ||
| if (metricsEventHandler.handleCatalog(message)) { | ||
| applied.add(message); // 새로 반영된 이벤트만 랭킹 대상 (중복 제외) | ||
| } | ||
| } catch (Exception e) { | ||
| // 개별 메시지 처리 실패는 로깅 후 스킵(포이즌 메시지가 파티션을 막지 않게). 필요 시 DLQ. | ||
| log.error("[catalog-events] 처리 실패 offset={} → 스킵", record.offset(), e); | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
메시지 파티션 로깅 누락 및 툼스톤 메시지 처리 부재를 개선해야 한다.
운영 환경에서 예외 발생 시, 토픽과 파티션 정보 없이 오프셋만으로는 Kafka 브로커에서 실패한 메시지를 추적할 수 없다. 또한, 값이 null인 툼스톤(Tombstone) 메시지가 유입될 경우, objectMapper.readValue에서 파싱 예외가 발생하여 불필요한 에러 로그가 기록된다. 이는 디버깅을 어렵게 하고 불필요한 로그 비용을 발생시킨다.
수정안:
record.value()가null인 경우 처리를 건너뛰는 방어적 코드를 추가한다.- 에러 로그에
record.topic()및record.partition()을 추가하여 장애 시 메시지 위치를 명확히 식별할 수 있도록 한다.
추가 테스트:
ConsumerRecord의 값이null로 들어왔을 때, 에러 로그가 발생하지 않고 무시되는지 검증하는 단위 테스트를 추가한다.- 메시지 파싱이 실패하는 포이즌 메시지를 주입했을 때, 토픽과 파티션 정보가 포함된 에러 로그가 남는지 확인하는 테스트를 추가한다.
💻 제안하는 수정안
List<CatalogEventMessage> applied = new ArrayList<>();
for (ConsumerRecord<String, byte[]> record : records) {
+ if (record.value() == null) {
+ continue; // 툼스톤 메시지 무시
+ }
try {
CatalogEventMessage message = objectMapper.readValue(record.value(), CatalogEventMessage.class);
if (metricsEventHandler.handleCatalog(message)) {
applied.add(message); // 새로 반영된 이벤트만 랭킹 대상 (중복 제외)
}
} catch (Exception e) {
// 개별 메시지 처리 실패는 로깅 후 스킵(포이즌 메시지가 파티션을 막지 않게). 필요 시 DLQ.
- log.error("[catalog-events] 처리 실패 offset={} → 스킵", record.offset(), e);
+ log.error("[catalog-events] 처리 실패 topic={} partition={} offset={} → 스킵",
+ record.topic(), record.partition(), record.offset(), e);
}
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| List<CatalogEventMessage> applied = new ArrayList<>(); | |
| for (ConsumerRecord<String, byte[]> record : records) { | |
| try { | |
| CatalogEventMessage message = objectMapper.readValue(record.value(), CatalogEventMessage.class); | |
| metricsEventHandler.handleCatalog(message); | |
| if (metricsEventHandler.handleCatalog(message)) { | |
| applied.add(message); // 새로 반영된 이벤트만 랭킹 대상 (중복 제외) | |
| } | |
| } catch (Exception e) { | |
| // 개별 메시지 처리 실패는 로깅 후 스킵(포이즌 메시지가 파티션을 막지 않게). 필요 시 DLQ. | |
| log.error("[catalog-events] 처리 실패 offset={} → 스킵", record.offset(), e); | |
| } | |
| } | |
| List<CatalogEventMessage> applied = new ArrayList<>(); | |
| for (ConsumerRecord<String, byte[]> record : records) { | |
| if (record.value() == null) { | |
| continue; // 툼스톤 메시지 무시 | |
| } | |
| try { | |
| CatalogEventMessage message = objectMapper.readValue(record.value(), CatalogEventMessage.class); | |
| if (metricsEventHandler.handleCatalog(message)) { | |
| applied.add(message); // 새로 반영된 이벤트만 랭킹 대상 (중복 제외) | |
| } | |
| } catch (Exception e) { | |
| // 개별 메시지 처리 실패는 로깅 후 스킵(포이즌 메시지가 파티션을 막지 않게). 필요 시 DLQ. | |
| log.error("[catalog-events] 처리 실패 topic={} partition={} offset={} → 스킵", | |
| record.topic(), record.partition(), record.offset(), e); | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/CatalogEventConsumer.java`
around lines 38 - 49, Update the record-processing loop in CatalogEventConsumer
to skip records whose record.value() is null before deserialization, without
logging an error. For deserialization or handling failures, extend the existing
log.error call to include record.topic(), record.partition(), and
record.offset() so the failed Kafka message is fully identifiable. Add unit
tests covering silent tombstone skipping and poison-message logs containing
topic and partition.
Source: Path instructions
Comment on lines
40
to
47
| try { | ||
| OrderEventMessage message = objectMapper.readValue(record.value(), OrderEventMessage.class); | ||
| metricsEventHandler.handleOrder(message); | ||
| if (metricsEventHandler.handleOrder(message)) { | ||
| applied.add(message); // 새로 반영된 이벤트만 랭킹 대상 (중복 제외) | ||
| } | ||
| } catch (Exception e) { | ||
| log.error("[order-events] 처리 실패 offset={} → 스킵", record.offset(), e); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
인프라 장애 시 데이터 유실을 막기 위해 예외 처리를 세분화한다.
운영 관점에서 MetricsEventHandler 호출 시 DB 연결 실패 등 일시적인 인프라 예외가 발생할 수 있다. 현재는 모든 Exception을 잡고 반복문을 계속 진행한 뒤 오프셋을 커밋(acknowledge)하므로, DB 장애 시 해당 배치의 모든 이벤트가 유실된다. 메시지 역직렬화 실패(포이즌 필)와 같은 복구 불가능한 예외만 건너뛰고, 트랜잭션/인프라 장애는 밖으로 던져 카프카 컨테이너가 재시도하도록 수정해야 한다. 재시도 시나리오에 대한 통합 테스트(예: DB 예외 발생 시 오프셋이 커밋되지 않는지)를 추가한다.
🚨 제안하는 예외 처리 세분화 수정안
- try {
- OrderEventMessage message = objectMapper.readValue(record.value(), OrderEventMessage.class);
- if (metricsEventHandler.handleOrder(message)) {
- applied.add(message); // 새로 반영된 이벤트만 랭킹 대상 (중복 제외)
- }
- } catch (Exception e) {
- log.error("[order-events] 처리 실패 offset={} → 스킵", record.offset(), e);
- }
+ OrderEventMessage message;
+ try {
+ message = objectMapper.readValue(record.value(), OrderEventMessage.class);
+ } catch (com.fasterxml.jackson.core.JsonProcessingException e) {
+ log.error("[order-events] 메시지 역직렬화 실패 offset={} → 스킵", record.offset(), e);
+ continue;
+ }
+
+ if (metricsEventHandler.handleOrder(message)) {
+ applied.add(message);
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| try { | |
| OrderEventMessage message = objectMapper.readValue(record.value(), OrderEventMessage.class); | |
| metricsEventHandler.handleOrder(message); | |
| if (metricsEventHandler.handleOrder(message)) { | |
| applied.add(message); // 새로 반영된 이벤트만 랭킹 대상 (중복 제외) | |
| } | |
| } catch (Exception e) { | |
| log.error("[order-events] 처리 실패 offset={} → 스킵", record.offset(), e); | |
| } | |
| OrderEventMessage message; | |
| try { | |
| message = objectMapper.readValue(record.value(), OrderEventMessage.class); | |
| } catch (com.fasterxml.jackson.core.JsonProcessingException e) { | |
| log.error("[order-events] 메시지 역직렬화 실패 offset={} → 스킵", record.offset(), e); | |
| continue; | |
| } | |
| if (metricsEventHandler.handleOrder(message)) { | |
| applied.add(message); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/OrderEventConsumer.java`
around lines 40 - 47, Update the exception handling in the OrderEventConsumer
processing loop to catch and skip only non-recoverable deserialization failures
from objectMapper.readValue, while allowing transaction or infrastructure
exceptions thrown by metricsEventHandler.handleOrder to propagate to the Kafka
container for retry and prevent acknowledge from committing the batch. Add an
integration test covering a DB/infrastructure failure and verifying the offset
is not committed.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
🧭 Context & Decision
문제 정의
product_metrics에 누적 집계만 한다. "오늘의 인기상품"을 뽑으려면 DBGROUP BY + ORDER BY가 필요한데, 랭킹은 조회 빈도가 매우 높은 지면이라 데이터가 쌓일수록 DB 부하로 직결된다. 또한 누적 집계만으로는 오래된 인기 상품이 상위권을 독식한다(롱테일).ranking:all:{yyyyMMdd}ZSET 에 가중치 점수 반영 (TTL 2일)GET /api/v1/rankings?date&page&size— 상품 정보 Aggregation 포함 랭킹 페이지 조회선택지와 결정
product_metrics가 갖는다. 현재 구조에서 예외를 던져 배치를 재소비시켜도 DB 멱등 체크에 걸려 applied 가 비므로 랭킹은 어차피 복구되지 않는다 — 던져서 얻는 게 없다면 명시적으로 삼키는 것이 낫다고 판단.price×amount는 보류. 가중치는 "주문 1건 > 좋아요 3건" 제약을 부등식으로 놓고 역산 (0.7 > 0.2×3).🤔 고민한 점 / 막혔던 부분
🙋 기타
JAVA_HOME=(jdk21) ./gradlew :apps:commerce-streamer:test :apps:commerce-api:test전체 통과 (Testcontainers MySQL/Redis)http/commerce-api/rankings.http