#5612

Committed 11 Jul 2026 03:47PM UTC coverage: 71.292% (-28.7%) from 100.0%

Build # #5612

Build Type

push

github

Committed by

ben-manes

Commit Message

Preserve timestamps on a raced same-instance refresh

The automatic refreshIfNeeded path checked for an equal-value reload
before the ABA/ownership check and returned without preserving the
timestamps, so a reload completing after a concurrent put(k,
sameInstance) did a full update — shifting the entry's expiration
forward by the reload's in-flight duration and stomping the write.
Mirror the explicit-refresh fix: check ownership first. An owned
reload installs the value (refreshing metadata even for a same
instance), while a raced reload preserves the write's timestamps and
only notifies REPLACED when the value differs.

Coverage Stats

2957 of 4134 branches covered (71.53%)

3 of 3 new or added lines in 1 file covered. (100.0%)

2408 existing lines in 53 files now uncovered.

5980 of 8388 relevant lines covered (71.29%)

0.71 hits per line

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

0.0

/jcache/src/main/java/com/github/benmanes/caffeine/jcache/event/EventDispatcher.java

/*
 * Copyright 2015 Ben Manes. 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 com.github.benmanes.caffeine.jcache.event;

import static java.util.Objects.requireNonNull;

import java.lang.System.Logger;
import java.lang.System.Logger.Level;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executor;

import javax.cache.Cache;
import javax.cache.configuration.CacheEntryListenerConfiguration;
import javax.cache.configuration.MutableCacheEntryListenerConfiguration;
import javax.cache.event.CacheEntryEventFilter;
import javax.cache.event.CacheEntryListener;
import javax.cache.event.EventType;

import org.jspecify.annotations.Nullable;

import com.google.errorprone.annotations.Var;

/**
 * A dispatcher that publishes cache events to listeners for asynchronous execution.
 * <p>
 * A {@link CacheEntryListener} is required to receive events in the order of the actions being
 * performed on the associated key. This implementation supports this by using a dispatch queue for
 * each listener and key pair, and provides the following characteristics:
 * <ul>
 *   <li>A listener may be executed in parallel for events with different keys
 *   <li>A listener is executed sequentially for events with the same key. This creates a dependency
 *       relationship between events and waiting dependents do not consume threads.
 *   <li>A listener receives a single event per invocation; batch processing is not supported
 *   <li>Multiple listeners may be executed in parallel for the same event
 *   <li>Listeners process events at their own rate and do not explicitly block each other
 *   <li>Listeners share a pool of threads for event processing. A slow listener may limit the
 *       throughput if all threads are busy handling distinct events, causing the execution of other
 *       listeners to be delayed until the executor is able to process the work.
 * </ul>
 * <p>
 * Some listeners may be configured as <code>synchronous</code>, meaning that the publishing thread
 * should wait until the listener has processed the event. The calling thread should publish within
 * an atomic block that mutates the entry, and complete the operation by calling
 * {@link #awaitSynchronous()} or {@link #ignoreSynchronous()}.
 *
 * @author ben.manes@gmail.com (Ben Manes)
 */
public final class EventDispatcher<K, V> {
  static final Logger logger = System.getLogger(EventDispatcher.class.getName());

  final ConcurrentMap<
      Registration<K, V>,
      ConcurrentMap<K, CompletableFuture<@Nullable Void>>> dispatchQueues;
  final ThreadLocal<List<CompletableFuture<@Nullable Void>>> pending;
  final Executor executor;

  public EventDispatcher(Executor executor) {
    this.pending = ThreadLocal.withInitial(ArrayList::new);
    this.dispatchQueues = new ConcurrentHashMap<>();
    this.executor = requireNonNull(executor);
  }

  /** Returns the cache entry listener registrations. */
  public Set<Registration<K, V>> registrations() {
    return Collections.unmodifiableSet(dispatchQueues.keySet());
  }

  /**
   * Registers a cache entry listener based on the supplied configuration.
   *
   * @param configuration the listener's configuration.
   */
  @SuppressWarnings("PMD.CloseResource")
  public void register(CacheEntryListenerConfiguration<K, V> configuration) {
    if (configuration.getCacheEntryListenerFactory() == null) {
      return;
    }
    var listener = new EventTypeAwareListener<K, V>(
        configuration.getCacheEntryListenerFactory().create());

    var factory = configuration.getCacheEntryEventFilterFactory();
    CacheEntryEventFilter<K, V> filter = (factory == null)
        ? event -> true
        : new EventTypeFilter<>(listener, factory.create());

    var registration = new Registration<>(configuration, filter, listener);
    dispatchQueues.putIfAbsent(registration, new ConcurrentHashMap<>());
  }

