27952839223

Committed 22 Jun 2026 12:31PM UTC coverage: 93.191% (-0.08%) from 93.275%

Build # 27952839223

Build Type

push

github

Committed by

Hyshmily

Commit Message

fix: revert hotKeyExecutor to ThreadPoolTaskExecutor (virtual threads cause AMQP channel-open timeouts)

Coverage Stats

1163 of 1289 branches covered (90.22%)

Branch coverage included in aggregate %.

10 of 15 new or added lines in 1 file covered. (66.67%)

3340 of 3543 relevant lines covered (94.27%)

4.24 hits per line

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

64.06

/common/src/main/java/io/github/hyshmily/hotkey/autoconfigure/HotKeyAutoConfiguration.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.autoconfigure;

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Expiry;
import io.github.hyshmily.hotkey.HotKey;
import io.github.hyshmily.hotkey.cache.CacheExpireManager;
import io.github.hyshmily.hotkey.cache.HotKeyCache;
import io.github.hyshmily.hotkey.cache.SingleFlight;
import io.github.hyshmily.hotkey.constants.HotKeyConstants;
import io.github.hyshmily.hotkey.hotkeydetector.HotKeyDetector;
import io.github.hyshmily.hotkey.hotkeydetector.heavykeeper.HeavyKeeper;
import io.github.hyshmily.hotkey.hotkeydetector.heavykeeper.TopK;
import io.github.hyshmily.hotkey.model.CacheEntry;
import io.github.hyshmily.hotkey.reporting.HotKeyReporter;
import io.github.hyshmily.hotkey.rule.RuleMatcher;
import io.github.hyshmily.hotkey.sharding.RingManager;
import io.github.hyshmily.hotkey.sharding.ClusterHealthView;
import io.github.hyshmily.hotkey.sync.local.CacheSyncPublisher;
import io.github.hyshmily.hotkey.util.version.VersionController;
import java.util.Optional;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.jspecify.annotations.NonNull;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

/**
 * App-side autoconfiguration for the HotKey library.
 *
 * <p>Creates the app-side {@link TopK} detector (HeavyKeeper), L1 Caffeine
 * cache, {@link SingleFlight} deduplication layer, executor, and the
 * primary {@link HotKeyCache} (without Redis version tracking when Redis
 * is absent).
 *
 * <p><b>Condition:</b> this configuration is <em>skipped</em> when the
 * Worker is active ({@code hotkey.worker.enabled=true}). It runs when
 * Worker is disabled or the property is absent.
 */
@AutoConfiguration(after = RedisAutoConfiguration.class)
@ConditionalOnProperty(prefix = "hotkey.worker", name = "enabled", havingValue = "false", matchIfMissing = true)
@EnableConfigurationProperties(HotKeyProperties.class)
@Slf4j
public class HotKeyAutoConfiguration {

  /**
   * Create the app-side TopK instance (HeavyKeeper) as a standalone bean.
   *
   * <p>The HeavyKeeper uses a Count-Min Sketch augmented with a minimum-count
   * threshold and exponential decay to track the top-K hottest keys with high
   * accuracy and low memory footprint. Configuration parameters (width, depth,
   * decay, minCount, topK capacity, expelled queue capacity) are read from
   * {@link HotKeyProperties}.
   *
   * @param properties the HotKey configuration properties (never {@code null})
   * @return a new HeavyKeeper TopK instance
   */
  @Bean
  @ConditionalOnMissingBean
  public HeavyKeeper heavyKeeper(HotKeyProperties properties) {
    return new HeavyKeeper(
      properties.getTopK(),
      properties.getWidth(),
      properties.getDepth(),
      properties.getDecay(),
      properties.getMinCount(),
      properties.getExpelledQueueCapacity()
    );
  }

  /**
   * Create the app-side {@link HotKeyDetector} facade that wraps the HeavyKeeper TopK
   * and schedules periodic decay.
   *
   * <p>The detector provides the primary hot-key detection API ({@code add()}, {@code list()},
   * {@code total()}) and schedules the periodic HeavyKeeper decay using the shared scheduler.
   * This bean is the {@code @Qualifier("hotKeyDetector")} target for injection into
   * {@link HotKeyCache} and other components.
   *
   * @param heavyKeeper      the HeavyKeeper TopK instance (never {@code null})
   * @param hotKeyScheduler  the shared scheduler for periodic tasks (never {@code null})
   * @return a new HotKeyDetector instance
   */
  @Bean
  @ConditionalOnMissingBean
  @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
  public HotKeyDetector hotKeyDetector(
    HeavyKeeper heavyKeeper,
    @Qualifier("hotKeyScheduler") ScheduledExecutorService hotKeyScheduler
  ) {
    return new HotKeyDetector(heavyKeeper, hotKeyScheduler);
  }

