Support reserved state to be spread accross several files

Author: jfreden

server/src/main/java/org/elasticsearch/reservedstate/service/ReservedClusterStateService.java

@@ -187,7 +187,7 @@ public void process(
         process(namespace, stateChunk, versionCheck, errorListener);
     }
 
-    ReservedStateChunk parse(ProjectId projectId, String namespace, XContentParser parser) {
+    public ReservedStateChunk parse(ProjectId projectId, String namespace, XContentParser parser) {
         try {
             return stateChunkParser.apply(parser, null);
         } catch (Exception e) {
@@ -377,6 +377,7 @@ public void onFailure(Exception e) {
      * @param projectId the project state to modify
      * @param namespace the namespace under which we'll store the reserved keys in the cluster state metadata
      * @param reservedStateChunk a {@link ReservedStateChunk} composite state object to process
+     * @param versionCheck  Enum representing whether a reserved state should be processed based on the current and new versions
      * @param errorListener a consumer called with {@link IllegalStateException} if the content has errors and the
      *        cluster state cannot be correctly applied, null if successful or the state failed to apply because of incompatible version.
      */
@@ -387,29 +388,56 @@ public void process(
         ReservedStateVersionCheck versionCheck,
         Consumer<Exception> errorListener
     ) {
-        Map<String, Object> reservedState = reservedStateChunk.state();
-        ReservedStateVersion reservedStateVersion = reservedStateChunk.metadata();
+        process(projectId, namespace, List.of(reservedStateChunk), versionCheck, errorListener);
+    }
 
+    /**
+     * Saves and reserves a chunk of the cluster state under a given 'namespace' from {@link XContentParser}
+     *
+     * @param projectId the project state to modify
+     * @param namespace the namespace under which we'll store the reserved keys in the cluster state metadata
+     * @param reservedStateChunks a {@link ReservedStateChunk} composite state object to process
+     * @param versionCheck  Enum representing whether a reserved state should be processed based on the current and new versions
+     * @param errorListener a consumer called with {@link IllegalStateException} if the content has errors and the
+     *        cluster state cannot be correctly applied, null if successful or the state failed to apply because of incompatible version.
+     */
+    public void process(
+        ProjectId projectId,
+        String namespace,
+        List<ReservedStateChunk> reservedStateChunks,
+        ReservedStateVersionCheck versionCheck,
+        Consumer<Exception> errorListener
+    ) {
+        ReservedStateChunk reservedStateChunk;
+        ReservedStateVersion reservedStateVersion;
         LinkedHashSet<String> orderedHandlers;
+
         try {
+            reservedStateChunk = mergeReservedStateChunks(reservedStateChunks);
+            Map<String, Object> reservedState = reservedStateChunk.state();
+            reservedStateVersion = reservedStateChunk.metadata();
             orderedHandlers = orderedProjectStateHandlers(reservedState.keySet());
         } catch (Exception e) {
             ErrorState errorState = new ErrorState(
                 projectId,
                 namespace,
-                reservedStateVersion.version(),
+                reservedStateChunks.getFirst().metadata().version(),
                 versionCheck,
                 e,
                 ReservedStateErrorMetadata.ErrorKind.PARSING
             );
 
-            updateErrorState(errorState);
-            logger.debug(
-                "error processing project [{}] change request for [{}] with the following errors [{}]",
-                projectId,
-                namespace,
-                errorState
-            );
+            if (clusterService.state().metadata().hasProject(projectId)) {
+                updateErrorState(errorState);
+                logger.debug(
+                    "error processing project [{}] change request for [{}] with the following errors [{}]",
+                    projectId,
+                    namespace,
+                    errorState
+                );
+            } else {
+                logger.info("Error processing project [{}] create request for [{}].", projectId, namespace);
+            }
 
             errorListener.accept(
                 new IllegalStateException(
@@ -476,6 +504,27 @@ public void onFailure(Exception e) {
         );
     }
 
+    private static ReservedStateChunk mergeReservedStateChunks(List<ReservedStateChunk> chunks) {
+        ReservedStateVersion reservedStateVersion = null;
+        Map<String, Object> mergedChunks = new HashMap<>();
+        for (var chunk : chunks) {
+            if (reservedStateVersion == null) {
+                reservedStateVersion = chunk.metadata();
+            } else if (chunk.metadata().equals(reservedStateVersion) == false) {
+                throw new IllegalStateException(
+                    "Failed to merge reserved state chunks because of version mismatch: ["
+                        + reservedStateVersion
+                        + "] != ["
+                        + chunk.metadata()
+                        + "]"
+                );
+            }
+            mergedChunks.putAll(chunk.state());
+        }
+
+        return new ReservedStateChunk(mergedChunks, reservedStateVersion);
+    }
+
     // package private for testing
     Exception checkAndReportError(
         Optional<ProjectId> projectId,