28503055923

Committed 01 Jul 2026 08:06AM UTC coverage: 90.445% (-0.4%) from 90.887%

Build # 28503055923

Build Type

push

github

Committed by

Hyshmily

Commit Message

perf : promote throughput performance

Signed-off-by: Hyshmily <cxm8607@outlook.com>

Coverage Stats

1248 of 1441 branches covered (86.61%)

Branch coverage included in aggregate %.

123 of 139 new or added lines in 10 files covered. (88.49%)

2 existing lines in 2 files now uncovered.

3532 of 3844 relevant lines covered (91.88%)

4.2 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.56

/worker/src/main/java/io/github/hyshmily/hotkey/worker/ingest/ReportConsumer.java

/*
 * Copyright 2026 Hyshmily. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.github.hyshmily.hotkey.worker.ingest;

import static io.github.hyshmily.hotkey.constants.HotKeyConstants.SOURCE_SLIDING_WINDOW;
import static io.github.hyshmily.hotkey.util.TimeSource.currentTimeMillis;

import io.github.hyshmily.hotkey.detection.HotKeyStateMachine;
import io.github.hyshmily.hotkey.hotkeydetector.heavykeeper.TopK;
import io.github.hyshmily.hotkey.model.HotKeyDecision;
import io.github.hyshmily.hotkey.reporting.ReportMessage;
import io.github.hyshmily.hotkey.worker.detection.GlobalQpsEstimator;
import io.github.hyshmily.hotkey.worker.detection.SlidingWindowDetector;
import io.github.hyshmily.hotkey.worker.detection.TopKValidator;
import io.github.hyshmily.hotkey.worker.dispatch.WorkerBroadcaster;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.LongAdder;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;

/**
 * Worker‑side message consumer that receives batched per‑key access counts
 * reported by application instances.
 *
 * <p>For every key in the batch the consumer:
 * <ol>
 *   <li>Feeds the count into the {@link SlidingWindowDetector} to update the
 *       sliding‑window sum and obtain a binary hot‑or‑not verdict for the
 *       current window.</li>
 *   <li>Passes that verdict to the {@link HotKeyStateMachine} which tracks
 *       consecutive hot/cold windows and decides whether a state transition
 *       (COLD → CONFIRMED_HOT → PRE_COOLING → COLD) has occurred.</li>
 *   <li>If the state machine returns a {@code HOT} decision, the consumer
 *       broadcasts a {@code HOT} message to all application instances and
 *       records the confirmation in the {@link TopKValidator}.</li>
 *   <li>If the state machine returns a {@code COOL} decision, it broadcasts
 *       a {@code COOL} message and marks the key as cooled in the
 *       {@link TopKValidator}.</li>
 * </ol>
 *
 * <p>Messages older than 5 seconds are silently discarded to prevent stale
 * data from distorting the sliding‑window view.
 *
 * <p>Because clients use consistent‑hash routing, every report for a given
 * key always reaches the same worker, guaranteeing correct per‑key state
 * without cross‑worker coordination.
 */
@RequiredArgsConstructor
@Slf4j
public class ReportConsumer {

  /** Sliding-window detector that tracks per-key access counts and returns hot/cold verdicts. */
  private final SlidingWindowDetector detector;
  /** Per-key lifecycle state machine managing COLD / CONFIRMED_HOT / PRE_COOLING transitions. */
  private final HotKeyStateMachine stateMachine;
  /** Publishes HOT and COOL decisions back to all application instances. */
  private final WorkerBroadcaster broadcaster;
  /** TopK pre-warm validator for cross-instance frequency-based confirmation. */
  private final TopKValidator topKValidator;
  /** Worker-scoped HeavyKeeper sketch for cross-instance frequency estimation. */
  private final TopK workerTopK;
  /** Global QPS estimator tracking overall throughput for dynamic threshold learning. */
  private final GlobalQpsEstimator globalQpsEstimator;

  /** Staleness threshold in milliseconds. Package-visible for testing. */
  long stalenessThresholdMs = 5000L;

  /** Max keys per chunk for parallel processing. Beyond this, keys are split into chunks. */
  private static final int CHUNK_SIZE = 1000;

  /** Log a drain-progress summary when the pending broadcast queue exceeds this threshold. */
  private static final int BATCH_DRAIN_WARN_THRESHOLD = 5000;

  /**
   * Main entry point for batched report messages.
   *
   * @param message the deserialized message containing counts for multiple keys
   */
  @RabbitListener(queues = "#{@reportQueue.name}", containerFactory = "reportListenerContainerFactory")
  public void onReport(ReportMessage message) {
    try {
      doOnReport(message);
    } catch (Exception e) {
      log.error(
        "Uncaught exception in onReport, discarding message to prevent poison-message requeue loop: appName={}",
        message != null ? message.appName() : "null",
        e
      );
    }
  }