  /**
   * Create the SingleFlight deduplication layer for concurrent cache-load requests.
   *
   * <p>When multiple threads request the same key simultaneously (a cache miss), the
   * first caller triggers the load and subsequent callers wait for the same result
   * rather than duplicating the load. Configuration parameters (max in-flight entries,
   * TTL, timeout) are read from {@link HotKeyProperties}.
   *
   * @param properties     the HotKey configuration properties (never {@code null})
   * @param hotKeyExecutor the dedicated HotKey executor for async load execution (never {@code null})
   * @return a new SingleFlight instance
   */
  @Bean
  @ConditionalOnMissingBean
  public SingleFlight singleFlight(HotKeyProperties properties, @Qualifier("hotKeyExecutor") Executor hotKeyExecutor) {
    return new SingleFlight(
      properties.getInflightMaxSize(),
      properties.getInflightTtlSeconds(),
      properties.getInflightTimeoutSeconds(),
      hotKeyExecutor
    );
  }

  /**
   * Create the soft/hard expiration manager that manages time-based eviction.
   *
   * <p>The soft-expire mechanism triggers an asynchronous refresh when a configurable
   * portion of the TTL has elapsed, serving stale data while fetching a fresh value
   * in the background. The hard-expire is the absolute maximum TTL enforced at the
   * Caffeine level. The refresh pool size limits concurrent background refresh tasks
   * to prevent resource exhaustion under high cache-miss rates.
   *
   * @param hotLocalCache  the L1 Caffeine cache (never {@code null})
   * @param hotKeyExecutor the dedicated HotKey executor for async refresh tasks (never {@code null})
   * @param properties     the HotKey configuration properties (never {@code null})
   * @return a new CacheExpireManager instance
   */
  @Bean
  @ConditionalOnMissingBean
  public CacheExpireManager expireManager(
    Cache<String, Object> hotLocalCache,
    @Qualifier("hotKeyExecutor") Executor hotKeyExecutor,
    HotKeyProperties properties
  ) {
    return new CacheExpireManager(hotLocalCache, hotKeyExecutor, properties, properties.getRefreshMaxPools());
  }

  /**
   * Create the dedicated thread-pool executor for asynchronous cache operations.
   *
   * <p>This executor handles all async operations in the HotKey data path: cache loading
   * via SingleFlight, soft-expiry refresh tasks, and cross-instance broadcast callbacks.
   * Uses a bounded thread pool to limit concurrent AMQP channel usage (RabbitMQ's
   * {@code CachingConnectionFactory} associates channels with platform threads, so
   * unbounded virtual-thread concurrency causes channel-open timeouts). The pool is
   * configured with core/max pool size, bounded queue capacity, and a rejection policy
   * that throws {@link RejectedExecutionException} when the queue is full. On shutdown,
   * in-progress tasks are allowed to complete with a 60-second grace period.
   *
   * @param properties the HotKey configuration properties (never {@code null})
   * @return a configured {@link ThreadPoolTaskExecutor}
   */
  @Bean("hotKeyExecutor")
  @ConditionalOnMissingBean(name = "hotKeyExecutor")
  public Executor hotKeyExecutor(HotKeyProperties properties) {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(properties.getExecutorCorePoolSize());
    executor.setMaxPoolSize(properties.getExecutorMaxPoolSize());
    executor.setQueueCapacity(properties.getExecutorQueueCapacity());
    executor.setThreadNamePrefix(HotKeyConstants.THREAD_PREFIX_HOTKEY);
    executor.setWaitForTasksToCompleteOnShutdown(true);
    executor.setAwaitTerminationSeconds(60);
    executor.setRejectedExecutionHandler((r, exe) -> {
      log.warn(
        "HotKey executor task rejected: corePool={}, maxPool={}, queueCapacity={}",
        properties.getExecutorCorePoolSize(),
        properties.getExecutorMaxPoolSize(),
        properties.getExecutorQueueCapacity()
      );
      throw new RejectedExecutionException("HotKey executor queue full");
    });
    executor.initialize();
    return executor;
  }

  /**
   * Create the {@link RuleMatcher} for key matching against user-defined rules.
   *
   * <p>This variant is wired with an empty Redis provider and an optional sync
   * publisher, since no {@link StringRedisTemplate} is available in the non-Redis
   * deployment mode. Rules are kept in-memory only and do not survive restarts.
   * When Redis is available, {@link HotKeyRedisAutoConfiguration#ruleMatcher}
   * takes over with Redis persistence.
   *
   * @param publisherProvider optional provider for the cache sync publisher (may be absent)
   * @return a new RuleMatcher instance (in-memory only)
   */
  @Bean
  @ConditionalOnMissingBean({ RuleMatcher.class, StringRedisTemplate.class })
  public RuleMatcher ruleMatcher(ObjectProvider<CacheSyncPublisher> publisherProvider) {
    return new RuleMatcher(Optional.empty(), Optional.ofNullable(publisherProvider.getIfAvailable()));
  }