  /**
   * Deregisters a cache entry listener based on the supplied configuration.
   *
   * @param configuration the listener's configuration.
   */
  public void deregister(CacheEntryListenerConfiguration<K, V> configuration) {
    requireNonNull(configuration);
    var key = new MutableCacheEntryListenerConfiguration<>(configuration);
    dispatchQueues.keySet().removeIf(registration -> key.equals(registration.getConfiguration()));
  }

  /**
   * Publishes a creation event for the entry to the interested listeners.
   *
   * @param cache the cache where the entry was created
   * @param key the entry's key
   * @param value the entry's value
   */
  public void publishCreated(Cache<K, V> cache, K key, V value) {
    publish(cache, EventType.CREATED, key, /* hasOldValue= */ false,
        /* oldValue= */ null, /* newValue= */ value, /* quiet= */ false);
  }

  /**
   * Publishes an update event for the entry to the interested listeners.
   *
   * @param cache the cache where the entry was updated
   * @param key the entry's key
   * @param oldValue the entry's old value
   * @param newValue the entry's new value
   */
  public void publishUpdated(Cache<K, V> cache, K key, V oldValue, V newValue) {
    publish(cache, EventType.UPDATED, key, /* hasOldValue= */ true,
        oldValue, newValue, /* quiet= */ false);
  }

  /**
   * Publishes an update event for the entry to the interested listeners. This method does not
   * register the synchronous listener's future with {@link #awaitSynchronous()}.
   *
   * @param cache the cache where the entry was updated
   * @param key the entry's key
   * @param oldValue the entry's old value
   * @param newValue the entry's new value
   */
  public void publishUpdatedQuietly(Cache<K, V> cache, K key, V oldValue, V newValue) {
    publish(cache, EventType.UPDATED, key, /* hasOldValue= */ true,
        oldValue, newValue, /* quiet= */ true);
  }

  /**
   * Publishes a removal event for the entry to the interested listeners.
   *
   * @param cache the cache where the entry was removed
   * @param key the entry's key
   * @param value the entry's value
   */
  public void publishRemoved(Cache<K, V> cache, K key, V value) {
    publish(cache, EventType.REMOVED, key, /* hasOldValue= */ true,
        /* oldValue= */ value, /* newValue= */ value, /* quiet= */ false);
  }

  /**
   * Publishes a removal event for the entry to the interested listeners. This method does not
   * register the synchronous listener's future with {@link #awaitSynchronous()}.
   *
   * @param cache the cache where the entry was removed
   * @param key the entry's key
   * @param value the entry's value
   */
  public void publishRemovedQuietly(Cache<K, V> cache, K key, V value) {
    publish(cache, EventType.REMOVED, key, /* hasOldValue= */ true,
        /* oldValue= */ value, /* newValue= */ value, /* quiet= */ true);
  }

  /**
   * Publishes an expiration event for the entry to the interested listeners.
   *
   * @param cache the cache where the entry expired
   * @param key the entry's key
   * @param value the entry's value
   */
  public void publishExpired(Cache<K, V> cache, K key, V value) {
    publish(cache, EventType.EXPIRED, key, /* hasOldValue= */ true,
        /* oldValue= */ value, /* newValue= */ value, /* quiet= */ false);
  }

  /**
   * Publishes an expiration event for the entry to the interested listeners. This method does not
   * register the synchronous listener's future with {@link #awaitSynchronous()}.
   *
   * @param cache the cache where the entry expired
   * @param key the entry's key
   * @param value the entry's value
   */
  public void publishExpiredQuietly(Cache<K, V> cache, K key, V value) {
    publish(cache, EventType.EXPIRED, key, /* hasOldValue= */ true,
        /* oldValue= */ value, /* newValue= */ value, /* quiet= */ true);
  }

