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 kStream(StreamsBuilder builder) { .filter((k,v) -> v != null) .selectKey((k,v) -> v.getUserId() + "|" + v.getStockId()); - // 1) BUY 롯 상태 스토어 (키=userId|stockId, 값=Deque 직렬화 바이트) + // ★ 3파티션 강제 + KStream execP3 = execStream.repartition( + Repartitioned.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 realizedRatePerTrade = - execStream.transformValues( + execP3.transformValues( () -> new RealizedRateFifoTransformer("buy-lots"), "buy-lots" ).filter((k,v) -> v != null); @@ -60,14 +69,22 @@ public KStream kStream(StreamsBuilder builder) { .withValueSerde(Serdes.Double()) ); + // ★ top10 앞단 3파티션 보장 + KStream ratesP3 = userLatestRate.toStream().repartition( + Repartitioned.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> top10 = userLatestRate - .toStream() + // ★ ratesP3 사용 + KStream> top10 = ratesP3 .transformValues(() -> new TopNTransformer("top10-store"), "top10-store"); // 5) JSON 직렬화 후 발행