  /**
   * Create the {@link HotKeyCache} (non-Redis variant).
   *
   * <p>Only active when {@code RedisTemplate} is absent; otherwise
   * {@link HotKeyRedisAutoConfiguration#hotKeyCache} takes over. Creates a
   * {@link VersionController} with an empty Redis provider (falling back to
   * node-local counters) and default ring manager / health view when none are
   * available from the application context.
   *
   * @param hotKeyDetector            the app-side TopK detector (never {@code null})
   * @param hotLocalCache             the L1 Caffeine cache (never {@code null})
   * @param singleFlight              the deduplication layer (never {@code null})
   * @param expireManager             the soft/hard expiration manager (never {@code null})
   * @param syncPublisher             optional cache sync publisher (may be absent)
   * @param hotKeyReporter            optional hot key reporter (may be absent; e.g. in Worker-only mode)
   * @param hotKeyExecutor            the dedicated HotKey executor (never {@code null})
   * @param properties                the HotKey configuration properties (never {@code null})
   * @param ruleMatcher               the rule matcher instance (never {@code null})
   * @param ringManagerProvider       provider for the consistent-hash ring manager (creates default if absent)
   * @param healthViewProvider        provider for the cluster health view (creates default if absent)
   * @return a new HotKeyCache instance with node-local version tracking
   */
  @Bean
  @ConditionalOnMissingBean(type = "org.springframework.data.redis.core.RedisTemplate")
  public HotKeyCache hotKeyCache(
    @Qualifier("hotKeyDetector") HotKeyDetector hotKeyDetector,
    Cache<String, Object> hotLocalCache,
    SingleFlight singleFlight,
    CacheExpireManager expireManager,
    Optional<CacheSyncPublisher> syncPublisher,
    Optional<HotKeyReporter> hotKeyReporter,
    @Qualifier("hotKeyExecutor") Executor hotKeyExecutor,
    HotKeyProperties properties,
    RuleMatcher ruleMatcher,
    ObjectProvider<RingManager> ringManagerProvider,
    ObjectProvider<ClusterHealthView> healthViewProvider
  ) {
    return new HotKeyCache(
      hotKeyDetector,
      hotLocalCache,
      singleFlight,
      expireManager,
      hotKeyExecutor,
      syncPublisher,
      hotKeyReporter,
      ruleMatcher,
      new VersionController(Optional.empty(), properties.getVersionKeyTtlMinutes()),
      ringManagerProvider.getIfAvailable(() -> new RingManager(properties.getConsistentHashing().getVirtualNodes())),
      new ClusterHealthView(
        properties.getExpectedWorkerCount(),
        properties.getHeartbeat().getTimeoutMs(),
        properties.getHeartbeat().getDegradeAfterFailures()
      )
    );
  }

  /**
   * Fallback {@link HotKey} facade bean.
   *
   * <p>Only active when a {@link HotKeyCache} bean exists but no {@link HotKey}
   * has been defined yet. The primary {@link HotKey} creator is
   * {@link HotKeyFacadeAutoConfiguration} — this fallback covers the case where
   * that auto-configuration is excluded or its bean is overridden.
   *
   * @param hotKeyCache    the HotKeyCache instance (never {@code null})
   * @param appHotKeyDetector the app-side TopK detector (never {@code null})
   * @param workerTopKProvider     the Worker-side TopK algorithm (may be {@code null})
   * @return a new HotKey facade instance
   */
  @Bean
  @ConditionalOnBean(HotKeyCache.class)
  @ConditionalOnMissingBean
  @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
  public HotKey hotKey(
    HotKeyCache hotKeyCache,
    @Qualifier("hotKeyDetector") HotKeyDetector appHotKeyDetector,
    @Qualifier("workerTopK") ObjectProvider<TopK> workerTopKProvider
  ) {
    return new HotKey(hotKeyCache, appHotKeyDetector, workerTopKProvider.getIfAvailable());
  }