  /**
   * Blocks until all of the synchronous listeners have finished processing the events this thread
   * published.
   */
  public void awaitSynchronous() {
    var futures = pending.get();
    if (futures.isEmpty()) {
      return;
    }
    try {
      CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)).join();
    } catch (CompletionException e) {
      logger.log(Level.WARNING, "", e);
    } finally {
      futures.clear();
    }
  }

  /**
   * Ignores and clears the queued futures to the synchronous listeners that are processing events
   * this thread published.
   */
  public void ignoreSynchronous() {
    pending.get().clear();
  }

  /** Broadcasts the event to the interested listener's dispatch queues. */
  @SuppressWarnings("FutureReturnValueIgnored")
  private void publish(Cache<K, V> cache, EventType eventType, K key,
      boolean hasOldValue, @Nullable V oldValue, @Nullable V newValue, boolean quiet) {
    if (dispatchQueues.isEmpty()) {
      return;
    }

    @Var JCacheEntryEvent<K, V> event = null;
    for (var entry : dispatchQueues.entrySet()) {
      var registration = entry.getKey();
      if (!registration.getCacheEntryListener().isCompatible(eventType)) {
        continue;
      }
      if (event == null) {
        event = new JCacheEntryEvent<>(cache, eventType, key, hasOldValue, oldValue, newValue);
      }
      if (!registration.getCacheEntryFilter().evaluate(event)) {
        continue;
      }

      JCacheEntryEvent<K, V> e = event;
      var dispatchQueue = entry.getValue();
      var future = dispatchQueue.compute(key, (k, queue) -> {
        Runnable action = () -> registration.getCacheEntryListener().dispatch(e);
        return (queue == null)
            ? CompletableFuture.runAsync(action, executor)
            : queue.thenRunAsync(action, executor);
      });
      future.whenComplete((result, error) -> {
        // optimistic check to avoid locking if not a match
        if (dispatchQueue.get(key) == future) {
          dispatchQueue.remove(key, future);
        }
      });
      if (registration.isSynchronous() && !quiet) {
        pending.get().add(future);
      }
    }
  }
}

