readURL from adaptContext?

incubator/catalog-filesystem/src/main/java/io/aklivity/zilla/runtime/catalog/filesystem/internal/FilesystemCatalogContext.java

@@ -35,6 +35,6 @@ public CatalogHandler attach(
         CatalogConfig catalog)
     {
         return new FilesystemCatalogHandler(
-            FilesystemOptionsConfig.class.cast(catalog.options), context, catalog.id, catalog.readURL);
+            FilesystemOptionsConfig.class.cast(catalog.options), context, catalog.id);
     }
 }

incubator/catalog-filesystem/src/main/java/io/aklivity/zilla/runtime/catalog/filesystem/internal/FilesystemCatalogHandler.java

@@ -37,14 +37,13 @@ public class FilesystemCatalogHandler implements CatalogHandler
     public FilesystemCatalogHandler(
         FilesystemOptionsConfig config,
         EngineContext context,
-        long catalogId,
-        Function<String, String> readURL)
+        long catalogId)
     {
         this.schemas = new HashMap<>();
         this.schemaIds =  new HashMap<>();
         this.crc32c = new CRC32C();
         this.event = new FilesystemEventContext(context);
-        this.readURL = readURL;
+        this.readURL = config.readURL;
         this.catalogId = catalogId;
         registerSchema(config.subjects);
     }

incubator/catalog-filesystem/src/main/java/io/aklivity/zilla/runtime/catalog/filesystem/internal/config/FilesystemOptionsConfig.java

@@ -22,6 +22,7 @@
 public class FilesystemOptionsConfig extends OptionsConfig
 {
     public final List<FilesystemSchemaConfig> subjects;
+    public final Function<String, String> readURL;
 
     public static FilesystemOptionsConfigBuilder<FilesystemOptionsConfig> builder()
     {
@@ -35,8 +36,10 @@ public static <T> FilesystemOptionsConfigBuilder<T> builder(
     }
 
     public FilesystemOptionsConfig(
-        List<FilesystemSchemaConfig> subjects)
+        List<FilesystemSchemaConfig> subjects,
+        Function<String, String> readURL)
     {
         this.subjects = subjects;
+        this.readURL = readURL;
     }
 }

incubator/catalog-filesystem/src/main/java/io/aklivity/zilla/runtime/catalog/filesystem/internal/config/FilesystemOptionsConfigAdapter.java

@@ -14,11 +14,14 @@
  */
 package io.aklivity.zilla.runtime.catalog.filesystem.internal.config;
 
+import java.util.function.Function;
+
 import jakarta.json.Json;
 import jakarta.json.JsonObject;
 import jakarta.json.JsonObjectBuilder;
 import jakarta.json.bind.adapter.JsonbAdapter;
 
+import io.aklivity.zilla.runtime.engine.config.ConfigAdapterContext;
 import io.aklivity.zilla.runtime.engine.config.OptionsConfig;
 import io.aklivity.zilla.runtime.engine.config.OptionsConfigAdapterSpi;
 
@@ -28,6 +31,7 @@ public class FilesystemOptionsConfigAdapter implements OptionsConfigAdapterSpi,
     private static final String VERSION_NAME = "version";
     private static final String URL_NAME = "url";
     private static final String VERSION_DEFAULT = "latest";
+    private Function<String, String> readURL;
 
     @Override
     public Kind kind()
@@ -92,6 +96,15 @@ public OptionsConfig adaptFromJson(
                 }
             }
         }
+        options.readURL(readURL);
+
         return options.build();
     }
+
+    @Override
+    public void adaptContext(
+        ConfigAdapterContext context)
+    {
+        this.readURL = context::readURL;
+    }
 }

incubator/catalog-filesystem/src/main/java/io/aklivity/zilla/runtime/catalog/filesystem/internal/config/FilesystemOptionsConfigBuilder.java

@@ -26,6 +26,7 @@ public final class FilesystemOptionsConfigBuilder<T> extends ConfigBuilder<T, Fi
     private final Function<OptionsConfig, T> mapper;
 
     private List<FilesystemSchemaConfig> subjects;
+    private Function<String, String> readURL;
 
     FilesystemOptionsConfigBuilder(
         Function<OptionsConfig, T> mapper)