  private void doOnReport(ReportMessage message) {
    long now = currentTimeMillis();
    LongAdder totalQps = new LongAdder();

    // Discard reports that are more than 5 seconds old.
    // Guards against delayed or re‑delivered messages that would
    // feed outdated counts into the sliding window.
    if (now - message.timestamp() > stalenessThresholdMs) {
      log.debug("Stale report message, skip: appName={}, age={}ms", message.appName(), now - message.timestamp());
      return;
    }

    Map<String, Long> keyCounts = message.counts();
    if (keyCounts.isEmpty()) return;

    // Feed all key counts into the Worker's HeavyKeeper in a single
    // batch call.  This executes sketch updates per-key under fine-grained
    // stripe locks, then acquires the global sortedTopK heap
    // lock exactly ONCE for the entire batch — rather than once per
    // key — eliminating the P1-3 lock-contention bottleneck.
    workerTopK.addDirect(keyCounts);

    List<Map.Entry<String, Long>> entries = new ArrayList<>(keyCounts.entrySet());
    int totalKeys = entries.size();

    for (int chunkStart = 0; chunkStart < totalKeys; chunkStart += CHUNK_SIZE) {
      int chunkEnd = Math.min(chunkStart + CHUNK_SIZE, totalKeys);
      List<Map.Entry<String, Long>> chunk = entries.subList(chunkStart, chunkEnd);

      // Accumulate broadcasts during parallel processing; drain serially
      // after the stream completes to avoid ForkJoin threads blocking on
      // AMQP channel write locks.
      Queue<Runnable> pendingBroadcasts = new ConcurrentLinkedQueue<>();

      // Process each key independently
      chunk
        .parallelStream()
        .forEach(entry -> {
          try {
            String key = entry.getKey();
            long count = entry.getValue();

            totalQps.add(count);

            // addCount atomically increments the current time slice and returns
            // true if the sum of the last windowSize slices exceeds the threshold.
            boolean isHot = detector.addCount(key, count);

            // The state machine tracks consecutive hot/cold windows and applies
            // hysteresis to decide when to transition between COLD, CONFIRMED_HOT
            // and PRE_COOLING.
            HotKeyDecision decision = stateMachine.evaluate(key, isHot);

            switch (decision.type()) {
              case HOT -> {
                // A new hot key has been confirmed. Pre-allocate a decision
                // version and enqueue the broadcast; actual AMQP send happens
                // on the consumer thread after parallelStream completes.
                long dv = broadcaster.nextDecisionVersion();
                pendingBroadcasts.add(() -> broadcaster.broadcastHot(key, SOURCE_SLIDING_WINDOW, dv));
                topKValidator.markConfirmed(key);
              }
              case COOL -> {
                long dv = broadcaster.nextDecisionVersion();
                pendingBroadcasts.add(() -> broadcaster.broadcastCool(key, dv));
                topKValidator.markCooled(key);
              }
              case NONE -> {
                // No state transition occurred – the key remains in its
                // current lifecycle stage.  Nothing to do.
              }
            }
          } catch (Exception e) {
            log.error(
              "Error processing report entry: appName={}, key={}, count={}",
              message.appName(),
              entry.getKey(),
              entry.getValue(),
              e
            );
          }
        });

      // Drain pending broadcasts serially on the consumer thread.
      // This avoids ForkJoinPool threads blocking on AMQP channel write
      // locks under high concurrency (8 concurrent consumers).
      // Per ADR-0007, lost messages are tolerated by the next periodic cycle.
      // sendBroadcast no longer throws — errors are logged and swallowed.
      int drainedCount = 0;
      Runnable task;
      while ((task = pendingBroadcasts.poll()) != null) {
        task.run();
        drainedCount++;
      }

      if (drainedCount >= BATCH_DRAIN_WARN_THRESHOLD) {
        log.info(
          "ReportConsumer drained {} broadcasts for chunk {}/{} of {} keys from app={}",
          drainedCount,
          (chunkStart / CHUNK_SIZE) + 1,
          (totalKeys + CHUNK_SIZE - 1) / CHUNK_SIZE,
          totalKeys,
          message.appName()
        );
      }
    }

    globalQpsEstimator.addTotal(totalQps.sum());
  }
}