1	/*
2	* Copyright 2015 Ben Manes. 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 com.github.benmanes.caffeine.jcache.event;
17
18	import static java.util.Objects.requireNonNull;
19
20	import java.lang.System.Logger;
21	import java.lang.System.Logger.Level;
22	import java.util.ArrayList;
23	import java.util.Collections;
24	import java.util.List;
25	import java.util.Set;
26	import java.util.concurrent.CompletableFuture;
27	import java.util.concurrent.CompletionException;
28	import java.util.concurrent.ConcurrentHashMap;
29	import java.util.concurrent.ConcurrentMap;
30	import java.util.concurrent.Executor;
31
32	import javax.cache.Cache;
33	import javax.cache.configuration.CacheEntryListenerConfiguration;
34	import javax.cache.configuration.MutableCacheEntryListenerConfiguration;
35	import javax.cache.event.CacheEntryEventFilter;
36	import javax.cache.event.CacheEntryListener;
37	import javax.cache.event.EventType;
38
39	import org.jspecify.annotations.Nullable;
40
41	import com.google.errorprone.annotations.Var;
42
43	/**
44	* A dispatcher that publishes cache events to listeners for asynchronous execution.
45	* <p>
46	* A {@link CacheEntryListener} is required to receive events in the order of the actions being
47	* performed on the associated key. This implementation supports this by using a dispatch queue for
48	* each listener and key pair, and provides the following characteristics:
49	* <ul>
50	* <li>A listener may be executed in parallel for events with different keys
51	* <li>A listener is executed sequentially for events with the same key. This creates a dependency
52	* relationship between events and waiting dependents do not consume threads.
53	* <li>A listener receives a single event per invocation; batch processing is not supported
54	* <li>Multiple listeners may be executed in parallel for the same event
55	* <li>Listeners process events at their own rate and do not explicitly block each other
56	* <li>Listeners share a pool of threads for event processing. A slow listener may limit the
57	* throughput if all threads are busy handling distinct events, causing the execution of other
58	* listeners to be delayed until the executor is able to process the work.
59	* </ul>
60	* <p>
61	* Some listeners may be configured as <code>synchronous</code>, meaning that the publishing thread
62	* should wait until the listener has processed the event. The calling thread should publish within
63	* an atomic block that mutates the entry, and complete the operation by calling
64	* {@link #awaitSynchronous()} or {@link #ignoreSynchronous()}.
65	*
66	* @author ben.manes@gmail.com (Ben Manes)
67	*/
68	public final class EventDispatcher<K, V> {
UNCOV 69	static final Logger logger = System.getLogger(EventDispatcher.class.getName());	×
70
71	final ConcurrentMap<
72	Registration<K, V>,
73	ConcurrentMap<K, CompletableFuture<@Nullable Void>>> dispatchQueues;
74	final ThreadLocal<List<CompletableFuture<@Nullable Void>>> pending;
75	final Executor executor;
76
UNCOV 77	public EventDispatcher(Executor executor) {	×
UNCOV 78	this.pending = ThreadLocal.withInitial(ArrayList::new);	×
UNCOV 79	this.dispatchQueues = new ConcurrentHashMap<>();	×
UNCOV 80	this.executor = requireNonNull(executor);	×
UNCOV 81	}	×
82
83	/** Returns the cache entry listener registrations. */
84	public Set<Registration<K, V>> registrations() {
UNCOV 85	return Collections.unmodifiableSet(dispatchQueues.keySet());	×
86	}
87
88	/**
89	* Registers a cache entry listener based on the supplied configuration.
90	*
91	* @param configuration the listener's configuration.
92	*/
93	@SuppressWarnings("PMD.CloseResource")
94	public void register(CacheEntryListenerConfiguration<K, V> configuration) {
UNCOV 95	if (configuration.getCacheEntryListenerFactory() == null) {	×
UNCOV 96	return;	×
97	}
UNCOV 98	var listener = new EventTypeAwareListener<K, V>(	×
UNCOV 99	configuration.getCacheEntryListenerFactory().create());	×
100
UNCOV 101	var factory = configuration.getCacheEntryEventFilterFactory();	×
UNCOV 102	CacheEntryEventFilter<K, V> filter = (factory == null)	×
UNCOV 103	? event -> true	×
UNCOV 104	: new EventTypeFilter<>(listener, factory.create());	×
105
UNCOV 106	var registration = new Registration<>(configuration, filter, listener);	×
UNCOV 107	dispatchQueues.putIfAbsent(registration, new ConcurrentHashMap<>());	×
UNCOV 108	}	×
109
110	/**
111	* Deregisters a cache entry listener based on the supplied configuration.
112	*
113	* @param configuration the listener's configuration.
114	*/
115	public void deregister(CacheEntryListenerConfiguration<K, V> configuration) {
UNCOV 116	requireNonNull(configuration);	×
UNCOV 117	var key = new MutableCacheEntryListenerConfiguration<>(configuration);	×
UNCOV 118	dispatchQueues.keySet().removeIf(registration -> key.equals(registration.getConfiguration()));	×
UNCOV 119	}	×
120
121	/**
122	* Publishes a creation event for the entry to the interested listeners.
123	*
124	* @param cache the cache where the entry was created
125	* @param key the entry's key
126	* @param value the entry's value
127	*/
128	public void publishCreated(Cache<K, V> cache, K key, V value) {
UNCOV 129	publish(cache, EventType.CREATED, key, /* hasOldValue= */ false,	×
130	/* oldValue= / null, / newValue= / value, / quiet= */ false);
UNCOV 131	}	×
132
133	/**
134	* Publishes an update event for the entry to the interested listeners.
135	*
136	* @param cache the cache where the entry was updated
137	* @param key the entry's key
138	* @param oldValue the entry's old value
139	* @param newValue the entry's new value
140	*/
141	public void publishUpdated(Cache<K, V> cache, K key, V oldValue, V newValue) {
UNCOV 142	publish(cache, EventType.UPDATED, key, /* hasOldValue= */ true,	×
143	oldValue, newValue, /* quiet= */ false);
UNCOV 144	}	×
145
146	/**
147	* Publishes an update event for the entry to the interested listeners. This method does not
148	* register the synchronous listener's future with {@link #awaitSynchronous()}.
149	*
150	* @param cache the cache where the entry was updated
151	* @param key the entry's key
152	* @param oldValue the entry's old value
153	* @param newValue the entry's new value
154	*/
155	public void publishUpdatedQuietly(Cache<K, V> cache, K key, V oldValue, V newValue) {
UNCOV 156	publish(cache, EventType.UPDATED, key, /* hasOldValue= */ true,	×
157	oldValue, newValue, /* quiet= */ true);
UNCOV 158	}	×
159
160	/**
161	* Publishes a removal event for the entry to the interested listeners.
162	*
163	* @param cache the cache where the entry was removed
164	* @param key the entry's key
165	* @param value the entry's value
166	*/
167	public void publishRemoved(Cache<K, V> cache, K key, V value) {
UNCOV 168	publish(cache, EventType.REMOVED, key, /* hasOldValue= */ true,	×
169	/* oldValue= / value, / newValue= / value, / quiet= */ false);
UNCOV 170	}	×
171
172	/**
173	* Publishes a removal event for the entry to the interested listeners. This method does not
174	* register the synchronous listener's future with {@link #awaitSynchronous()}.
175	*
176	* @param cache the cache where the entry was removed
177	* @param key the entry's key
178	* @param value the entry's value
179	*/
180	public void publishRemovedQuietly(Cache<K, V> cache, K key, V value) {
UNCOV 181	publish(cache, EventType.REMOVED, key, /* hasOldValue= */ true,	×
182	/* oldValue= / value, / newValue= / value, / quiet= */ true);
UNCOV 183	}	×
184
185	/**
186	* Publishes an expiration event for the entry to the interested listeners.
187	*
188	* @param cache the cache where the entry expired
189	* @param key the entry's key
190	* @param value the entry's value
191	*/
192	public void publishExpired(Cache<K, V> cache, K key, V value) {
UNCOV 193	publish(cache, EventType.EXPIRED, key, /* hasOldValue= */ true,	×
194	/* oldValue= / value, / newValue= / value, / quiet= */ false);
UNCOV 195	}	×
196
197	/**
198	* Publishes an expiration event for the entry to the interested listeners. This method does not
199	* register the synchronous listener's future with {@link #awaitSynchronous()}.
200	*
201	* @param cache the cache where the entry expired
202	* @param key the entry's key
203	* @param value the entry's value
204	*/
205	public void publishExpiredQuietly(Cache<K, V> cache, K key, V value) {
UNCOV 206	publish(cache, EventType.EXPIRED, key, /* hasOldValue= */ true,	×
207	/* oldValue= / value, / newValue= / value, / quiet= */ true);
UNCOV 208	}	×
209
210	/**
211	* Blocks until all of the synchronous listeners have finished processing the events this thread
212	* published.
213	*/
214	public void awaitSynchronous() {
UNCOV 215	var futures = pending.get();	×
UNCOV 216	if (futures.isEmpty()) {	×
UNCOV 217	return;	×
218	}
219	try {
UNCOV 220	CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)).join();	×
UNCOV 221	} catch (CompletionException e) {	×
UNCOV 222	logger.log(Level.WARNING, "", e);	×
223	} finally {
UNCOV 224	futures.clear();	×
225	}
UNCOV 226	}	×
227
228	/**
229	* Ignores and clears the queued futures to the synchronous listeners that are processing events
230	* this thread published.
231	*/
232	public void ignoreSynchronous() {
UNCOV 233	pending.get().clear();	×
UNCOV 234	}	×
235
236	/** Broadcasts the event to the interested listener's dispatch queues. */
237	@SuppressWarnings("FutureReturnValueIgnored")
238	private void publish(Cache<K, V> cache, EventType eventType, K key,
239	boolean hasOldValue, @Nullable V oldValue, @Nullable V newValue, boolean quiet) {
UNCOV 240	if (dispatchQueues.isEmpty()) {	×
UNCOV 241	return;	×
242	}
243
UNCOV 244	@Var JCacheEntryEvent<K, V> event = null;	×
UNCOV 245	for (var entry : dispatchQueues.entrySet()) {	×
UNCOV 246	var registration = entry.getKey();	×
UNCOV 247	if (!registration.getCacheEntryListener().isCompatible(eventType)) {	×
UNCOV 248	continue;	×
249	}
UNCOV 250	if (event == null) {	×
UNCOV 251	event = new JCacheEntryEvent<>(cache, eventType, key, hasOldValue, oldValue, newValue);	×
252	}
UNCOV 253	if (!registration.getCacheEntryFilter().evaluate(event)) {	×
UNCOV 254	continue;	×
255	}
256
UNCOV 257	JCacheEntryEvent<K, V> e = event;	×
UNCOV 258	var dispatchQueue = entry.getValue();	×
UNCOV 259	var future = dispatchQueue.compute(key, (k, queue) -> {	×
UNCOV 260	Runnable action = () -> registration.getCacheEntryListener().dispatch(e);	×
UNCOV 261	return (queue == null)	×
UNCOV 262	? CompletableFuture.runAsync(action, executor)	×
UNCOV 263	: queue.thenRunAsync(action, executor);	×
264	});
UNCOV 265	future.whenComplete((result, error) -> {	×
266	// optimistic check to avoid locking if not a match
UNCOV 267	if (dispatchQueue.get(key) == future) {	×
UNCOV 268	dispatchQueue.remove(key, future);	×
269	}
UNCOV 270	});	×
UNCOV 271	if (registration.isSynchronous() && !quiet) {	×
UNCOV 272	pending.get().add(future);	×
273	}
UNCOV 274	}	×
UNCOV 275	}	×
276	}

ben-manes / caffeine / #5612

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