Skip to content

main merge #27

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kdyann merged 2 commits into from
Oct 17, 2025
Merged

main merge #27

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b7869d1
[FIX] 파티션 수
kdyann Oct 17, 2025
66e1c31
Merge pull request #26 from KE-WhyNot/fix/stream
kdyann Oct 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 21 additions & 4 deletions ranking/src/main/java/youthfi/ranking/config/TopologyConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand All @@ -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 직렬화 후 발행
Expand Down