1	/*
2	* Copyright 2026 Hyshmily. All Rights Reserved.
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16	package io.github.hyshmily.hotkey.worker.ingest;
17
18	import static io.github.hyshmily.hotkey.constants.HotKeyConstants.SOURCE_SLIDING_WINDOW;
19	import static io.github.hyshmily.hotkey.util.TimeSource.currentTimeMillis;
20
21	import io.github.hyshmily.hotkey.detection.HotKeyStateMachine;
22	import io.github.hyshmily.hotkey.hotkeydetector.heavykeeper.TopK;
23	import io.github.hyshmily.hotkey.model.HotKeyDecision;
24	import io.github.hyshmily.hotkey.reporting.ReportMessage;
25	import io.github.hyshmily.hotkey.worker.detection.GlobalQpsEstimator;
26	import io.github.hyshmily.hotkey.worker.detection.SlidingWindowDetector;
27	import io.github.hyshmily.hotkey.worker.detection.TopKValidator;
28	import io.github.hyshmily.hotkey.worker.dispatch.WorkerBroadcaster;
29	import java.util.ArrayList;
30	import java.util.List;
31	import java.util.Map;
32	import java.util.Queue;
33	import java.util.concurrent.ConcurrentLinkedQueue;
34	import java.util.concurrent.atomic.LongAdder;
35	import lombok.RequiredArgsConstructor;
36	import lombok.extern.slf4j.Slf4j;
37	import org.springframework.amqp.rabbit.annotation.RabbitListener;
38
39	/**
40	* Worker‑side message consumer that receives batched per‑key access counts
41	* reported by application instances.
42	*
43	* <p>For every key in the batch the consumer:
44	* <ol>
45	* <li>Feeds the count into the {@link SlidingWindowDetector} to update the
46	* sliding‑window sum and obtain a binary hot‑or‑not verdict for the
47	* current window.</li>
48	* <li>Passes that verdict to the {@link HotKeyStateMachine} which tracks
49	* consecutive hot/cold windows and decides whether a state transition
50	* (COLD → CONFIRMED_HOT → PRE_COOLING → COLD) has occurred.</li>
51	* <li>If the state machine returns a {@code HOT} decision, the consumer
52	* broadcasts a {@code HOT} message to all application instances and
53	* records the confirmation in the {@link TopKValidator}.</li>
54	* <li>If the state machine returns a {@code COOL} decision, it broadcasts
55	* a {@code COOL} message and marks the key as cooled in the
56	* {@link TopKValidator}.</li>
57	* </ol>
58	*
59	* <p>Messages older than 5 seconds are silently discarded to prevent stale
60	* data from distorting the sliding‑window view.
61	*
62	* <p>Because clients use consistent‑hash routing, every report for a given
63	* key always reaches the same worker, guaranteeing correct per‑key state
64	* without cross‑worker coordination.
65	*/
66	@RequiredArgsConstructor
67	@Slf4j	4✔
68	public class ReportConsumer {
69
70	/** Sliding-window detector that tracks per-key access counts and returns hot/cold verdicts. */
71	private final SlidingWindowDetector detector;
72	/** Per-key lifecycle state machine managing COLD / CONFIRMED_HOT / PRE_COOLING transitions. */
73	private final HotKeyStateMachine stateMachine;
74	/** Publishes HOT and COOL decisions back to all application instances. */
75	private final WorkerBroadcaster broadcaster;
76	/** TopK pre-warm validator for cross-instance frequency-based confirmation. */
77	private final TopKValidator topKValidator;
78	/** Worker-scoped HeavyKeeper sketch for cross-instance frequency estimation. */
79	private final TopK workerTopK;
80	/** Global QPS estimator tracking overall throughput for dynamic threshold learning. */
81	private final GlobalQpsEstimator globalQpsEstimator;
82
83	/** Staleness threshold in milliseconds. Package-visible for testing. */
84	long stalenessThresholdMs = 5000L;
85
86	/** Max keys per chunk for parallel processing. Beyond this, keys are split into chunks. */
87	private static final int CHUNK_SIZE = 1000;
88
89	/** Log a drain-progress summary when the pending broadcast queue exceeds this threshold. */
90	private static final int BATCH_DRAIN_WARN_THRESHOLD = 5000;
91
92	/**
93	* Main entry point for batched report messages.
94	*
95	* @param message the deserialized message containing counts for multiple keys
96	*/
97	@RabbitListener(queues = "#{@reportQueue.name}", containerFactory = "reportListenerContainerFactory")
98	public void onReport(ReportMessage message) {
99	try {
100	doOnReport(message);	3✔
101	} catch (Exception e) {	×
102	log.error(	×
103	"Uncaught exception in onReport, discarding message to prevent poison-message requeue loop: appName={}",
104	message != null ? message.appName() : "null",	×
105	e
106	);
107	}	1✔
108	}	1✔
109
110	private void doOnReport(ReportMessage message) {
111	long now = currentTimeMillis();	2✔
112	LongAdder totalQps = new LongAdder();	4✔
113
114	// Discard reports that are more than 5 seconds old.
115	// Guards against delayed or re‑delivered messages that would
116	// feed outdated counts into the sliding window.
117	if (now - message.timestamp() > stalenessThresholdMs) {	8✔
118	log.debug("Stale report message, skip: appName={}, age={}ms", message.appName(), now - message.timestamp());	10✔
119	return;	1✔
120	}
121
122	Map<String, Long> keyCounts = message.counts();	3✔
123	if (keyCounts.isEmpty()) return;	4✔
124
125	// Feed all key counts into the Worker's HeavyKeeper in a single
126	// batch call. This executes sketch updates per-key under fine-grained
127	// stripe locks, then acquires the global sortedTopK heap
128	// lock exactly ONCE for the entire batch — rather than once per
129	// key — eliminating the P1-3 lock-contention bottleneck.
130	workerTopK.addDirect(keyCounts);	5✔
131
132	List<Map.Entry<String, Long>> entries = new ArrayList<>(keyCounts.entrySet());	6✔
133	int totalKeys = entries.size();	3✔
134
135	for (int chunkStart = 0; chunkStart < totalKeys; chunkStart += CHUNK_SIZE) {	7✔
136	int chunkEnd = Math.min(chunkStart + CHUNK_SIZE, totalKeys);	6✔
137	List<Map.Entry<String, Long>> chunk = entries.subList(chunkStart, chunkEnd);	5✔
138
139	// Accumulate broadcasts during parallel processing; drain serially
140	// after the stream completes to avoid ForkJoin threads blocking on
141	// AMQP channel write locks.
142	Queue<Runnable> pendingBroadcasts = new ConcurrentLinkedQueue<>();	4✔
143
144	// Process each key independently
145	chunk	1✔
146	.parallelStream()	6✔
147	.forEach(entry -> {	1✔
148	try {
149	String key = entry.getKey();	4✔
150	long count = entry.getValue();	5✔
151
152	totalQps.add(count);	3✔
153
154	// addCount atomically increments the current time slice and returns
155	// true if the sum of the last windowSize slices exceeds the threshold.
156	boolean isHot = detector.addCount(key, count);	6✔
157
158	// The state machine tracks consecutive hot/cold windows and applies
159	// hysteresis to decide when to transition between COLD, CONFIRMED_HOT
160	// and PRE_COOLING.
161	HotKeyDecision decision = stateMachine.evaluate(key, isHot);	6✔
162
163	switch (decision.type()) {	6✔
164	case HOT -> {
165	// A new hot key has been confirmed. Pre-allocate a decision
166	// version and enqueue the broadcast; actual AMQP send happens
167	// on the consumer thread after parallelStream completes.
168	long dv = broadcaster.nextDecisionVersion();	4✔
169	pendingBroadcasts.add(() -> broadcaster.broadcastHot(key, SOURCE_SLIDING_WINDOW, dv));	14✔
170	topKValidator.markConfirmed(key);	4✔
171	}	1✔
172	case COOL -> {
173	long dv = broadcaster.nextDecisionVersion();	4✔
174	pendingBroadcasts.add(() -> broadcaster.broadcastCool(key, dv));	13✔
175	topKValidator.markCooled(key);	4✔
176	}	1✔
177	case NONE -> {
178	// No state transition occurred – the key remains in its
179	// current lifecycle stage. Nothing to do.
180	}
181	}
182	} catch (Exception e) {	1✔
183	log.error(	8✔
184	"Error processing report entry: appName={}, key={}, count={}",
185	message.appName(),	5✔
186	entry.getKey(),	5✔
187	entry.getValue(),	6✔
188	e
189	);
190	}	1✔
191	});	1✔
192
193	// Drain pending broadcasts serially on the consumer thread.
194	// This avoids ForkJoinPool threads blocking on AMQP channel write
195	// locks under high concurrency (8 concurrent consumers).
196	// Per ADR-0007, lost messages are tolerated by the next periodic cycle.
197	// sendBroadcast no longer throws — errors are logged and swallowed.
198	int drainedCount = 0;	2✔
199	Runnable task;
200	while ((task = pendingBroadcasts.poll()) != null) {	6✔
201	task.run();	2✔
202	drainedCount++;	2✔
203	}
204
205	if (drainedCount >= BATCH_DRAIN_WARN_THRESHOLD) {	3!
NEW 206	log.info(	×
207	"ReportConsumer drained {} broadcasts for chunk {}/{} of {} keys from app={}",
NEW 208	drainedCount,	×
NEW 209	(chunkStart / CHUNK_SIZE) + 1,	×
NEW 210	(totalKeys + CHUNK_SIZE - 1) / CHUNK_SIZE,	×
NEW 211	totalKeys,	×
NEW 212	message.appName()	×
213	);
214	}
215	}
216
217	globalQpsEstimator.addTotal(totalQps.sum());	5✔
218	}	1✔
219	}

Hyshmily / hotkey / 28503055923

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous