Infinispan, upload schemas on backup clusters

docs/src/main/asciidoc/infinispan-client-reference.adoc

@@ -174,6 +174,12 @@ public class InfinispanExample {
 <2> The client connects to the 'lon-site'.
 <3> The client connects to the default site.
 
+By default, Protobuf Schemas will also be uploaded to the backup clusters. However, it might be required to handle
+the registration manually as a schema may evolve over time when used in production, so you can
+disable this from occurring in each backup site by configuring the
+`quarkus.infinispan-client.backup-cluster.YOUR_SITE_NAME.use-schema-registration` to `false`.
+The value of this property will be ignored if the `use-schema-registration` global property is `false`.
+
 [NOTE]
 ====
 Cross-site replication is a powerful feature offered by Infinispan that facilitates data backup between clusters

extensions/infinispan-client/deployment/src/main/java/io/quarkus/infinispan/client/deployment/MarshallingBuildItem.java

@@ -7,7 +7,7 @@
 
 public final class MarshallingBuildItem extends SimpleBuildItem {
 
-    // holds protostream requierements
+    // holds protostream requirements
     private final Properties properties;
 
     // a marshaller can be defined for different client names

extensions/infinispan-client/runtime/src/main/java/io/quarkus/infinispan/client/runtime/InfinispanClientProducer.java

@@ -67,6 +67,8 @@ private void registerSchemaInServer(String infinispanConfigName,
         RemoteCache<String, String> protobufMetadataCache = null;
         Properties namedProperties = properties.get(infinispanConfigName);
         Set<SerializationContextInitializer> initializers = (Set) namedProperties.remove(PROTOBUF_INITIALIZERS);
+        InfinispanClientRuntimeConfig runtimeConfig = this.infinispanClientsRuntimeConfigHandle.get()
+                .getInfinispanClientRuntimeConfig(infinispanConfigName);
         if (initializers != null) {
             for (SerializationContextInitializer initializer : initializers) {
                 if (protobufMetadataCache == null) {
@@ -75,6 +77,20 @@ private void registerSchemaInServer(String infinispanConfigName,
                 }
                 protobufMetadataCache.put(initializer.getProtoFileName(), initializer.getProtoFile());
             }
+            runtimeConfig.backupCluster.entrySet().forEach(backup -> {
+                if (backup.getValue().useSchemaRegistration.orElse(true)) {
+                    cacheManager.switchToCluster(backup.getKey());
+                    for (SerializationContextInitializer initializer : initializers) {
+                        RemoteCache<String, String> backupProtobufMetadataCache = null;
+                        if (backupProtobufMetadataCache == null) {
+                            backupProtobufMetadataCache = cacheManager.getCache(
+                                    ProtobufMetadataManagerConstants.PROTOBUF_METADATA_CACHE_NAME);
+                        }
+                        backupProtobufMetadataCache.put(initializer.getProtoFileName(), initializer.getProtoFile());
+                    }
+                    cacheManager.switchToDefaultCluster();
+                }
+            });
         }
 
         for (Map.Entry<Object, Object> property : namedProperties.entrySet()) {
@@ -92,6 +108,29 @@ private void registerSchemaInServer(String infinispanConfigName,
                 }
             }
         }
+
+        runtimeConfig.backupCluster.entrySet().forEach(backupConfigEntry -> {
+            if (backupConfigEntry.getValue().useSchemaRegistration.orElse(true)) {
+                cacheManager.switchToCluster(backupConfigEntry.getKey());
+                RemoteCache<String, String> backupProtobufMetadataCache = null;
+                for (Map.Entry<Object, Object> property : namedProperties.entrySet()) {
+                    Object key = property.getKey();
+                    if (key instanceof String) {
+                        String keyString = (String) key;
+                        if (keyString.startsWith(PROTOBUF_FILE_PREFIX)) {
+                            String fileName = keyString.substring(PROTOBUF_FILE_PREFIX.length());
+                            String fileContents = (String) property.getValue();
+                            if (backupProtobufMetadataCache == null) {
+                                backupProtobufMetadataCache = cacheManager.getCache(
+                                        ProtobufMetadataManagerConstants.PROTOBUF_METADATA_CACHE_NAME);
+                            }
+                            backupProtobufMetadataCache.put(fileName, fileContents);
+                        }
+                    }
+                }
+                cacheManager.switchToDefaultCluster();
+            }
+        });
     }
 
     private void initialize(String infinispanConfigName, Map<String, Properties> properties) {

extensions/infinispan-client/runtime/src/main/java/io/quarkus/infinispan/client/runtime/InfinispanClientRuntimeConfig.java

@@ -283,6 +283,17 @@ public static class BackupClusterConfig {
         // @formatter:on
         @ConfigItem(defaultValue = "HASH_DISTRIBUTION_AWARE")
         Optional<String> clientIntelligence;
+
+        // @formatter:off
+        /**
+         * Enables or disables Protobuf generated schemas upload to the backup.
+         * Set it to 'false' when you need to handle the lifecycle of the Protobuf Schemas on Server side yourself.
+         * Default is 'true'.
+         * This setting will be ignored if the Global Setting is set up to false.
+         */
+        // @formatter:on
+        @ConfigItem(defaultValue = "true")
+        Optional<Boolean> useSchemaRegistration;
     }
 
     @Override