ISPN-8670 Refactor the configuration storage API

- add an enum-based ConfigurationStorage type for cleaner API/XML configuration
- rename the EmbeddedLocalConfigurationStorage to OverlayLocalConfigurationStorage (as it can be used for servers too)
- add a VolatileLocalConfigurationStorage

client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/RemoteCacheManagerAdminImpl.java

@@ -64,7 +64,10 @@ public void createCache(String name, String template, EnumSet<org.infinispan.cli
    @Override
    public void removeCache(String name) {
       remover.accept(name);
-      operationsFactory.newExecuteOperation("@@cache@remove", Collections.singletonMap(CACHE_NAME, string(name))).execute();
+      Map<String, byte[]> params = new HashMap<>(2);
+      params.put(CACHE_NAME, string(name));
+      if (flags != null && !flags.isEmpty()) params.put(FLAGS, flags(flags));
+      operationsFactory.newExecuteOperation("@@cache@remove", params).execute();
    }
 
    @Override

commons/src/main/java/org/infinispan/commons/configuration/attributes/AttributeDefinition.java

@@ -1,5 +1,7 @@
 package org.infinispan.commons.configuration.attributes;
 
+import java.util.function.Supplier;
+
 import org.infinispan.commons.CacheConfigurationException;
 import org.infinispan.commons.util.Util;
 
@@ -112,6 +114,14 @@ public static <T> Builder<T> builder(String name, T defaultValue, Class<T> klass
       return new Builder<>(name, defaultValue, klass);
    }
 
+   public static <T> Builder<Class<? extends T>> classBuilder(String name, Class<T> klass) {
+      return new Builder(name, null, Class.class);
+   }
+
+   public static <T> Builder<Supplier<? extends T>> supplierBuilder(String name, Class<T> klass) {
+      return new Builder(name, null, Supplier.class);
+   }
+
    @Override
    public boolean equals(Object o) {
       if (this == o) return true;

commons/src/main/java/org/infinispan/commons/util/Util.java

@@ -32,6 +32,7 @@
 import java.util.UUID;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
 
 import javax.naming.Context;
 import javax.security.auth.Subject;
@@ -1027,6 +1028,14 @@ else if (ch == 'f')
       return out.toString();
    }
 
+   public static <T> Supplier<T> getInstanceSupplier(Class<T> klass) {
+      return () -> getInstance(klass);
+   }
+
+   public static <T> Supplier<T> getInstanceSupplier(String className, ClassLoader classLoader) {
+      return () -> getInstance(className, classLoader);
+   }
+
   /**
     * Deletes directory recursively.
     *

core/src/main/java/org/infinispan/configuration/global/GlobalStateConfiguration.java

@@ -1,10 +1,12 @@
 package org.infinispan.configuration.global;
 
+import java.util.function.Supplier;
+
 import org.infinispan.commons.configuration.attributes.Attribute;
 import org.infinispan.commons.configuration.attributes.AttributeDefinition;
 import org.infinispan.commons.configuration.attributes.AttributeSet;
-import org.infinispan.globalstate.LocalConfigurationManager;
-import org.infinispan.globalstate.impl.EmbeddedLocalConfigurationManager;
+import org.infinispan.globalstate.ConfigurationStorage;
+import org.infinispan.globalstate.LocalConfigurationStorage;
 
 /**
  *
@@ -22,26 +24,31 @@ public class GlobalStateConfiguration {
    public static final AttributeDefinition<String> TEMPORARY_LOCATION = AttributeDefinition
          .builder("temporaryLocation", null, String.class)
             .initializer(() -> SecurityActions.getSystemProperty("java.io.tmpdir")).immutable().build();
-   public static final AttributeDefinition<LocalConfigurationManager> LOCAL_CONFIGURATION_MANAGER = AttributeDefinition
-         .builder("localConfigurationManager", null, LocalConfigurationManager.class)
-         .initializer(EmbeddedLocalConfigurationManager::new).immutable().build();
+   public static final AttributeDefinition<ConfigurationStorage> CONFIGURATION_STORAGE = AttributeDefinition
+         .builder("configurationStorage", ConfigurationStorage.VOLATILE, ConfigurationStorage.class).autoPersist(false)
+         .immutable().build();
+   public static final AttributeDefinition<Supplier<? extends LocalConfigurationStorage>> CONFIGURATION_STORAGE_SUPPLIER = AttributeDefinition
+         .supplierBuilder("configurationStorageSupplier", LocalConfigurationStorage.class).autoPersist(false)
+         .immutable().build();
 
    public static AttributeSet attributeDefinitionSet() {
-      return new AttributeSet(GlobalStateConfiguration.class, ENABLED, PERSISTENT_LOCATION, TEMPORARY_LOCATION, LOCAL_CONFIGURATION_MANAGER);
+      return new AttributeSet(GlobalStateConfiguration.class, ENABLED, PERSISTENT_LOCATION, TEMPORARY_LOCATION, CONFIGURATION_STORAGE, CONFIGURATION_STORAGE_SUPPLIER);
    }
 
    private final AttributeSet attributes;
    private final Attribute<Boolean> enabled;
-   private Attribute<String> persistentLocation;
-   private Attribute<String> temporaryLocation;
-   private Attribute<LocalConfigurationManager> localConfigurationManager;
+   private final Attribute<String> persistentLocation;
+   private final Attribute<String> temporaryLocation;
+   private final Attribute<ConfigurationStorage> configurationStorage;
+   private final Attribute<Supplier<? extends LocalConfigurationStorage>> configurationStorageSupplier;
 
    public GlobalStateConfiguration(AttributeSet attributes) {
       this.attributes = attributes.checkProtection();
       this.enabled = attributes.attribute(ENABLED);
       this.persistentLocation = attributes.attribute(PERSISTENT_LOCATION);
       this.temporaryLocation = attributes.attribute(TEMPORARY_LOCATION);
-      this.localConfigurationManager = attributes.attribute(LOCAL_CONFIGURATION_MANAGER);
+      this.configurationStorage = attributes.attribute(CONFIGURATION_STORAGE);
+      this.configurationStorageSupplier = attributes.attribute(CONFIGURATION_STORAGE_SUPPLIER);
    }
 
    public boolean enabled() {
@@ -65,11 +72,15 @@ public String temporaryLocation() {
       return temporaryLocation.get();
    }
 
+   public ConfigurationStorage configurationStorage() {
+      return configurationStorage.get();
+   }
+
    /**
-    * Returns the @{@link LocalConfigurationManager}
+    * Returns the {@link LocalConfigurationStorage} {@link Supplier}
     */
-   public LocalConfigurationManager localConfigurationManager() {
-      return localConfigurationManager.get();
+   public Supplier<? extends LocalConfigurationStorage> configurationStorageClass() {
+      return configurationStorageSupplier.get();
    }
 
    public AttributeSet attributes() {

core/src/main/java/org/infinispan/configuration/global/GlobalStateConfigurationBuilder.java

@@ -1,15 +1,18 @@
 package org.infinispan.configuration.global;
 
+import static org.infinispan.configuration.global.GlobalStateConfiguration.CONFIGURATION_STORAGE;
+import static org.infinispan.configuration.global.GlobalStateConfiguration.CONFIGURATION_STORAGE_SUPPLIER;
 import static org.infinispan.configuration.global.GlobalStateConfiguration.ENABLED;
-import static org.infinispan.configuration.global.GlobalStateConfiguration.LOCAL_CONFIGURATION_MANAGER;
 import static org.infinispan.configuration.global.GlobalStateConfiguration.PERSISTENT_LOCATION;
 import static org.infinispan.configuration.global.GlobalStateConfiguration.TEMPORARY_LOCATION;
 
 import java.lang.invoke.MethodHandles;
+import java.util.function.Supplier;
 
 import org.infinispan.commons.configuration.Builder;
 import org.infinispan.commons.configuration.attributes.AttributeSet;
-import org.infinispan.globalstate.LocalConfigurationManager;
+import org.infinispan.globalstate.ConfigurationStorage;
+import org.infinispan.globalstate.LocalConfigurationStorage;
 import org.infinispan.util.logging.Log;
 import org.infinispan.util.logging.LogFactory;
 
@@ -70,10 +73,20 @@ public GlobalStateConfigurationBuilder temporaryLocation(String location) {
    }
 
    /**
-    * Defines the @{@link LocalConfigurationManager}. Defaults to @{@link org.infinispan.globalstate.impl.EmbeddedLocalConfigurationManager}
+    * Defines the {@link ConfigurationStorage} strategy to use. If using {@link ConfigurationStorage#CUSTOM}, then
+    * the actual implementation must be passed by invoking {@link #localConfigurationManager(LocalConfigurationStorage)}
     */
-   public GlobalStateConfigurationBuilder localConfigurationManager(LocalConfigurationManager localConfigurationManager) {
-      attributes.attribute(LOCAL_CONFIGURATION_MANAGER).set(localConfigurationManager);
+   public GlobalStateConfigurationBuilder configurationStorage(ConfigurationStorage storage) {
+      attributes.attribute(CONFIGURATION_STORAGE).set(storage);
+      return this;
+   }
+
+   /**
+    * Defines the @{@link LocalConfigurationStorage}. Defaults to @{@link org.infinispan.globalstate.impl.VolatileLocalConfigurationStorage}
+    */
+   public GlobalStateConfigurationBuilder configurationStorageSupplier(Supplier<? extends LocalConfigurationStorage> configurationStorageSupplier) {
+      configurationStorage(ConfigurationStorage.CUSTOM);
+      attributes.attribute(CONFIGURATION_STORAGE_SUPPLIER).set(configurationStorageSupplier);
       return this;
    }
 
@@ -82,6 +95,9 @@ public void validate() {
       if (attributes.attribute(ENABLED).get() && attributes.attribute(PERSISTENT_LOCATION).isNull()) {
          log.missingGlobalStatePersistentLocation();
       }
+      if (attributes.attribute(CONFIGURATION_STORAGE).get().equals(ConfigurationStorage.CUSTOM) && attributes.attribute(CONFIGURATION_STORAGE_SUPPLIER).isNull()) {
+         throw log.customStorageStrategyNotSet();
+      }
    }
 
    @Override
@@ -94,5 +110,4 @@ public Builder<?> read(GlobalStateConfiguration template) {
       attributes.read(template.attributes());
       return this;
    }
-
 }

core/src/main/java/org/infinispan/configuration/parsing/Element.java

@@ -31,6 +31,7 @@ public enum Element {
     COMMON_NAME_ROLE_MAPPER("common-name-role-mapper"),
     COMPATIBILITY("compatibility"),
     CUSTOM_INTERCEPTORS("custom-interceptors"),
+    CUSTOM_CONFIGURATION_STORAGE("custom-configuration-storage"),
     CUSTOM_ROLE_MAPPER("custom-role-mapper"),
     DATA_CONTAINER("data-container"),
     DEFAULT("default"),
@@ -63,10 +64,13 @@ public enum Element {
     PERSISTENCE("persistence"),
     PERSISTENT_LOCATION("persistent-location"),
     LOCKING("locking"),
+    MANAGED_CONFIGURATION_STORAGE("managed-configuration-storage"),
     MEMORY("memory"),
     MODULES("modules"),
     OBJECT("object"),
     OFFHEAP("off-heap"),
+    OVERLAY_CONFIGURATION_STORAGE("overlay-configuration-storage"),
+    PARTITION_HANDLING("partition-handling"),
     PROPERTIES("properties"),
     PROPERTY("property"),
     RECOVERY("recovery"),
@@ -98,8 +102,8 @@ public enum Element {
     UNSAFE("unsafe"),
     VALUE_DATA_TYPE("value"),
     VERSIONING("versioning"),
+    VOLATILE_CONFIGURATION_STORAGE("volatile-configuration-storage"),
     WRITE_BEHIND("write-behind"),
-    PARTITION_HANDLING("partition-handling"),
     ;
 
     private final String name;

core/src/main/java/org/infinispan/configuration/parsing/Parser.java

@@ -16,6 +16,7 @@
 import java.util.Properties;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
 
 import javax.xml.stream.XMLStreamConstants;
 import javax.xml.stream.XMLStreamException;
@@ -65,6 +66,8 @@
 import org.infinispan.conflict.MergePolicy;
 import org.infinispan.eviction.EvictionType;
 import org.infinispan.factories.threads.DefaultThreadFactory;
+import org.infinispan.globalstate.ConfigurationStorage;
+import org.infinispan.globalstate.LocalConfigurationStorage;
 import org.infinispan.partitionhandling.PartitionHandling;
 import org.infinispan.persistence.cluster.ClusterLoader;
 import org.infinispan.persistence.file.SingleFileStore;
@@ -1127,6 +1130,7 @@ private void parseTransport(XMLExtendedStreamReader reader, ConfigurationBuilder
    private void parseGlobalState(XMLExtendedStreamReader reader, ConfigurationBuilderHolder holder) throws XMLStreamException {
       ParseUtils.requireNoAttributes(reader);
       GlobalStateConfigurationBuilder builder = holder.getGlobalConfigurationBuilder().globalState().enable();
+      ConfigurationStorage storage = null;
       while (reader.hasNext() && (reader.nextTag() != XMLStreamConstants.END_ELEMENT)) {
          Element element = Element.forName(reader.getLocalName());
          switch (element) {
@@ -1138,11 +1142,43 @@ private void parseGlobalState(XMLExtendedStreamReader reader, ConfigurationBuild
                builder.temporaryLocation(parseGlobalStatePath(reader));
                break;
             }
+            case VOLATILE_CONFIGURATION_STORAGE: {
+               if (storage != null) {
+                  throw ParseUtils.unexpectedElement(reader);
+               }
+               storage = ConfigurationStorage.VOLATILE;
+               break;
+            }
+            case OVERLAY_CONFIGURATION_STORAGE: {
+               if (storage != null) {
+                  throw ParseUtils.unexpectedElement(reader);
+               }
+               storage = ConfigurationStorage.OVERLAY;
+               break;
+            }
+            case MANAGED_CONFIGURATION_STORAGE: {
+               if (storage != null) {
+                  throw ParseUtils.unexpectedElement(reader);
+               } else {
+                  throw log.managerConfigurationStorageUnavailable();
+               }
+            }
+            case CUSTOM_CONFIGURATION_STORAGE: {
+               if (storage != null) {
+                  throw ParseUtils.unexpectedElement(reader);
+               }
+               storage = ConfigurationStorage.CUSTOM;
+               builder.configurationStorageSupplier(parseCustomConfigurationStorage(reader, holder));
+               break;
+            }
             default: {
                throw ParseUtils.unexpectedElement(reader);
             }
          }
       }
+      if (storage != null) {
+         builder.configurationStorage(storage);
+      }
    }
 
    private String parseGlobalStatePath(XMLExtendedStreamReader reader) throws XMLStreamException {
@@ -1167,6 +1203,12 @@ private String parseGlobalStatePath(XMLExtendedStreamReader reader) throws XMLSt
       return path;
    }
 
+   private Supplier<? extends LocalConfigurationStorage> parseCustomConfigurationStorage(XMLExtendedStreamReader reader, ConfigurationBuilderHolder holder) throws XMLStreamException {
+      String storageClass = ParseUtils.requireSingleAttribute(reader, Attribute.CLASS.getLocalName());
+      ParseUtils.requireNoContent(reader);
+      return Util.getInstanceSupplier(storageClass, holder.getClassLoader());
+   }
+
    private ThreadPoolConfiguration createThreadPoolConfiguration(String threadPoolName, String componentName) {
       ThreadPoolConfigurationBuilder threadPool = threadPools.get(threadPoolName);
       if (threadPool == null)

core/src/main/java/org/infinispan/globalstate/ConfigurationStorage.java

@@ -0,0 +1,27 @@
+package org.infinispan.globalstate;
+
+/**
+ * Configuration storage
+ *
+ * @author Tristan Tarrant
+ * @since 9.2
+ */
+
+public enum ConfigurationStorage {
+   /**
+    * A volatile configuration storage provider which doesn't support {@link org.infinispan.commons.api.CacheContainerAdmin.AdminFlag#PERMANENT}
+    */
+   VOLATILE,
+   /**
+    * A configuration storage provider which stores configurations on the filesystem
+    */
+   OVERLAY,
+   /**
+    * A configuration storage provider which stores configurations in a managed environment (e.g. the server model)
+    */
+   MANAGED,
+   /**
+    * A custom configuration storage provider
+    */
+   CUSTOM;
+}

core/src/main/java/org/infinispan/globalstate/LocalConfigurationManager.java → core/src/main/java/org/infinispan/globalstate/LocalConfigurationStorage.java

@@ -8,42 +8,46 @@
 import org.infinispan.manager.EmbeddedCacheManager;
 
 /**
- * The {@link LocalConfigurationManager} is responsible for applying on each node the configuration changes initiated
- * through the {@link org.infinispan.globalstate.GlobalConfigurationManager}.
+ * The {@link LocalConfigurationStorage} is responsible for applying on each node the configuration changes initiated
+ * through the {@link org.infinispan.globalstate.GlobalConfigurationManager} and persist them if requested via
+ * {@link org.infinispan.commons.api.CacheContainerAdmin.AdminFlag#PERMANENT}.
  *
  * @author Tristan Tarrant
  * @since 9.2
  */
 
-public interface LocalConfigurationManager {
+public interface LocalConfigurationStorage {
    /**
-    * Initialization entry point for the {@link LocalConfigurationManager}
+    * Initialization entry point for the {@link LocalConfigurationStorage}
     * @param embeddedCacheManager
-    * @param globalConfigurationManager
     */
-   void initialize(EmbeddedCacheManager embeddedCacheManager, GlobalConfigurationManager globalConfigurationManager);
+   void initialize(EmbeddedCacheManager embeddedCacheManager);
    /**
-    * Checks whether this {@link LocalConfigurationManager} supports the supplied flags.
+    * Checks whether this {@link LocalConfigurationStorage} supports the supplied flags.
     * A {@link org.infinispan.commons.CacheConfigurationException} will be thrown in case this cannot be done.
     *
     */
    void validateFlags(EnumSet<CacheContainerAdmin.AdminFlag> flags);
 
    /**
-    * Creates the cache using the supplied configuration and flags
+    * Creates the cache using the supplied template, configuration and flags. This method may be invoked either with or
+    * without a template. In both cases a concrete configuration will also be available. If a template name is present,
+    * the {@link LocalConfigurationStorage} should use it, e.g. when persisting the configuration.
     *
     * @param name the name of the cache to create
+    * @param template the template that should be used to configure the cache. Can be null.
     * @param configuration the {@link Configuration} to use
     * @param flags the desired {@link org.infinispan.commons.api.CacheContainerAdmin.AdminFlag}s
     */
-   void createCache(String name, Configuration configuration, EnumSet<CacheContainerAdmin.AdminFlag> flags);
+   void createCache(String name, String template, Configuration configuration, EnumSet<CacheContainerAdmin.AdminFlag> flags);
 
    /**
     * Removes the specified cache.
     *
     * @param name the name of the cache to remove
+    * @param flags the desired {@link org.infinispan.commons.api.CacheContainerAdmin.AdminFlag}s
     */
-   void removeCache(String name);
+   void removeCache(String name, EnumSet<CacheContainerAdmin.AdminFlag> flags);
 
    /**
     * Loads all persisted cache configurations