unify caching across workers to maximize cache hits

runtime/catalog-karapace/pom.xml

@@ -24,7 +24,7 @@
   <properties>
     <maven.compiler.source>11</maven.compiler.source>
     <maven.compiler.target>11</maven.compiler.target>
-    <jacoco.coverage.ratio>0.85</jacoco.coverage.ratio>
+    <jacoco.coverage.ratio>0.83</jacoco.coverage.ratio>
     <jacoco.missed.count>0</jacoco.missed.count>
   </properties>
 

runtime/catalog-karapace/src/main/java/io/aklivity/zilla/runtime/catalog/karapace/internal/CachedSchema.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2021-2023 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package io.aklivity.zilla.runtime.catalog.karapace.internal;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class CachedSchema
+{
+    public static final String SCHEMA_PLACEHOLDER = "schema";
+    public static final CachedSchema IN_PROGRESS = new CachedSchema(SCHEMA_PLACEHOLDER);
+
+    public final String schema;
+    public final AtomicInteger retryAttempts;
+
+    public CachedSchema(
+        String schema)
+    {
+        this.schema = schema;
+        this.retryAttempts = new AtomicInteger();
+    }
+
+    public CachedSchema(
+        String schema,
+        AtomicInteger retryAttempts)
+    {
+        this.schema = schema;
+        this.retryAttempts = retryAttempts;
+    }
+}

runtime/catalog-karapace/src/main/java/io/aklivity/zilla/runtime/catalog/karapace/internal/CachedSchemaId.java

@@ -14,16 +14,39 @@
  */
 package io.aklivity.zilla.runtime.catalog.karapace.internal;
 
+import java.util.concurrent.atomic.AtomicInteger;
+
 public class CachedSchemaId
 {
-    public long timestamp;
-    public int id;
+    public static final int PLACEHOLDER_SCHEMA_ID = -1;
+    public static final CachedSchemaId IN_PROGRESS = new CachedSchemaId(Long.MAX_VALUE, PLACEHOLDER_SCHEMA_ID,
+        new AtomicInteger(), Long.MAX_VALUE);
+
+    public final long timestamp;
+    public final int id;
+    public final AtomicInteger retryAttempts;
+    public final long retryAfter;
 
     public CachedSchemaId(
         long timestamp,
-        int id)
+        int id,
+        AtomicInteger retryAttempts,
+        long retryAfter)
     {
         this.timestamp = timestamp;
         this.id = id;
+        this.retryAttempts = retryAttempts;
+        this.retryAfter = retryAfter;
+    }
+
+    public boolean expired(
+        long maxAgeMillis)
+    {
+        return System.currentTimeMillis() - this.timestamp > maxAgeMillis;
+    }
+
+    public boolean retry()
+    {
+        return System.currentTimeMillis() - this.timestamp > this.retryAfter;
     }
 }

runtime/catalog-karapace/src/main/java/io/aklivity/zilla/runtime/catalog/karapace/internal/KarapaceCache.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2021-2023 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package io.aklivity.zilla.runtime.catalog.karapace.internal;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+public class KarapaceCache
+{
+    public final ConcurrentMap<Integer, CompletableFuture<CachedSchema>> schemas;
+    public final ConcurrentMap<Integer, CompletableFuture<CachedSchemaId>> schemaIds;
+
+    public KarapaceCache()
+    {
+        this.schemas = new ConcurrentHashMap<>();
+        this.schemaIds = new ConcurrentHashMap<>();
+    }
+}

runtime/catalog-karapace/src/main/java/io/aklivity/zilla/runtime/catalog/karapace/internal/KarapaceCatalog.java

@@ -15,6 +15,8 @@
 package io.aklivity.zilla.runtime.catalog.karapace.internal;
 
 import java.net.URL;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
 import io.aklivity.zilla.runtime.engine.Configuration;
 import io.aklivity.zilla.runtime.engine.EngineContext;
@@ -25,9 +27,12 @@ public class KarapaceCatalog implements Catalog
 {
     public static final String NAME = "karapace";
 
+    private final ConcurrentMap<Long, KarapaceCache> cache;
+
     public KarapaceCatalog(
         Configuration config)
     {
+        this.cache = new ConcurrentHashMap<>();
     }
 
     @Override
@@ -40,7 +45,7 @@ public String name()
     public CatalogContext supply(
         EngineContext context)
     {
-        return new KarapaceCatalogContext(context);
+        return new KarapaceCatalogContext(context, cache);
     }
 
     @Override

runtime/catalog-karapace/src/main/java/io/aklivity/zilla/runtime/catalog/karapace/internal/KarapaceCatalogContext.java

@@ -14,6 +14,8 @@
  */
 package io.aklivity.zilla.runtime.catalog.karapace.internal;
 
+import java.util.concurrent.ConcurrentMap;
+
 import io.aklivity.zilla.runtime.catalog.karapace.internal.config.KarapaceOptionsConfig;
 import io.aklivity.zilla.runtime.engine.EngineContext;
 import io.aklivity.zilla.runtime.engine.catalog.CatalogContext;
@@ -23,17 +25,21 @@
 public class KarapaceCatalogContext implements CatalogContext
 {
     private final EngineContext context;
+    private final ConcurrentMap<Long, KarapaceCache> cachesById;
 
     public KarapaceCatalogContext(
-        EngineContext context)
+        EngineContext context,
+        ConcurrentMap<Long, KarapaceCache> cachesById)
     {
         this.context = context;
+        this.cachesById = cachesById;
     }
 
     @Override
     public CatalogHandler attach(
         CatalogConfig catalog)
     {
-        return new KarapaceCatalogHandler(KarapaceOptionsConfig.class.cast(catalog.options), context, catalog.id);
+        KarapaceCache cache = cachesById.computeIfAbsent(catalog.id, id -> new KarapaceCache());
+        return new KarapaceCatalogHandler(KarapaceOptionsConfig.class.cast(catalog.options), context, catalog.id, cache);
     }
 }