fix: redirect configuration (#2955)

* fix: redirect filter is applied only to '/'

* fix: mark filter as supporting async, to work with ibmcompat API

* fix: attempt to avoid CheckPeriodCache error [1], add eviction

[1]: java.lang.IllegalStateException: Recursive update
 at java.base/java.util.concurrent.ConcurrentHashMap.compute(ConcurrentHashMap.java:1991)
 at io.apicurio.registry.utils.CheckPeriodCache.compute(CheckPeriodCache.java:36)
 at io.apicurio.registry.mt.TenantContextLoader.loadRequestContext(TenantContextLoader.java:81)

* fix: add eviction check to all methods that potentially increase the size

* Update pom.xml

Co-authored-by: Eric Wittmann <eric.wittmann@gmail.com>

Author: jsenko

app/src/main/java/io/apicurio/registry/mt/TenantContextLoader.java

@@ -25,6 +25,7 @@
 import io.quarkus.runtime.StartupEvent;
 import org.eclipse.microprofile.config.inject.ConfigProperty;
 
+import java.time.Duration;
 import javax.enterprise.context.ApplicationScoped;
 import javax.enterprise.event.Observes;
 import javax.inject.Inject;
@@ -54,7 +55,7 @@ public class TenantContextLoader {
     Long cacheCheckPeriod;
 
     public void onStart(@Observes StartupEvent ev) {
-        contextsCache = new CheckPeriodCache<>(cacheCheckPeriod);
+        contextsCache = new CheckPeriodCache<>(Duration.ofMillis(cacheCheckPeriod));
     }
 
     /**

app/src/main/java/io/apicurio/registry/storage/metrics/StorageMetricsStore.java

@@ -16,21 +16,21 @@
 
 package io.apicurio.registry.storage.metrics;
 
-import java.util.concurrent.atomic.AtomicLong;
-
-import javax.enterprise.context.ApplicationScoped;
-import javax.enterprise.event.Observes;
-import javax.inject.Inject;
-
-import org.eclipse.microprofile.config.inject.ConfigProperty;
-import org.slf4j.Logger;
 import io.apicurio.common.apps.config.Info;
 import io.apicurio.registry.mt.TenantContext;
 import io.apicurio.registry.storage.RegistryStorage;
 import io.apicurio.registry.types.Current;
 import io.apicurio.registry.utils.CheckPeriodCache;
 import io.quarkus.runtime.StartupEvent;
 import lombok.EqualsAndHashCode;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.slf4j.Logger;
+
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicLong;
+import javax.enterprise.context.ApplicationScoped;
+import javax.enterprise.event.Observes;
+import javax.inject.Inject;
 
 /**
  * This class provides a set of per-tenant counters. Counters such as "number of artifacts"
@@ -75,9 +75,9 @@ private static class ArtifactVersionKey {
     }
 
     public void onStart(@Observes StartupEvent ev) {
-        totalSchemasCounters = new CheckPeriodCache<>(limitsCheckPeriod);
-        artifactsCounters = new CheckPeriodCache<>(limitsCheckPeriod);
-        artifactVersionsCounters = new CheckPeriodCache<>(limitsCheckPeriod);
+        totalSchemasCounters = new CheckPeriodCache<>(Duration.ofMillis(limitsCheckPeriod));
+        artifactsCounters = new CheckPeriodCache<>(Duration.ofMillis(limitsCheckPeriod));
+        artifactVersionsCounters = new CheckPeriodCache<>(Duration.ofMillis(limitsCheckPeriod));
     }
 
     public long getOrInitializeTotalSchemasCounter() {

app/src/main/resources-unfiltered/META-INF/web.xml

@@ -7,10 +7,11 @@
     <filter>
         <filter-name>RedirectFilter</filter-name>
         <filter-class>io.apicurio.registry.ui.servlets.RedirectFilter</filter-class>
+        <async-supported>true</async-supported>
     </filter>
     <filter-mapping>
         <filter-name>RedirectFilter</filter-name>
-        <url-pattern>/</url-pattern>
+        <url-pattern>/*</url-pattern>
     </filter-mapping>
 
     <filter>

common/pom.xml

@@ -50,6 +50,12 @@
             <scope>compile</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+
     </dependencies>
 
 

common/src/main/java/io/apicurio/registry/utils/CheckPeriodCache.java

@@ -16,56 +16,106 @@
 
 package io.apicurio.registry.utils;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.time.Instant;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 
 /**
  * @author Fabian Martinez
+ * @author Jakub Senko <m@jsenko.net>
  */
 public class CheckPeriodCache<K, V> {
 
-    private Map<K, CheckValue<V>> cache = new ConcurrentHashMap<>();
-    private long checkPeriodMillis = 0;
+    private static final Logger log = LoggerFactory.getLogger(CheckPeriodCache.class);
+
+    private final ConcurrentHashMap<K, CheckValue<V>> cache = new ConcurrentHashMap<>();
+
+    private final Duration checkPeriod;
+
+    private final AtomicBoolean evicting = new AtomicBoolean(false);
+    private final int evictionThreshold;
+
+    public CheckPeriodCache(Duration checkPeriod, int evictionThreshold) {
+        this.checkPeriod = checkPeriod;
+        this.evictionThreshold = evictionThreshold;
+    }
+
+    public CheckPeriodCache(Duration checkPeriod) {
+        this.checkPeriod = checkPeriod;
+        this.evictionThreshold = 1000;
+    }
 
-    public CheckPeriodCache(long checkPeriodMillis) {
-        this.checkPeriodMillis = checkPeriodMillis;
+    private boolean isExpired(CheckValue<?> checkedValue) {
+        return Instant.now().isAfter(checkedValue.lastUpdate.plus(checkPeriod));
+    }
+
+    private void checkEviction() {
+        final int currentSize = cache.size();
+        if (currentSize > evictionThreshold) {
+            if (evicting.compareAndSet(false, true)) {
+                try {
+                    // This thread gets to evict
+                    log.debug("Thread {} is evicting the cache. Current size is {}, threshold is {}.",
+                            Thread.currentThread().getName(), currentSize, evictionThreshold);
+                    var toEvict = cache.entrySet().stream()
+                            .filter(entry -> isExpired(entry.getValue()))
+                            .map(Entry::getKey)
+                            .collect(Collectors.toList());
+
+                    log.debug("Thread {} is evicting the cache. Found {} candidates for eviction.",
+                            Thread.currentThread().getName(), toEvict.size());
+                    toEvict.forEach(k -> {
+                        cache.compute(k, (key, value) -> {
+                            if (value == null || isExpired(value)) {
+                                return null;
+                            } else {
+                                return value;
+                            }
+                        });
+                    });
+
+                    log.debug("Thread {} has finished evicting the cache. The new size is {}.",
+                            Thread.currentThread().getName(), cache.size());
+                } finally {
+                    evicting.set(false);
+                }
+            }
+        }
     }
 
     public V compute(K k, Function<K, V> remappingFunction) {
+        checkEviction();
         CheckValue<V> returnValue = cache.compute(k, (key, checkedValue) -> {
-            long now = System.currentTimeMillis();
-            if (checkedValue == null) {
+            if (checkedValue == null || isExpired(checkedValue)) {
+                // Only execute if expired, but do it first in case an exception is thrown
                 V value = remappingFunction.apply(key);
-                return new CheckValue<>(now, value);
+                return new CheckValue<>(Instant.now(), value);
             } else {
-                if (checkedValue.lastUpdate + checkPeriodMillis < now) {
-                    V value = remappingFunction.apply(key);
-                    checkedValue.lastUpdate = now;
-                    checkedValue.value = value;
-                }
                 return checkedValue;
             }
         });
         return returnValue.value;
     }
 
     public void put(K k, V v) {
-        cache.put(k, new CheckValue<>(System.currentTimeMillis(), v));
+        checkEviction();
+        cache.put(k, new CheckValue<>(Instant.now(), v));
     }
 
     public V get(K k) {
         CheckValue<V> value = cache.compute(k, (key, checkedValue) -> {
-            if (checkedValue == null) {
+            if (checkedValue == null || isExpired(checkedValue)) {
                 return null;
             } else {
-                long now = System.currentTimeMillis();
-                if (checkedValue.lastUpdate + checkPeriodMillis < now) {
-                    //value expired
-                    return null;
-                } else {
-                    return checkedValue;
-                }
+                return checkedValue;
             }
         });
         return value == null ? null : value.value;
@@ -79,15 +129,25 @@ public void clear() {
         cache.clear();
     }
 
-    private static class CheckValue<V> {
+    /**
+     * USE FOR TEST ONLY
+     */
+    Map<K, CheckValue<V>> getInternal() {
+        return cache;
+    }
+
+    int size() {
+        return cache.size();
+    }
 
-        CheckValue(long ts, V value) {
-            this.lastUpdate = ts;
+    static class CheckValue<V> {
+
+        CheckValue(Instant lastUpdate, V value) {
+            this.lastUpdate = lastUpdate;
             this.value = value;
         }
 
-        long lastUpdate;
+        Instant lastUpdate;
         V value;
     }
-
 }

common/src/test/java/io/apicurio/registry/utils/CheckPeriodCacheTest.java

@@ -0,0 +1,58 @@
+package io.apicurio.registry.utils;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+class CheckPeriodCacheTest {
+
+    @Test
+    void testEviction() throws InterruptedException {
+        for (int attempt = 0; attempt < 5; attempt++) {
+
+            var cache = new CheckPeriodCache<Integer, Integer>(Duration.ofSeconds(1), 10);
+            var t1 = new Thread(() -> {
+                for (int i = 1; i <= 1000; i++) {
+                    cache.compute(i, k -> k);
+                }
+                Assertions.assertEquals(1000, cache.size());
+                Assertions.assertEquals(cache.getInternal().size(), cache.size());
+                try {
+                    Thread.sleep(1000);
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
+                for (int i = 1001; i <= 1010; i++) {
+                    cache.compute(i, k -> k);
+                }
+            });
+            var t2 = new Thread(() -> {
+                for (int i = 1; i <= 1000; i++) {
+                    cache.compute(i, k -> k);
+                }
+                Assertions.assertEquals(1000, cache.size());
+                Assertions.assertEquals(cache.getInternal().size(), cache.size());
+                try {
+                    Thread.sleep(1000);
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
+                for (int i = 1001; i <= 1010; i++) {
+                    cache.compute(i, k -> k);
+                }
+            });
+            t1.start();
+            t2.start();
+            t1.join();
+            t2.join();
+
+            Assertions.assertEquals(10, cache.size());
+            var actual = cache.getInternal().values().stream().map(c -> c.value).collect(Collectors.toSet());
+            var expected = IntStream.range(1001, 1011).boxed().collect(Collectors.toSet());
+            Assertions.assertEquals(expected, actual);
+        }
+    }
+}

integration-tests/testsuite/pom.xml

@@ -77,6 +77,12 @@
             <scope>provided</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.jboss.slf4j</groupId>
+            <artifactId>slf4j-jboss-logging</artifactId>
+            <version>${jboss-slf4j.version}</version>
+        </dependency>
+
         <!-- Test Only -->
         <dependency>
             <groupId>io.confluent</groupId>