  /**
   * Create the L1 Caffeine cache instance.
   *
   * <p>Time-based expiry operates at the <em>Caffeine</em> level via a custom
   * {@link Expiry} implementation, computing remaining nanoseconds from
   * . Entries with
   * {@code hardExpireAtMs == Long.MAX_VALUE} are purely logical-expiry
   * — Caffeine never evicts them by time; they live until size eviction or
   * manual invalidation. Reads never extend the expiry duration (no read-based
   * refresh), ensuring predictable TTL behavior.
   *
   * <p>The maximum cache size is configured via {@code hotkey.local.local-cache-max-size}
   * (default 1000 entries). Time-based TTL for entries without an explicit hard-expire
   * timestamp defaults to {@code hotkey.local.local-cache-ttl-minutes} (default 5 minutes).
   *
   * @param properties the HotKey configuration properties (never {@code null})
   * @return a configured Caffeine {@link Cache} instance
   */
  @Bean
  @ConditionalOnMissingBean
  public Cache<String, Object> hotLocalCache(HotKeyProperties properties) {
    Caffeine<Object, Object> builder = Caffeine.newBuilder()
      .maximumSize(properties.getLocalCacheMaxSize())
      .expireAfter(
        new Expiry<>() {
          /**
           * Compute the time-to-live for a newly created cache entry.
           * Returns {@link Long#MAX_VALUE} for pure logical-expiry entries
           * (where {@code hardExpireAtMs == Long.MAX_VALUE}); otherwise
           * computes the remaining wall-clock time.
           *
           * @param key              the cache key
           * @param value            the cache value (expected to be a {@link CacheEntry})
           * @param currentTimeNanos the current time in nanoseconds (provided by Caffeine)
           * @return the expiry duration in nanoseconds, or {@link Long#MAX_VALUE} for no expiry
           */
          @Override
          public long expireAfterCreate(@NonNull Object key, @NonNull Object value, long currentTimeNanos) {
            if (value instanceof CacheEntry entry) {
              if (entry.getHardExpireAtMs() == Long.MAX_VALUE) {
                // Pure logical expiry: Caffeine never time-evicts this entry.
                // See Expiry Javadoc: Long.MAX_VALUE signals "no expiration".
                return Long.MAX_VALUE;
              }
              long remainingMs = entry.getHardExpireAtMs() - System.currentTimeMillis();
              return TimeUnit.MILLISECONDS.toNanos(Math.max(1, remainingMs));
            }
            return TimeUnit.MINUTES.toNanos(properties.getLocalCacheTtlMinutes());
          }

          /**
           * Re-compute the expiry duration after an entry is updated.
           * Preserves pure logical expiry across updates; otherwise
           * recalculates from the entry's {@code hardExpireAtMs}.
           *
           * @param key              the cache key
           * @param value            the cache value (expected to be a {@link CacheEntry})
           * @param currentTimeNanos the current time in nanoseconds (provided by Caffeine)
           * @param currentDuration  the current expiry duration in nanoseconds
           * @return the updated expiry duration in nanoseconds, or {@link Long#MAX_VALUE} for no expiry
           */
          @Override
          public long expireAfterUpdate(
            @NonNull Object key,
            @NonNull Object value,
            long currentTimeNanos,
            long currentDuration
          ) {
            if (value instanceof CacheEntry entry) {
              if (entry.getHardExpireAtMs() == Long.MAX_VALUE) {
                // Preserve pure logical expiry across updates (e.g. broadcast refresh).
                return Long.MAX_VALUE;
              }
              long remainingMs = entry.getHardExpireAtMs() - System.currentTimeMillis();
              return TimeUnit.MILLISECONDS.toNanos(Math.max(1, remainingMs));
            }
            return currentDuration;
          }

          /**
           * Preserve the current expiry duration on read — reads never
           * extend or shorten the entry's time-to-live.
           *
           * @param key              the cache key
           * @param value            the cache value
           * @param currentTimeNanos the current time in nanoseconds (provided by Caffeine)
           * @param currentDuration  the current expiry duration in nanoseconds
           * @return the unchanged expiry duration in nanoseconds
           */
          @Override
          public long expireAfterRead(
            @NonNull Object key,
            @NonNull Object value,
            long currentTimeNanos,
            long currentDuration
          ) {
            return currentDuration;
          }
        }
      );
    return builder.build();
  }
}

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.autoconfigure;
17
18	import com.github.benmanes.caffeine.cache.Cache;
19	import com.github.benmanes.caffeine.cache.Caffeine;
20	import com.github.benmanes.caffeine.cache.Expiry;
21	import io.github.hyshmily.hotkey.HotKey;
22	import io.github.hyshmily.hotkey.cache.CacheExpireManager;
23	import io.github.hyshmily.hotkey.cache.HotKeyCache;
24	import io.github.hyshmily.hotkey.cache.SingleFlight;
25	import io.github.hyshmily.hotkey.constants.HotKeyConstants;
26	import io.github.hyshmily.hotkey.hotkeydetector.HotKeyDetector;
27	import io.github.hyshmily.hotkey.hotkeydetector.heavykeeper.HeavyKeeper;
28	import io.github.hyshmily.hotkey.hotkeydetector.heavykeeper.TopK;
29	import io.github.hyshmily.hotkey.model.CacheEntry;
30	import io.github.hyshmily.hotkey.reporting.HotKeyReporter;
31	import io.github.hyshmily.hotkey.rule.RuleMatcher;
32	import io.github.hyshmily.hotkey.sharding.RingManager;
33	import io.github.hyshmily.hotkey.sharding.ClusterHealthView;
34	import io.github.hyshmily.hotkey.sync.local.CacheSyncPublisher;
35	import io.github.hyshmily.hotkey.util.version.VersionController;
36	import java.util.Optional;
37	import java.util.concurrent.Executor;
38	import java.util.concurrent.RejectedExecutionException;
39	import java.util.concurrent.ScheduledExecutorService;
40	import java.util.concurrent.TimeUnit;
41	import lombok.extern.slf4j.Slf4j;
42	import org.jspecify.annotations.NonNull;
43	import org.springframework.beans.factory.ObjectProvider;
44	import org.springframework.beans.factory.annotation.Qualifier;
45	import org.springframework.boot.autoconfigure.AutoConfiguration;
46	import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
47	import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
48	import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
49	import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
50	import org.springframework.boot.context.properties.EnableConfigurationProperties;
51	import org.springframework.context.annotation.Bean;
52	import org.springframework.data.redis.core.StringRedisTemplate;
53	import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
54
55	/**
56	* App-side autoconfiguration for the HotKey library.
57	*
58	* <p>Creates the app-side {@link TopK} detector (HeavyKeeper), L1 Caffeine
59	* cache, {@link SingleFlight} deduplication layer, executor, and the
60	* primary {@link HotKeyCache} (without Redis version tracking when Redis
61	* is absent).
62	*
63	* <p><b>Condition:</b> this configuration is <em>skipped</em> when the
64	* Worker is active ({@code hotkey.worker.enabled=true}). It runs when
65	* Worker is disabled or the property is absent.
66	*/
67	@AutoConfiguration(after = RedisAutoConfiguration.class)
68	@ConditionalOnProperty(prefix = "hotkey.worker", name = "enabled", havingValue = "false", matchIfMissing = true)
69	@EnableConfigurationProperties(HotKeyProperties.class)
70	@Slf4j	4✔
71	public class HotKeyAutoConfiguration {	3✔
72
73	/**
74	* Create the app-side TopK instance (HeavyKeeper) as a standalone bean.
75	*
76	* <p>The HeavyKeeper uses a Count-Min Sketch augmented with a minimum-count
77	* threshold and exponential decay to track the top-K hottest keys with high
78	* accuracy and low memory footprint. Configuration parameters (width, depth,
79	* decay, minCount, topK capacity, expelled queue capacity) are read from
80	* {@link HotKeyProperties}.
81	*
82	* @param properties the HotKey configuration properties (never {@code null})
83	* @return a new HeavyKeeper TopK instance
84	*/
85	@Bean
86	@ConditionalOnMissingBean
87	public HeavyKeeper heavyKeeper(HotKeyProperties properties) {
88	return new HeavyKeeper(	4✔
89	properties.getTopK(),	2✔
90	properties.getWidth(),	2✔
91	properties.getDepth(),	2✔
92	properties.getDecay(),	2✔
93	properties.getMinCount(),	2✔
94	properties.getExpelledQueueCapacity()	2✔
95	);
96	}
97
98	/**
99	* Create the app-side {@link HotKeyDetector} facade that wraps the HeavyKeeper TopK
100	* and schedules periodic decay.
101	*
102	* <p>The detector provides the primary hot-key detection API ({@code add()}, {@code list()},
103	* {@code total()}) and schedules the periodic HeavyKeeper decay using the shared scheduler.
104	* This bean is the {@code @Qualifier("hotKeyDetector")} target for injection into
105	* {@link HotKeyCache} and other components.
106	*
107	* @param heavyKeeper the HeavyKeeper TopK instance (never {@code null})
108	* @param hotKeyScheduler the shared scheduler for periodic tasks (never {@code null})
109	* @return a new HotKeyDetector instance
110	*/
111	@Bean
112	@ConditionalOnMissingBean
113	@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
114	public HotKeyDetector hotKeyDetector(
115	HeavyKeeper heavyKeeper,
116	@Qualifier("hotKeyScheduler") ScheduledExecutorService hotKeyScheduler
117	) {
118	return new HotKeyDetector(heavyKeeper, hotKeyScheduler);	6✔
119	}
120
121	/**
122	* Create the SingleFlight deduplication layer for concurrent cache-load requests.
123	*
124	* <p>When multiple threads request the same key simultaneously (a cache miss), the
125	* first caller triggers the load and subsequent callers wait for the same result
126	* rather than duplicating the load. Configuration parameters (max in-flight entries,
127	* TTL, timeout) are read from {@link HotKeyProperties}.
128	*
129	* @param properties the HotKey configuration properties (never {@code null})
130	* @param hotKeyExecutor the dedicated HotKey executor for async load execution (never {@code null})
131	* @return a new SingleFlight instance
132	*/
133	@Bean
134	@ConditionalOnMissingBean
135	public SingleFlight singleFlight(HotKeyProperties properties, @Qualifier("hotKeyExecutor") Executor hotKeyExecutor) {
136	return new SingleFlight(	4✔
137	properties.getInflightMaxSize(),	2✔
138	properties.getInflightTtlSeconds(),	2✔
139	properties.getInflightTimeoutSeconds(),	3✔
140	hotKeyExecutor
141	);
142	}
143
144	/**
145	* Create the soft/hard expiration manager that manages time-based eviction.
146	*
147	* <p>The soft-expire mechanism triggers an asynchronous refresh when a configurable
148	* portion of the TTL has elapsed, serving stale data while fetching a fresh value
149	* in the background. The hard-expire is the absolute maximum TTL enforced at the
150	* Caffeine level. The refresh pool size limits concurrent background refresh tasks
151	* to prevent resource exhaustion under high cache-miss rates.
152	*
153	* @param hotLocalCache the L1 Caffeine cache (never {@code null})
154	* @param hotKeyExecutor the dedicated HotKey executor for async refresh tasks (never {@code null})
155	* @param properties the HotKey configuration properties (never {@code null})
156	* @return a new CacheExpireManager instance
157	*/
158	@Bean
159	@ConditionalOnMissingBean
160	public CacheExpireManager expireManager(
161	Cache<String, Object> hotLocalCache,
162	@Qualifier("hotKeyExecutor") Executor hotKeyExecutor,
163	HotKeyProperties properties
164	) {
165	return new CacheExpireManager(hotLocalCache, hotKeyExecutor, properties, properties.getRefreshMaxPools());	9✔
166	}
167
168	/**
169	* Create the dedicated thread-pool executor for asynchronous cache operations.
170	*
171	* <p>This executor handles all async operations in the HotKey data path: cache loading
172	* via SingleFlight, soft-expiry refresh tasks, and cross-instance broadcast callbacks.
173	* Uses a bounded thread pool to limit concurrent AMQP channel usage (RabbitMQ's
174	* {@code CachingConnectionFactory} associates channels with platform threads, so
175	* unbounded virtual-thread concurrency causes channel-open timeouts). The pool is
176	* configured with core/max pool size, bounded queue capacity, and a rejection policy
177	* that throws {@link RejectedExecutionException} when the queue is full. On shutdown,
178	* in-progress tasks are allowed to complete with a 60-second grace period.
179	*
180	* @param properties the HotKey configuration properties (never {@code null})
181	* @return a configured {@link ThreadPoolTaskExecutor}
182	*/
183	@Bean("hotKeyExecutor")
184	@ConditionalOnMissingBean(name = "hotKeyExecutor")
185	public Executor hotKeyExecutor(HotKeyProperties properties) {
186	ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();	4✔
187	executor.setCorePoolSize(properties.getExecutorCorePoolSize());	4✔
188	executor.setMaxPoolSize(properties.getExecutorMaxPoolSize());	4✔
189	executor.setQueueCapacity(properties.getExecutorQueueCapacity());	4✔
190	executor.setThreadNamePrefix(HotKeyConstants.THREAD_PREFIX_HOTKEY);	3✔
191	executor.setWaitForTasksToCompleteOnShutdown(true);	3✔
192	executor.setAwaitTerminationSeconds(60);	3✔
193	executor.setRejectedExecutionHandler((r, exe) -> {	4✔
NEW 194	log.warn(	×
195	"HotKey executor task rejected: corePool={}, maxPool={}, queueCapacity={}",
NEW 196	properties.getExecutorCorePoolSize(),	×
NEW 197	properties.getExecutorMaxPoolSize(),	×
NEW 198	properties.getExecutorQueueCapacity()	×
199	);
NEW 200	throw new RejectedExecutionException("HotKey executor queue full");	×
201	});
202	executor.initialize();	2✔
203	return executor;	2✔
204	}
205
206	/**
207	* Create the {@link RuleMatcher} for key matching against user-defined rules.
208	*
209	* <p>This variant is wired with an empty Redis provider and an optional sync
210	* publisher, since no {@link StringRedisTemplate} is available in the non-Redis
211	* deployment mode. Rules are kept in-memory only and do not survive restarts.
212	* When Redis is available, {@link HotKeyRedisAutoConfiguration#ruleMatcher}
213	* takes over with Redis persistence.
214	*
215	* @param publisherProvider optional provider for the cache sync publisher (may be absent)
216	* @return a new RuleMatcher instance (in-memory only)
217	*/
218	@Bean
219	@ConditionalOnMissingBean({ RuleMatcher.class, StringRedisTemplate.class })
220	public RuleMatcher ruleMatcher(ObjectProvider<CacheSyncPublisher> publisherProvider) {
221	return new RuleMatcher(Optional.empty(), Optional.ofNullable(publisherProvider.getIfAvailable()));	9✔
222	}
223
224	/**
225	* Create the {@link HotKeyCache} (non-Redis variant).
226	*
227	* <p>Only active when {@code RedisTemplate} is absent; otherwise
228	* {@link HotKeyRedisAutoConfiguration#hotKeyCache} takes over. Creates a
229	* {@link VersionController} with an empty Redis provider (falling back to
230	* node-local counters) and default ring manager / health view when none are
231	* available from the application context.
232	*
233	* @param hotKeyDetector the app-side TopK detector (never {@code null})
234	* @param hotLocalCache the L1 Caffeine cache (never {@code null})
235	* @param singleFlight the deduplication layer (never {@code null})
236	* @param expireManager the soft/hard expiration manager (never {@code null})
237	* @param syncPublisher optional cache sync publisher (may be absent)
238	* @param hotKeyReporter optional hot key reporter (may be absent; e.g. in Worker-only mode)
239	* @param hotKeyExecutor the dedicated HotKey executor (never {@code null})
240	* @param properties the HotKey configuration properties (never {@code null})
241	* @param ruleMatcher the rule matcher instance (never {@code null})
242	* @param ringManagerProvider provider for the consistent-hash ring manager (creates default if absent)
243	* @param healthViewProvider provider for the cluster health view (creates default if absent)
244	* @return a new HotKeyCache instance with node-local version tracking
245	*/
246	@Bean
247	@ConditionalOnMissingBean(type = "org.springframework.data.redis.core.RedisTemplate")
248	public HotKeyCache hotKeyCache(
249	@Qualifier("hotKeyDetector") HotKeyDetector hotKeyDetector,
250	Cache<String, Object> hotLocalCache,
251	SingleFlight singleFlight,
252	CacheExpireManager expireManager,
253	Optional<CacheSyncPublisher> syncPublisher,
254	Optional<HotKeyReporter> hotKeyReporter,
255	@Qualifier("hotKeyExecutor") Executor hotKeyExecutor,
256	HotKeyProperties properties,
257	RuleMatcher ruleMatcher,
258	ObjectProvider<RingManager> ringManagerProvider,
259	ObjectProvider<ClusterHealthView> healthViewProvider
260	) {
261	return new HotKeyCache(	13✔
262	hotKeyDetector,
263	hotLocalCache,
264	singleFlight,
265	expireManager,
266	hotKeyExecutor,
267	syncPublisher,
268	hotKeyReporter,
269	ruleMatcher,
270	new VersionController(Optional.empty(), properties.getVersionKeyTtlMinutes()),	7✔
271	ringManagerProvider.getIfAvailable(() -> new RingManager(properties.getConsistentHashing().getVirtualNodes())),	12✔
272	new ClusterHealthView(
273	properties.getExpectedWorkerCount(),	2✔
274	properties.getHeartbeat().getTimeoutMs(),	4✔
275	properties.getHeartbeat().getDegradeAfterFailures()	4✔
276	)
277	);
278	}
279
280	/**
281	* Fallback {@link HotKey} facade bean.
282	*
283	* <p>Only active when a {@link HotKeyCache} bean exists but no {@link HotKey}
284	* has been defined yet. The primary {@link HotKey} creator is
285	* {@link HotKeyFacadeAutoConfiguration} — this fallback covers the case where
286	* that auto-configuration is excluded or its bean is overridden.
287	*
288	* @param hotKeyCache the HotKeyCache instance (never {@code null})
289	* @param appHotKeyDetector the app-side TopK detector (never {@code null})
290	* @param workerTopKProvider the Worker-side TopK algorithm (may be {@code null})
291	* @return a new HotKey facade instance
292	*/
293	@Bean
294	@ConditionalOnBean(HotKeyCache.class)
295	@ConditionalOnMissingBean
296	@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
297	public HotKey hotKey(
298	HotKeyCache hotKeyCache,
299	@Qualifier("hotKeyDetector") HotKeyDetector appHotKeyDetector,
300	@Qualifier("workerTopK") ObjectProvider<TopK> workerTopKProvider
301	) {
302	return new HotKey(hotKeyCache, appHotKeyDetector, workerTopKProvider.getIfAvailable());	9✔
303	}
304
305	/**
306	* Create the L1 Caffeine cache instance.
307	*
308	* <p>Time-based expiry operates at the <em>Caffeine</em> level via a custom
309	* {@link Expiry} implementation, computing remaining nanoseconds from
310	* . Entries with
311	* {@code hardExpireAtMs == Long.MAX_VALUE} are purely logical-expiry
312	* — Caffeine never evicts them by time; they live until size eviction or
313	* manual invalidation. Reads never extend the expiry duration (no read-based
314	* refresh), ensuring predictable TTL behavior.
315	*
316	* <p>The maximum cache size is configured via {@code hotkey.local.local-cache-max-size}
317	* (default 1000 entries). Time-based TTL for entries without an explicit hard-expire
318	* timestamp defaults to {@code hotkey.local.local-cache-ttl-minutes} (default 5 minutes).
319	*
320	* @param properties the HotKey configuration properties (never {@code null})
321	* @return a configured Caffeine {@link Cache} instance
322	*/
323	@Bean
324	@ConditionalOnMissingBean
325	public Cache<String, Object> hotLocalCache(HotKeyProperties properties) {
326	Caffeine<Object, Object> builder = Caffeine.newBuilder()	2✔
327	.maximumSize(properties.getLocalCacheMaxSize())	8✔
328	.expireAfter(	2✔
329	new Expiry<>() {	6✔
330	/**
331	* Compute the time-to-live for a newly created cache entry.
332	* Returns {@link Long#MAX_VALUE} for pure logical-expiry entries
333	* (where {@code hardExpireAtMs == Long.MAX_VALUE}); otherwise
334	* computes the remaining wall-clock time.
335	*
336	* @param key the cache key
337	* @param value the cache value (expected to be a {@link CacheEntry})
338	* @param currentTimeNanos the current time in nanoseconds (provided by Caffeine)
339	* @return the expiry duration in nanoseconds, or {@link Long#MAX_VALUE} for no expiry
340	*/
341	@Override
342	public long expireAfterCreate(@NonNull Object key, @NonNull Object value, long currentTimeNanos) {
343	if (value instanceof CacheEntry entry) {	3!
344	if (entry.getHardExpireAtMs() == Long.MAX_VALUE) {	×
345	// Pure logical expiry: Caffeine never time-evicts this entry.
346	// See Expiry Javadoc: Long.MAX_VALUE signals "no expiration".
347	return Long.MAX_VALUE;	×
348	}
349	long remainingMs = entry.getHardExpireAtMs() - System.currentTimeMillis();	×
350	return TimeUnit.MILLISECONDS.toNanos(Math.max(1, remainingMs));	×
351	}
352	return TimeUnit.MINUTES.toNanos(properties.getLocalCacheTtlMinutes());	7✔
353	}
354
355	/**
356	* Re-compute the expiry duration after an entry is updated.
357	* Preserves pure logical expiry across updates; otherwise
358	* recalculates from the entry's {@code hardExpireAtMs}.
359	*
360	* @param key the cache key
361	* @param value the cache value (expected to be a {@link CacheEntry})
362	* @param currentTimeNanos the current time in nanoseconds (provided by Caffeine)
363	* @param currentDuration the current expiry duration in nanoseconds
364	* @return the updated expiry duration in nanoseconds, or {@link Long#MAX_VALUE} for no expiry
365	*/
366	@Override
367	public long expireAfterUpdate(
368	@NonNull Object key,
369	@NonNull Object value,
370	long currentTimeNanos,
371	long currentDuration
372	) {
373	if (value instanceof CacheEntry entry) {	×
374	if (entry.getHardExpireAtMs() == Long.MAX_VALUE) {	×
375	// Preserve pure logical expiry across updates (e.g. broadcast refresh).
376	return Long.MAX_VALUE;	×
377	}
378	long remainingMs = entry.getHardExpireAtMs() - System.currentTimeMillis();	×
379	return TimeUnit.MILLISECONDS.toNanos(Math.max(1, remainingMs));	×
380	}
381	return currentDuration;	×
382	}
383
384	/**
385	* Preserve the current expiry duration on read — reads never
386	* extend or shorten the entry's time-to-live.
387	*
388	* @param key the cache key
389	* @param value the cache value
390	* @param currentTimeNanos the current time in nanoseconds (provided by Caffeine)
391	* @param currentDuration the current expiry duration in nanoseconds
392	* @return the unchanged expiry duration in nanoseconds
393	*/
394	@Override
395	public long expireAfterRead(
396	@NonNull Object key,
397	@NonNull Object value,
398	long currentTimeNanos,
399	long currentDuration
400	) {
401	return currentDuration;	×
402	}
403	}
404	);
405	return builder.build();	3✔
406	}
407	}

Hyshmily / hotkey / 27952839223

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