@@ -56,9 +57,16 @@ public FilesystemOptionsConfigBuilder<T> subjects(
         return this;
     }
 
+    public FilesystemOptionsConfigBuilder<T> readURL(
+        Function<String, String> readURL)
+    {
+        this.readURL = readURL;
+        return this;
+    }
+
     @Override
     public T build()
     {
-        return mapper.apply(new FilesystemOptionsConfig(subjects));
+        return mapper.apply(new FilesystemOptionsConfig(subjects, readURL));
     }
 }

incubator/catalog-filesystem/src/test/java/io/aklivity/zilla/runtime/catalog/filesystem/internal/FilesystemCatalogFactoryTest.java

@@ -51,14 +51,13 @@ public void shouldLoadAndCreate()
         assertThat(context, instanceOf(FilesystemCatalogContext.class));
 
         Function<String, String> readURL = mock(Function.class);
+        Mockito.doReturn("test").when(readURL).apply("asyncapi/mqtt.yaml");
 
         FilesystemOptionsConfig catalogConfig =
             new FilesystemOptionsConfig(singletonList(
-                new FilesystemSchemaConfig("subject1", "asyncapi/mqtt.yaml", "latest")));
+                new FilesystemSchemaConfig("subject1", "asyncapi/mqtt.yaml", "latest")), readURL);
 
         CatalogConfig options = new CatalogConfig("test", "catalog0", "filesystem", catalogConfig);
-        options.readURL = readURL;
-        Mockito.doReturn("test").when(readURL).apply("asyncapi/mqtt.yaml");
         CatalogHandler handler = context.attach(options);
 
         assertThat(handler, instanceOf(FilesystemCatalogHandler.class));

incubator/catalog-filesystem/src/test/java/io/aklivity/zilla/runtime/catalog/filesystem/internal/FilesystemIT.java

@@ -50,7 +50,7 @@ public class FilesystemIT
     public void setup() throws IOException
     {
         config = new FilesystemOptionsConfig(singletonList(
-            new FilesystemSchemaConfig("subject1", "asyncapi/mqtt.yaml", "latest")));
+            new FilesystemSchemaConfig("subject1", "asyncapi/mqtt.yaml", "latest")), readURL);
 
         String content;
         try (InputStream resource = FilesystemIT.class
@@ -76,7 +76,7 @@ public void shouldResolveSchemaViaSchemaId()
             "    protocol: mqtt\n" +
             "defaultContentType: application/json";
 
-        FilesystemCatalogHandler catalog = new FilesystemCatalogHandler(config, context, 0L, readURL);
+        FilesystemCatalogHandler catalog = new FilesystemCatalogHandler(config, context, 0L);
 
         int schemaId = catalog.resolve("subject1", "latest");
         String schema = catalog.resolve(schemaId);
@@ -88,7 +88,7 @@ public void shouldResolveSchemaViaSchemaId()
     @Test
     public void shouldResolveSchemaIdAndProcessData()
     {
-        FilesystemCatalogHandler catalog = new FilesystemCatalogHandler(config, context, 0L, readURL);
+        FilesystemCatalogHandler catalog = new FilesystemCatalogHandler(config, context, 0L);
 
         DirectBuffer data = new UnsafeBuffer();
 
@@ -108,7 +108,7 @@ public void shouldResolveSchemaIdAndProcessData()
     @Test
     public void shouldVerifyEncodedData()
     {
-        FilesystemCatalogHandler catalog = new FilesystemCatalogHandler(config, context, 0L, readURL);
+        FilesystemCatalogHandler catalog = new FilesystemCatalogHandler(config, context, 0L);
 
         DirectBuffer data = new UnsafeBuffer();
 

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/config/CatalogConfig.java

@@ -17,12 +17,9 @@
 
 import static java.util.Objects.requireNonNull;
 
-import java.util.function.Function;
-
 public class CatalogConfig
 {
     public transient long id;
-    public transient Function<String, String> readURL;
 
     public final String namespace;
     public final String name;

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/internal/registry/EngineManager.java

@@ -243,7 +243,6 @@ private void process(
         for (CatalogConfig catalog : namespace.catalogs)
         {
             catalog.id = resolver.resolve(catalog.name);
-            catalog.readURL = namespace.readURL;
         }
 
         for (MetricConfig metric : namespace.telemetry.metrics)