From b7869d14667cd41d2922b30ea9aa4c5b27dfbcfc Mon Sep 17 00:00:00 2001
From: dayoung <dayoungann@naver.com>
Date: Fri, 17 Oct 2025 10:00:57 +0900
Subject: [PATCH] =?UTF-8?q?[FIX]=20=ED=8C=8C=ED=8B=B0=EC=85=98=20=EC=88=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../ranking/config/TopologyConfig.java        | 25 ++++++++++++++++---
 1 file changed, 21 insertions(+), 4 deletions(-)

diff --git a/ranking/src/main/java/youthfi/ranking/config/TopologyConfig.java b/ranking/src/main/java/youthfi/ranking/config/TopologyConfig.java
index 46ca230..6a8369b 100644
--- a/ranking/src/main/java/youthfi/ranking/config/TopologyConfig.java
+++ b/ranking/src/main/java/youthfi/ranking/config/TopologyConfig.java
@@ -37,15 +37,24 @@ public KStream<String, String> kStream(StreamsBuilder builder) {
                 .filter((k,v) -> v != null)
                 .selectKey((k,v) -> v.getUserId() + "|" + v.getStockId());
 
-        // 1) BUY 롯 상태 스토어 (키=userId|stockId, 값=Deque<BuyLot> 직렬화 바이트)
+        // ★ 3파티션 강제
+        KStream<String, ExecutionRow> execP3 = execStream.repartition(
+                Repartitioned.<String, ExecutionRow>as("exec-p3")
+                        .withNumberOfPartitions(3)
+                        .withKeySerde(Serdes.String())
+                        .withValueSerde(executionSerde())
+        );
+
+        // 1) BUY 롯 상태 스토어
         var lotsStoreBuilder = Stores.keyValueStoreBuilder(
                 Stores.persistentKeyValueStore("buy-lots"),
                 Serdes.String(), Serdes.ByteArray());
         builder.addStateStore(lotsStoreBuilder);
 
         // 2) FIFO 처리 → SELL 시 실현 수익률(Double) 방출
+        //    ★ execP3 사용
         KStream<String, Double> realizedRatePerTrade =
-                execStream.transformValues(
+                execP3.transformValues(
                         () -> new RealizedRateFifoTransformer("buy-lots"),
                         "buy-lots"
                 ).filter((k,v) -> v != null);
@@ -60,14 +69,22 @@ public KStream<String, String> kStream(StreamsBuilder builder) {
                                 .withValueSerde(Serdes.Double())
                 );
 
+        // ★ top10 앞단 3파티션 보장
+        KStream<String, Double> ratesP3 = userLatestRate.toStream().repartition(
+                Repartitioned.<String, Double>as("rates-p3")
+                        .withNumberOfPartitions(3)
+                        .withKeySerde(Serdes.String())
+                        .withValueSerde(Serdes.Double())
+        );
+
         // 4) Top10 계산 (StateStore)
         var topStoreBuilder = Stores.keyValueStoreBuilder(
                 Stores.persistentKeyValueStore("top10-store"),
                 Serdes.String(), Serdes.Double());
         builder.addStateStore(topStoreBuilder);
 
-        KStream<String, List<RankItem>> top10 = userLatestRate
-                .toStream()
+        // ★ ratesP3 사용
+        KStream<String, List<RankItem>> top10 = ratesP3
                 .transformValues(() -> new TopNTransformer("top10-store"), "top10-store");
 
         // 5) JSON 직렬화 후 발행