diff --git a/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedFlowSnapshot.java b/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedFlowSnapshot.java
index 76aceab..4611b9e 100644
--- a/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedFlowSnapshot.java
+++ b/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedFlowSnapshot.java
@@ -25,6 +25,7 @@
 import javax.validation.constraints.NotNull;
 import javax.xml.bind.annotation.XmlRootElement;
 import javax.xml.bind.annotation.XmlTransient;
+import java.util.Map;
 import java.util.Objects;
 
 /**
@@ -47,13 +48,21 @@
     @NotNull
     private VersionedProcessGroup flowContents;
 
+    // optional map of external controller service references
+    private Map<String,ExternalControllerServiceReference> externalControllerServices;
+
+    // optional parameter contexts mapped by their name
+    private Map<String,VersionedParameterContext> parameterContexts;;
+
+    // optional encoding version that clients may specify to track how the flow contents are encoded
+    private String flowEncodingVersion;
+
     // read-only, only populated from retrieval of a snapshot
     private VersionedFlow flow;
 
     // read-only, only populated from retrieval of a snapshot
     private Bucket bucket;
 
-
     @ApiModelProperty(value = "The metadata for this snapshot", required = true)
     public VersionedFlowSnapshotMetadata getSnapshotMetadata() {
         return snapshotMetadata;
@@ -72,6 +81,15 @@
         this.flowContents = flowContents;
     }
 
+    @ApiModelProperty("The information about controller services that exist outside this versioned flow, but are referenced by components within the versioned flow.")
+    public Map<String, ExternalControllerServiceReference> getExternalControllerServices() {
+        return externalControllerServices;
+    }
+
+    public void setExternalControllerServices(Map<String, ExternalControllerServiceReference> externalControllerServices) {
+        this.externalControllerServices = externalControllerServices;
+    }
+
     @ApiModelProperty(value = "The flow this snapshot is for", readOnly = true)
     public VersionedFlow getFlow() {
         return flow;
@@ -90,6 +108,26 @@
         this.bucket = bucket;
     }
 
+    @ApiModelProperty(value = "The parameter contexts referenced by process groups in the flow contents. " +
+            "The mapping is from the name of the context to the context instance, and it is expected that any " +
+            "context in this map is referenced by at least one process group in this flow.")
+    public Map<String,VersionedParameterContext> getParameterContexts() {
+        return parameterContexts;
+    }
+
+    public void setParameterContexts(Map<String,VersionedParameterContext> parameterContexts) {
+        this.parameterContexts = parameterContexts;
+    }
+
+    @ApiModelProperty(value = "The optional encoding version of the flow contents.")
+    public String getFlowEncodingVersion() {
+        return flowEncodingVersion;
+    }
+
+    public void setFlowEncodingVersion(String flowEncodingVersion) {
+        this.flowEncodingVersion = flowEncodingVersion;
+    }
+
     /**
      * This is a convenience method that will return true when flow is populated and when the flow's versionCount
      * is equal to the version of this snapshot.
diff --git a/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedParameter.java b/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedParameter.java
new file mode 100644
index 0000000..857dd16
--- /dev/null
+++ b/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedParameter.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.registry.flow;
+
+import io.swagger.annotations.ApiModelProperty;
+
+import java.util.Objects;
+
+public class VersionedParameter {
+
+    private String name;
+    private String description;
+    private boolean sensitive;
+    private String value;
+
+    @ApiModelProperty("The name of the parameter")
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    @ApiModelProperty("The description of the param")
+    public String getDescription() {
+        return description;
+    }
+
+    public void setDescription(String description) {
+        this.description = description;
+    }
+
+    @ApiModelProperty("Whether or not the parameter value is sensitive")
+    public boolean isSensitive() {
+        return sensitive;
+    }
+
+    public void setSensitive(boolean sensitive) {
+        this.sensitive = sensitive;
+    }
+
+    @ApiModelProperty("The value of the parameter")
+    public String getValue() {
+        return value;
+    }
+
+    public void setValue(String value) {
+        this.value = value;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        VersionedParameter that = (VersionedParameter) o;
+        return Objects.equals(name, that.name);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(name);
+    }
+}
diff --git a/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedParameterContext.java b/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedParameterContext.java
new file mode 100644
index 0000000..5294c61
--- /dev/null
+++ b/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedParameterContext.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.registry.flow;
+
+import io.swagger.annotations.ApiModelProperty;
+
+import java.util.Set;
+
+public class VersionedParameterContext {
+
+    private String name;
+    private Set<VersionedParameter> parameters;
+
+    @ApiModelProperty("The name of the context")
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    @ApiModelProperty("The parameters in the context")
+    public Set<VersionedParameter> getParameters() {
+        return parameters;
+    }
+
+    public void setParameters(Set<VersionedParameter> parameters) {
+        this.parameters = parameters;
+    }
+}
diff --git a/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedPort.java b/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedPort.java
index 2b7cccd..947b52e 100644
--- a/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedPort.java
+++ b/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedPort.java
@@ -23,6 +23,7 @@
     private PortType type;
     private Integer concurrentlySchedulableTaskCount;
     private ScheduledState scheduledState;
+    private boolean allowRemoteAccess;
 
     @ApiModelProperty("The number of tasks that should be concurrently scheduled for the port.")
     public Integer getConcurrentlySchedulableTaskCount() {
@@ -51,6 +52,15 @@
         this.scheduledState = scheduledState;
     }
 
+    @ApiModelProperty("Whether or not this port allows remote access for site-to-site")
+    public boolean isAllowRemoteAccess() {
+        return allowRemoteAccess;
+    }
+
+    public void setAllowRemoteAccess(boolean allowRemoteAccess) {
+        this.allowRemoteAccess = allowRemoteAccess;
+    }
+
     @Override
     public ComponentType getComponentType() {
         if (type == PortType.OUTPUT_PORT) {
diff --git a/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedProcessGroup.java b/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedProcessGroup.java
index 071574d..ef0d3fe 100644
--- a/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedProcessGroup.java
+++ b/nifi-registry-core/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedProcessGroup.java
@@ -40,7 +40,8 @@
     private VersionedFlowCoordinates versionedFlowCoordinates = null;
 
     private Map<String, String> variables = new HashMap<>();
-    private Map<String,ExternalControllerServiceReference> externalControllerServices;
+
+    private String parameterContextName;
 
     @ApiModelProperty("The child Process Groups")
     public Set<VersionedProcessGroup> getProcessGroups() {
@@ -146,12 +147,13 @@
         return versionedFlowCoordinates;
     }
 
-    @ApiModelProperty("The information about controller services that exist outside the versioned process group, but are referenced by components within the versioned process group.")
-    public Map<String, ExternalControllerServiceReference> getExternalControllerServices() {
-        return externalControllerServices;
+    @ApiModelProperty("The name of the parameter context used by this process group")
+    public String getParameterContextName() {
+        return parameterContextName;
     }
 
-    public void setExternalControllerServices(Map<String, ExternalControllerServiceReference> externalControllerServices) {
-        this.externalControllerServices = externalControllerServices;
+    public void setParameterContextName(String parameterContextName) {
+        this.parameterContextName = parameterContextName;
     }
+
 }
diff --git a/nifi-registry-core/nifi-registry-framework/pom.xml b/nifi-registry-core/nifi-registry-framework/pom.xml
index f5f1d65..d427678 100644
--- a/nifi-registry-core/nifi-registry-framework/pom.xml
+++ b/nifi-registry-core/nifi-registry-framework/pom.xml
@@ -156,6 +156,7 @@
                         <exclude>src/test/resources/serialization/ver1.snapshot</exclude>
                         <exclude>src/test/resources/serialization/ver2.snapshot</exclude>
                         <exclude>src/test/resources/serialization/ver3.snapshot</exclude>
+                        <exclude>src/test/resources/serialization/ver9999.snapshot</exclude>
                         <exclude>src/test/resources/extensions/ConsumeKafkaRecord_1_0.json</exclude>
                     </excludes>
                 </configuration>
diff --git a/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/FlowContent.java b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/FlowContent.java
new file mode 100644
index 0000000..8340bdd
--- /dev/null
+++ b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/FlowContent.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.registry.serialization;
+
+import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
+
+/**
+ * Wrapper element to contain everything that is serialized for a given version of a flow.
+ */
+public class FlowContent {
+
+    private VersionedFlowSnapshot flowSnapshot;
+
+    public VersionedFlowSnapshot getFlowSnapshot() {
+        return flowSnapshot;
+    }
+
+    public void setFlowSnapshot(VersionedFlowSnapshot flowSnapshot) {
+        this.flowSnapshot = flowSnapshot;
+    }
+}
diff --git a/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/FlowContentSerializer.java b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/FlowContentSerializer.java
new file mode 100644
index 0000000..3466aa9
--- /dev/null
+++ b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/FlowContentSerializer.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.registry.serialization;
+
+import org.apache.nifi.registry.flow.VersionedProcessGroup;
+import org.apache.nifi.registry.serialization.jackson.JacksonFlowContentSerializer;
+import org.apache.nifi.registry.serialization.jackson.JacksonVersionedProcessGroupSerializer;
+import org.apache.nifi.registry.serialization.jaxb.JAXBVersionedProcessGroupSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import java.io.BufferedInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Serializer that handles versioned serialization for flow content.
+ *
+ * <p>
+ * Current data model version is 3.
+ * Data Model Version Histories:
+ * <ul>
+ *     <li>version 3: Serialized by {@link JacksonFlowContentSerializer}</li>
+ *     <li>version 2: Serialized by {@link JacksonVersionedProcessGroupSerializer}</li>
+ *     <li>version 1: Serialized by {@link JAXBVersionedProcessGroupSerializer}</li>
+ * </ul>
+ * </p>
+ */
+@Service
+public class FlowContentSerializer {
+
+    private static final Logger logger = LoggerFactory.getLogger(FlowContentSerializer.class);
+
+    static final Integer START_USING_SNAPSHOT_VERSION = 3;
+    static final Integer CURRENT_DATA_MODEL_VERSION = 3;
+
+    private final Map<Integer, VersionedSerializer<VersionedProcessGroup>> processGroupSerializers;
+    private final Map<Integer, VersionedSerializer<FlowContent>> flowContentSerializers;
+    private final Map<Integer, VersionedSerializer<?>> allSerializers;
+
+    private final List<Integer> descendingVersions;
+
+    public FlowContentSerializer() {
+        final Map<Integer, VersionedSerializer<FlowContent>> tempFlowContentSerializers = new HashMap<>();
+        tempFlowContentSerializers.put(3, new JacksonFlowContentSerializer());
+        flowContentSerializers = Collections.unmodifiableMap(tempFlowContentSerializers);
+
+        final Map<Integer, VersionedSerializer<VersionedProcessGroup>> tempProcessGroupSerializers = new HashMap<>();
+        tempProcessGroupSerializers.put(2, new JacksonVersionedProcessGroupSerializer());
+        tempProcessGroupSerializers.put(1, new JAXBVersionedProcessGroupSerializer());
+        processGroupSerializers = Collections.unmodifiableMap(tempProcessGroupSerializers);
+
+        final Map<Integer,VersionedSerializer<?>> tempAllSerializers = new HashMap<>();
+        tempAllSerializers.putAll(processGroupSerializers);
+        tempAllSerializers.putAll(flowContentSerializers);
+        allSerializers = Collections.unmodifiableMap(tempAllSerializers);
+
+        final List<Integer> sortedVersions = new ArrayList<>(allSerializers.keySet());
+        sortedVersions.sort(Collections.reverseOrder(Integer::compareTo));
+        this.descendingVersions = sortedVersions;
+    }
+
+    /**
+     * Tries to read a data model version using each VersionedSerializer, in descending version order.
+     * If no version could be read from any serializer, then a SerializationException is thrown.
+     *
+     * When deserializing, clients are expected to call this method to obtain the version, then call
+     * {@method isProcessGroupVersion}, which then determines if {@method deserializeProcessGroup}
+     * should be used, or if {@method deserializeFlowContent} should be used.
+     *
+     * @param input the input stream containing serialized flow content
+     * @return the data model version from the input stream
+     * @throws SerializationException if the data model version could not be read with any serializer
+     */
+    public int readDataModelVersion(final InputStream input) throws SerializationException {
+        final InputStream markSupportedInput = input.markSupported() ? input : new BufferedInputStream(input);
+
+        // Mark the beginning of the stream.
+        markSupportedInput.mark(SerializationConstants.MAX_HEADER_BYTES);
+
+        // Try each serializer in descending version order
+        for (final int serializerVersion : descendingVersions) {
+            final VersionedSerializer<?> serializer = allSerializers.get(serializerVersion);
+            try {
+                return serializer.readDataModelVersion(markSupportedInput);
+            } catch (SerializationException e) {
+                if (logger.isDebugEnabled()) {
+                    logger.error("Unable to read the data model version due to: {}", e.getMessage());
+                }
+                continue;
+            } finally {
+                // Reset the stream position.
+                try {
+                    markSupportedInput.reset();
+                } catch (IOException resetException) {
+                    // Should not happen.
+                    logger.error("Unable to reset the input stream.", resetException);
+                }
+            }
+        }
+
+        throw new SerializationException("Unable to read the data model version for the flow content.");
+    }
+
+    public Integer getCurrentDataModelVersion() {
+        return CURRENT_DATA_MODEL_VERSION;
+    }
+
+    public boolean isProcessGroupVersion(final int dataModelVersion) {
+        return dataModelVersion < START_USING_SNAPSHOT_VERSION;
+    }
+
+    public VersionedProcessGroup deserializeProcessGroup(final int dataModelVersion, final InputStream input) throws SerializationException {
+        final VersionedSerializer<VersionedProcessGroup> serializer = processGroupSerializers.get(dataModelVersion);
+        if (serializer == null) {
+            throw new IllegalArgumentException("No VersionedProcessGroup serializer exists for data model version: " + dataModelVersion);
+        }
+
+        return serializer.deserialize(input);
+    }
+
+    public FlowContent deserializeFlowContent(final int dataModelVersion, final InputStream input) throws SerializationException {
+        final VersionedSerializer<FlowContent> serializer = flowContentSerializers.get(dataModelVersion);
+        if (serializer == null) {
+            throw new IllegalArgumentException("No FlowContent serializer exists for data model version: " + dataModelVersion);
+        }
+
+        return serializer.deserialize(input);
+    }
+
+    public void serializeFlowContent(final FlowContent flowContent, final OutputStream out) throws SerializationException {
+        final VersionedSerializer<FlowContent> serializer = flowContentSerializers.get(CURRENT_DATA_MODEL_VERSION);
+        serializer.serialize(CURRENT_DATA_MODEL_VERSION, flowContent, out);
+    }
+}
diff --git a/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/SerializationConstants.java b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/SerializationConstants.java
new file mode 100644
index 0000000..9c9c384
--- /dev/null
+++ b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/SerializationConstants.java
@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.registry.serialization;
+
+public interface SerializationConstants {
+
+    int MAX_HEADER_BYTES = 1024;
+
+}
diff --git a/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/VersionedProcessGroupSerializer.java b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/VersionedProcessGroupSerializer.java
deleted file mode 100644
index 8d7cb7b..0000000
--- a/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/VersionedProcessGroupSerializer.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.registry.serialization;
-
-import org.apache.nifi.registry.flow.VersionedProcessGroup;
-import org.apache.nifi.registry.serialization.jackson.JacksonVersionedProcessGroupSerializer;
-import org.apache.nifi.registry.serialization.jaxb.JAXBVersionedProcessGroupSerializer;
-import org.springframework.stereotype.Service;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * See {@link AbstractMultiVersionSerializer} for further information.
- *
- * <p>
- * Current data model version is 2.
- * Data Model Version Histories:
- * <ul>
- *     <li>version 2: Serialized by {@link JacksonVersionedProcessGroupSerializer}</li>
- *     <li>version 1: Serialized by {@link JAXBVersionedProcessGroupSerializer}</li>
- * </ul>
- * </p>
- */
-@Service
-public class VersionedProcessGroupSerializer extends AbstractMultiVersionSerializer<VersionedProcessGroup> {
-
-    static final Integer CURRENT_DATA_MODEL_VERSION = 2;
-
-    @Override
-    protected Map<Integer, VersionedSerializer<VersionedProcessGroup>> createVersionedSerializers() {
-        final Map<Integer, VersionedSerializer<VersionedProcessGroup>> tempSerializers = new HashMap<>();
-        tempSerializers.put(2, new JacksonVersionedProcessGroupSerializer());
-        tempSerializers.put(1, new JAXBVersionedProcessGroupSerializer());
-        return tempSerializers;
-    }
-
-    @Override
-    protected int getCurrentDataModelVersion() {
-        return CURRENT_DATA_MODEL_VERSION;
-    }
-
-}
diff --git a/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/jackson/JacksonFlowContentSerializer.java b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/jackson/JacksonFlowContentSerializer.java
new file mode 100644
index 0000000..77dd37a
--- /dev/null
+++ b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/jackson/JacksonFlowContentSerializer.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.registry.serialization.jackson;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.nifi.registry.serialization.FlowContent;
+import org.apache.nifi.registry.serialization.SerializationException;
+
+/**
+ * A Jackson serializer for FlowContent.
+ */
+public class JacksonFlowContentSerializer extends JacksonSerializer<FlowContent> {
+
+    @Override
+    TypeReference<SerializationContainer<FlowContent>> getDeserializeTypeRef() throws SerializationException {
+        return new TypeReference<SerializationContainer<org.apache.nifi.registry.serialization.FlowContent>>() {};
+    }
+}
diff --git a/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/jackson/JacksonSerializer.java b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/jackson/JacksonSerializer.java
index 4098c77..d159ea3 100644
--- a/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/jackson/JacksonSerializer.java
+++ b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/jackson/JacksonSerializer.java
@@ -18,6 +18,7 @@
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.registry.serialization.SerializationConstants;
 import org.apache.nifi.registry.serialization.SerializationException;
 import org.apache.nifi.registry.serialization.VersionedSerializer;
 import org.slf4j.Logger;
@@ -30,8 +31,6 @@
 import java.util.Collections;
 import java.util.HashMap;
 
-import static org.apache.nifi.registry.serialization.VersionedProcessGroupSerializer.MAX_HEADER_BYTES;
-
 /**
  * A Serializer that uses Jackson for serializing/deserializing.
  */
@@ -80,7 +79,7 @@
 
     @Override
     public int readDataModelVersion(InputStream input) throws SerializationException {
-        final byte[] headerBytes = new byte[MAX_HEADER_BYTES];
+        final byte[] headerBytes = new byte[SerializationConstants.MAX_HEADER_BYTES];
         final int readHeaderBytes;
         try {
             readHeaderBytes = input.read(headerBytes);
diff --git a/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/jackson/JacksonVersionedProcessGroupSerializer.java b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/jackson/JacksonVersionedProcessGroupSerializer.java
index 21bdecc..4bd763e 100644
--- a/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/jackson/JacksonVersionedProcessGroupSerializer.java
+++ b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/serialization/jackson/JacksonVersionedProcessGroupSerializer.java
@@ -21,11 +21,10 @@
 import org.apache.nifi.registry.serialization.SerializationException;
 
 /**
- * A Jackson serializer for VersionedFlowSnapshots.
+ * A Jackson serializer for VersionedProcessGroups.
  */
 public class JacksonVersionedProcessGroupSerializer extends JacksonSerializer<VersionedProcessGroup> {
 
-
     @Override
     TypeReference<SerializationContainer<VersionedProcessGroup>> getDeserializeTypeRef() throws SerializationException {
         return new TypeReference<SerializationContainer<VersionedProcessGroup>>() {};
diff --git a/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/service/RegistryService.java b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/service/RegistryService.java
index 7a87ec0..26057ae 100644
--- a/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/service/RegistryService.java
+++ b/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/service/RegistryService.java
@@ -37,10 +37,10 @@
 import org.apache.nifi.registry.extension.bundle.BundleVersion;
 import org.apache.nifi.registry.extension.bundle.BundleVersionFilterParams;
 import org.apache.nifi.registry.extension.bundle.BundleVersionMetadata;
-import org.apache.nifi.registry.extension.component.manifest.Extension;
 import org.apache.nifi.registry.extension.component.ExtensionFilterParams;
 import org.apache.nifi.registry.extension.component.ExtensionMetadata;
 import org.apache.nifi.registry.extension.component.TagCount;
+import org.apache.nifi.registry.extension.component.manifest.Extension;
 import org.apache.nifi.registry.extension.component.manifest.ProvidedServiceAPI;
 import org.apache.nifi.registry.extension.repo.ExtensionRepoArtifact;
 import org.apache.nifi.registry.extension.repo.ExtensionRepoBucket;
@@ -62,7 +62,8 @@
 import org.apache.nifi.registry.flow.diff.StandardFlowComparator;
 import org.apache.nifi.registry.provider.extension.StandardBundleCoordinate;
 import org.apache.nifi.registry.provider.flow.StandardFlowSnapshotContext;
-import org.apache.nifi.registry.serialization.Serializer;
+import org.apache.nifi.registry.serialization.FlowContent;
+import org.apache.nifi.registry.serialization.FlowContentSerializer;
 import org.apache.nifi.registry.service.alias.RegistryUrlAliasService;
 import org.apache.nifi.registry.service.extension.ExtensionService;
 import org.apache.nifi.registry.service.mapper.BucketMappings;
@@ -111,7 +112,7 @@
     private final MetadataService metadataService;
     private final FlowPersistenceProvider flowPersistenceProvider;
     private final BundlePersistenceProvider bundlePersistenceProvider;
-    private final Serializer<VersionedProcessGroup> processGroupSerializer;
+    private final FlowContentSerializer flowContentSerializer;
     private final ExtensionService extensionService;
     private final Validator validator;
     private final RegistryUrlAliasService registryUrlAliasService;
@@ -124,14 +125,14 @@
     public RegistryService(final MetadataService metadataService,
                            final FlowPersistenceProvider flowPersistenceProvider,
                            final BundlePersistenceProvider bundlePersistenceProvider,
-                           final Serializer<VersionedProcessGroup> processGroupSerializer,
+                           final FlowContentSerializer flowContentSerializer,
                            final ExtensionService extensionService,
                            final Validator validator,
                            final RegistryUrlAliasService registryUrlAliasService) {
         this.metadataService = Validate.notNull(metadataService);
         this.flowPersistenceProvider = Validate.notNull(flowPersistenceProvider);
         this.bundlePersistenceProvider = Validate.notNull(bundlePersistenceProvider);
-        this.processGroupSerializer = Validate.notNull(processGroupSerializer);
+        this.flowContentSerializer = Validate.notNull(flowContentSerializer);
         this.extensionService = Validate.notNull(extensionService);
         this.validator = Validate.notNull(validator);
         this.registryUrlAliasService = Validate.notNull(registryUrlAliasService);
@@ -688,7 +689,14 @@
             // serialize the snapshot
             final ByteArrayOutputStream out = new ByteArrayOutputStream();
             registryUrlAliasService.setInternal(flowSnapshot.getFlowContents());
-            processGroupSerializer.serialize(flowSnapshot.getFlowContents(), out);
+
+            final FlowContent flowContent = new FlowContent();
+            flowContent.setFlowSnapshot(flowSnapshot);
+
+            // temporarily remove the metadata so it isn't serialized, but then put it back for returning the response
+            flowSnapshot.setSnapshotMetadata(null);
+            flowContentSerializer.serializeFlowContent(flowContent, out);
+            flowSnapshot.setSnapshotMetadata(snapshotMetadata);
 
             // save the serialized snapshot to the persistence provider
             final Bucket bucket = BucketMappings.map(existingBucket);
@@ -772,9 +780,9 @@
                     + flowEntity.getId() + " and version " + version);
         }
 
-        // deserialize the contents
+        // deserialize the content
         final InputStream input = new ByteArrayInputStream(serializedSnapshot);
-        final VersionedProcessGroup flowContents = processGroupSerializer.deserialize(input);
+        final VersionedFlowSnapshot snapshot = deserializeFlowContent(input);
 
         // map entities to data model
         final Bucket bucket = BucketMappings.map(bucketEntity);
@@ -782,15 +790,29 @@
         final VersionedFlowSnapshotMetadata snapshotMetadata = FlowMappings.map(bucketEntity, snapshotEntity);
 
         // create the snapshot to return
-        final VersionedFlowSnapshot snapshot = new VersionedFlowSnapshot();
-        registryUrlAliasService.setExternal(flowContents);
-        snapshot.setFlowContents(flowContents);
+        registryUrlAliasService.setExternal(snapshot.getFlowContents());
         snapshot.setSnapshotMetadata(snapshotMetadata);
         snapshot.setFlow(versionedFlow);
         snapshot.setBucket(bucket);
         return snapshot;
     }
 
+    private VersionedFlowSnapshot deserializeFlowContent(final InputStream input) {
+        // attempt to read the version header from the serialized content
+        final int dataModelVersion = flowContentSerializer.readDataModelVersion(input);
+
+        // determine how to do deserialize based on the data model version
+        if (flowContentSerializer.isProcessGroupVersion(dataModelVersion)) {
+            final VersionedProcessGroup processGroup = flowContentSerializer.deserializeProcessGroup(dataModelVersion, input);
+            final VersionedFlowSnapshot snapshot = new VersionedFlowSnapshot();
+            snapshot.setFlowContents(processGroup);
+            return snapshot;
+        } else {
+            final FlowContent flowContent = flowContentSerializer.deserializeFlowContent(dataModelVersion, input);
+            return flowContent.getFlowSnapshot();
+        }
+    }
+
     /**
      * Returns all versions of a flow, sorted newest to oldest.
      *
@@ -1009,9 +1031,12 @@
 
             // deserialize the contents
             final InputStream inputA = new ByteArrayInputStream(serializedSnapshotA);
-            final VersionedProcessGroup flowContentsA = processGroupSerializer.deserialize(inputA);
+            final VersionedFlowSnapshot snapshotA = deserializeFlowContent(inputA);
+            final VersionedProcessGroup flowContentsA = snapshotA.getFlowContents();
+
             final InputStream inputB = new ByteArrayInputStream(serializedSnapshotB);
-            final VersionedProcessGroup flowContentsB = processGroupSerializer.deserialize(inputB);
+            final VersionedFlowSnapshot snapshotB = deserializeFlowContent(inputB);
+            final VersionedProcessGroup flowContentsB = snapshotB.getFlowContents();
 
             final ComparableDataFlow comparableFlowA = new StandardComparableDataFlow(String.format("Version %d", older), flowContentsA);
             final ComparableDataFlow comparableFlowB = new StandardComparableDataFlow(String.format("Version %d", newer), flowContentsB);
@@ -1021,13 +1046,13 @@
                     null, new ConciseEvolvingDifferenceDescriptor());
             final FlowComparison flowComparison = flowComparator.compare();
 
-            VersionedFlowDifference result = new VersionedFlowDifference();
+            final VersionedFlowDifference result = new VersionedFlowDifference();
             result.setBucketId(bucketIdentifier);
             result.setFlowId(flowIdentifier);
             result.setVersionA(older);
             result.setVersionB(newer);
 
-            Set<ComponentDifferenceGroup> differenceGroups = getStringComponentDifferenceGroupMap(flowComparison.getDifferences());
+            final Set<ComponentDifferenceGroup> differenceGroups = getStringComponentDifferenceGroupMap(flowComparison.getDifferences());
             result.setComponentDifferenceGroups(differenceGroups);
 
             return result;
diff --git a/nifi-registry-core/nifi-registry-framework/src/test/java/org/apache/nifi/registry/serialization/TestFlowContentSerializer.java b/nifi-registry-core/nifi-registry-framework/src/test/java/org/apache/nifi/registry/serialization/TestFlowContentSerializer.java
new file mode 100644
index 0000000..21d3b43
--- /dev/null
+++ b/nifi-registry-core/nifi-registry-framework/src/test/java/org/apache/nifi/registry/serialization/TestFlowContentSerializer.java
@@ -0,0 +1,282 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.registry.serialization;
+
+import org.apache.nifi.registry.flow.ExternalControllerServiceReference;
+import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
+import org.apache.nifi.registry.flow.VersionedProcessGroup;
+import org.apache.nifi.registry.flow.VersionedProcessor;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+import static org.junit.Assert.assertTrue;
+
+public class TestFlowContentSerializer {
+
+    private FlowContentSerializer serializer;
+
+    @Before
+    public void setup() {
+        serializer = new FlowContentSerializer();
+    }
+
+    @Test
+    public void testSerializeDeserializeFlowContent() {
+        final VersionedProcessor processor1 = new VersionedProcessor();
+        processor1.setIdentifier("processor1");
+        processor1.setName("My Processor 1");
+
+        final VersionedProcessGroup processGroup1 = new VersionedProcessGroup();
+        processGroup1.setIdentifier("pg1");
+        processGroup1.setName("My Process Group");
+        processGroup1.getProcessors().add(processor1);
+
+        final VersionedFlowSnapshot snapshot = new VersionedFlowSnapshot();
+        snapshot.setFlowContents(processGroup1);
+
+        final FlowContent flowContent = new FlowContent();
+        flowContent.setFlowSnapshot(snapshot);
+
+        final ByteArrayOutputStream out = new ByteArrayOutputStream();
+        serializer.serializeFlowContent(flowContent, out);
+
+        //final String json = new String(out.toByteArray(), StandardCharsets.UTF_8);
+        //System.out.println(json);
+
+        final ByteArrayInputStream in = new ByteArrayInputStream(out.toByteArray());
+
+        // make sure we can read the version from the input stream and it should be the current version
+        final Integer version = serializer.readDataModelVersion(in);
+        assertEquals(serializer.getCurrentDataModelVersion(), version);
+        assertEquals(false, serializer.isProcessGroupVersion(version));
+
+        // make sure we can deserialize back to FlowContent
+        final FlowContent deserializedFlowContent = serializer.deserializeFlowContent(version, in);
+        assertNotNull(deserializedFlowContent);
+
+        final VersionedFlowSnapshot deserializedSnapshot = deserializedFlowContent.getFlowSnapshot();
+        assertNotNull(deserializedSnapshot);
+
+        final VersionedProcessGroup deserializedProcessGroup1 = deserializedSnapshot.getFlowContents();
+        assertNotNull(deserializedProcessGroup1);
+        assertEquals(processGroup1.getIdentifier(), deserializedProcessGroup1.getIdentifier());
+        assertEquals(processGroup1.getName(), deserializedProcessGroup1.getName());
+
+        assertEquals(1, deserializedProcessGroup1.getProcessors().size());
+
+        final VersionedProcessor deserializedProcessor1 = deserializedProcessGroup1.getProcessors().iterator().next();
+        assertEquals(processor1.getIdentifier(), deserializedProcessor1.getIdentifier());
+        assertEquals(processor1.getName(), deserializedProcessor1.getName());
+    }
+
+    @Test
+    public void testSerializeDeserializeWithExternalServices() throws SerializationException {
+        final VersionedProcessGroup processGroup1 = new VersionedProcessGroup();
+        processGroup1.setIdentifier("pg1");
+        processGroup1.setName("My Process Group");
+
+        final ExternalControllerServiceReference serviceReference1 = new ExternalControllerServiceReference();
+        serviceReference1.setIdentifier("1");
+        serviceReference1.setName("Service 1");
+
+        final ExternalControllerServiceReference serviceReference2 = new ExternalControllerServiceReference();
+        serviceReference2.setIdentifier("2");
+        serviceReference2.setName("Service 2");
+
+        final Map<String,ExternalControllerServiceReference> serviceReferences = new HashMap<>();
+        serviceReferences.put(serviceReference1.getIdentifier(), serviceReference1);
+        serviceReferences.put(serviceReference2.getIdentifier(), serviceReference2);
+
+        final VersionedFlowSnapshot snapshot = new VersionedFlowSnapshot();
+        snapshot.setFlowContents(processGroup1);
+        snapshot.setExternalControllerServices(serviceReferences);
+
+        final FlowContent flowContent = new FlowContent();
+        flowContent.setFlowSnapshot(snapshot);
+
+        final ByteArrayOutputStream out = new ByteArrayOutputStream();
+        serializer.serializeFlowContent(flowContent, out);
+
+        final ByteArrayInputStream in = new ByteArrayInputStream(out.toByteArray());
+
+        // make sure we can read the version from the input stream and it should be the current version
+        final Integer version = serializer.readDataModelVersion(in);
+        assertEquals(serializer.getCurrentDataModelVersion(), version);
+
+        // make sure we can deserialize back to FlowContent
+        final FlowContent deserializedFlowContent = serializer.deserializeFlowContent(version, in);
+        assertNotNull(deserializedFlowContent);
+
+        final VersionedFlowSnapshot deserializedSnapshot = deserializedFlowContent.getFlowSnapshot();
+        assertNotNull(deserializedSnapshot);
+
+        final VersionedProcessGroup deserializedProcessGroup = deserializedSnapshot.getFlowContents();
+        assertEquals(processGroup1.getIdentifier(), deserializedProcessGroup.getIdentifier());
+        assertEquals(processGroup1.getName(), deserializedProcessGroup.getName());
+
+        final Map<String,ExternalControllerServiceReference> deserializedServiceReferences = deserializedSnapshot.getExternalControllerServices();
+        assertNotNull(deserializedServiceReferences);
+        assertEquals(2, deserializedServiceReferences.size());
+
+        final ExternalControllerServiceReference deserializedServiceReference1 = deserializedServiceReferences.get(serviceReference1.getIdentifier());
+        assertNotNull(deserializedServiceReference1);
+        assertEquals(serviceReference1.getIdentifier(), deserializedServiceReference1.getIdentifier());
+        assertEquals(serviceReference1.getName(), deserializedServiceReference1.getName());
+    }
+
+    @Test
+    public void testDeserializeJsonNonIntegerVersion() throws IOException {
+        final String file = "/serialization/json/non-integer-version.snapshot";
+        try (final InputStream is = this.getClass().getResourceAsStream(file)) {
+            try {
+                serializer.readDataModelVersion(is);
+                fail("Should fail");
+            } catch (SerializationException e) {
+                assertEquals("Unable to read the data model version for the flow content.", e.getMessage());
+            }
+        }
+    }
+
+    @Test
+    public void testDeserializeJsonNoVersion() throws IOException {
+        final String file = "/serialization/json/no-version.snapshot";
+        try (final InputStream is = this.getClass().getResourceAsStream(file)) {
+            try {
+                serializer.readDataModelVersion(is);
+                fail("Should fail");
+            } catch (SerializationException e) {
+                assertEquals("Unable to read the data model version for the flow content.", e.getMessage());
+            }
+        }
+    }
+
+    @Test
+    public void testDeserializeVer1() throws IOException {
+        final String file = "/serialization/ver1.snapshot";
+        final VersionedProcessGroup processGroup;
+        try (final InputStream is = this.getClass().getResourceAsStream(file)) {
+            final Integer version = serializer.readDataModelVersion(is);
+            assertNotNull(version);
+            assertEquals(1, version.intValue());
+
+            if (serializer.isProcessGroupVersion(version)) {
+                processGroup = serializer.deserializeProcessGroup(version, is);
+            } else {
+                processGroup = null;
+            }
+        }
+
+        assertNotNull(processGroup);
+        assertNotNull(processGroup.getProcessors());
+        assertTrue(processGroup.getProcessors().size() > 0);
+        //System.out.printf("processGroup=" + processGroup);
+    }
+
+    @Test
+    public void testDeserializeVer2() throws IOException {
+        final String file = "/serialization/ver2.snapshot";
+        final VersionedProcessGroup processGroup;
+        try (final InputStream is = this.getClass().getResourceAsStream(file)) {
+            final Integer version = serializer.readDataModelVersion(is);
+            assertNotNull(version);
+            assertEquals(2, version.intValue());
+
+            if (serializer.isProcessGroupVersion(version)) {
+                processGroup = serializer.deserializeProcessGroup(version, is);
+            } else {
+                processGroup = null;
+            }
+        }
+
+        assertNotNull(processGroup);
+        assertNotNull(processGroup.getProcessors());
+        assertTrue(processGroup.getProcessors().size() > 0);
+        //System.out.printf("processGroup=" + processGroup);
+    }
+
+    @Test
+    public void testDeserializeVer3() throws IOException {
+        final String file = "/serialization/ver3.snapshot";
+        try (final InputStream is = this.getClass().getResourceAsStream(file)) {
+            final Integer version = serializer.readDataModelVersion(is);
+            assertNotNull(version);
+            assertEquals(3, version.intValue());
+            assertFalse(serializer.isProcessGroupVersion(version));
+
+            final FlowContent flowContent = serializer.deserializeFlowContent(version, is);
+            assertNotNull(flowContent);
+
+            final VersionedFlowSnapshot flowSnapshot = flowContent.getFlowSnapshot();
+            assertNotNull(flowSnapshot);
+
+            final VersionedProcessGroup processGroup = flowSnapshot.getFlowContents();
+            assertNotNull(processGroup);
+            assertNotNull(processGroup.getProcessors());
+            assertEquals(1, processGroup.getProcessors().size());
+        }
+    }
+
+    @Test
+    public void testDeserializeVer9999() throws IOException {
+        final String file = "/serialization/ver9999.snapshot";
+        try (final InputStream is = this.getClass().getResourceAsStream(file)) {
+            final Integer version = serializer.readDataModelVersion(is);
+            assertNotNull(version);
+            assertEquals(9999, version.intValue());
+            assertFalse(serializer.isProcessGroupVersion(version));
+
+            try {
+                serializer.deserializeFlowContent(version, is);
+                fail("Should fail");
+            } catch (IllegalArgumentException e) {
+                assertEquals("No FlowContent serializer exists for data model version: " + version, e.getMessage());
+            }
+
+            try {
+                serializer.deserializeProcessGroup(version, is);
+                fail("Should fail");
+            } catch (IllegalArgumentException e) {
+                assertEquals("No VersionedProcessGroup serializer exists for data model version: " + version, e.getMessage());
+            }
+        }
+    }
+
+    @Test
+    public void testDeserializeProcessGroupAsFlowContent() throws IOException {
+        final String file = "/serialization/ver2.snapshot";
+        try (final InputStream is = this.getClass().getResourceAsStream(file)) {
+            final Integer version = serializer.readDataModelVersion(is);
+            assertNotNull(version);
+            assertEquals(2, version.intValue());
+            assertTrue(serializer.isProcessGroupVersion(version));
+
+            final VersionedProcessGroup processGroup = serializer.deserializeProcessGroup(version, is);
+            assertNotNull(processGroup);
+        }
+    }
+}
diff --git a/nifi-registry-core/nifi-registry-framework/src/test/java/org/apache/nifi/registry/serialization/TestVersionedProcessGroupSerializer.java b/nifi-registry-core/nifi-registry-framework/src/test/java/org/apache/nifi/registry/serialization/TestVersionedProcessGroupSerializer.java
deleted file mode 100644
index 145204b..0000000
--- a/nifi-registry-core/nifi-registry-framework/src/test/java/org/apache/nifi/registry/serialization/TestVersionedProcessGroupSerializer.java
+++ /dev/null
@@ -1,174 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.registry.serialization;
-
-import org.apache.nifi.registry.flow.ExternalControllerServiceReference;
-import org.apache.nifi.registry.flow.VersionedProcessGroup;
-import org.apache.nifi.registry.flow.VersionedProcessor;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.HashMap;
-import java.util.Map;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-
-public class TestVersionedProcessGroupSerializer {
-
-    @Test
-    public void testSerializeDeserializeFlowSnapshot() throws SerializationException {
-        final Serializer<VersionedProcessGroup> serializer = new VersionedProcessGroupSerializer();
-
-        final VersionedProcessGroup processGroup1 = new VersionedProcessGroup();
-        processGroup1.setIdentifier("pg1");
-        processGroup1.setName("My Process Group");
-
-        final VersionedProcessor processor1 = new VersionedProcessor();
-        processor1.setIdentifier("processor1");
-        processor1.setName("My Processor 1");
-
-        // make sure nested objects are serialized/deserialized
-        processGroup1.getProcessors().add(processor1);
-
-        final ByteArrayOutputStream out = new ByteArrayOutputStream();
-        serializer.serialize(processGroup1, out);
-
-        final ByteArrayInputStream in = new ByteArrayInputStream(out.toByteArray());
-        final VersionedProcessGroup deserializedProcessGroup1 = serializer.deserialize(in);
-
-        Assert.assertEquals(processGroup1.getIdentifier(), deserializedProcessGroup1.getIdentifier());
-        Assert.assertEquals(processGroup1.getName(), deserializedProcessGroup1.getName());
-
-        Assert.assertEquals(1, deserializedProcessGroup1.getProcessors().size());
-
-        final VersionedProcessor deserializedProcessor1 = deserializedProcessGroup1.getProcessors().iterator().next();
-        Assert.assertEquals(processor1.getIdentifier(), deserializedProcessor1.getIdentifier());
-        Assert.assertEquals(processor1.getName(), deserializedProcessor1.getName());
-
-    }
-
-    @Test
-    public void testSerializeDeserializeWithExternalServices() throws SerializationException {
-        final Serializer<VersionedProcessGroup> serializer = new VersionedProcessGroupSerializer();
-
-        final VersionedProcessGroup processGroup1 = new VersionedProcessGroup();
-        processGroup1.setIdentifier("pg1");
-        processGroup1.setName("My Process Group");
-
-        final ExternalControllerServiceReference serviceReference1 = new ExternalControllerServiceReference();
-        serviceReference1.setIdentifier("1");
-        serviceReference1.setName("Service 1");
-
-        final ExternalControllerServiceReference serviceReference2 = new ExternalControllerServiceReference();
-        serviceReference2.setIdentifier("2");
-        serviceReference2.setName("Service 2");
-
-        final Map<String,ExternalControllerServiceReference> serviceReferences = new HashMap<>();
-        serviceReferences.put(serviceReference1.getIdentifier(), serviceReference1);
-        serviceReferences.put(serviceReference2.getIdentifier(), serviceReference2);
-
-        processGroup1.setExternalControllerServices(serviceReferences);
-
-        final ByteArrayOutputStream out = new ByteArrayOutputStream();
-        serializer.serialize(processGroup1, out);
-
-        final ByteArrayInputStream in = new ByteArrayInputStream(out.toByteArray());
-        final VersionedProcessGroup deserializedProcessGroup = serializer.deserialize(in);
-
-        Assert.assertEquals(processGroup1.getIdentifier(), deserializedProcessGroup.getIdentifier());
-        Assert.assertEquals(processGroup1.getName(), deserializedProcessGroup.getName());
-
-        final Map<String,ExternalControllerServiceReference> deserializedServiceReferences = deserializedProcessGroup.getExternalControllerServices();
-        Assert.assertNotNull(deserializedServiceReferences);
-        Assert.assertEquals(2, deserializedServiceReferences.size());
-
-        final ExternalControllerServiceReference deserializedServiceReference1 = deserializedServiceReferences.get(serviceReference1.getIdentifier());
-        Assert.assertNotNull(deserializedServiceReference1);
-        Assert.assertEquals(serviceReference1.getIdentifier(), deserializedServiceReference1.getIdentifier());
-        Assert.assertEquals(serviceReference1.getName(), deserializedServiceReference1.getName());
-    }
-
-    @Test
-    public void testDeserializeJsonNonIntegerVersion() throws IOException {
-        final String file = "/serialization/json/non-integer-version.snapshot";
-        final VersionedProcessGroupSerializer serializer = new VersionedProcessGroupSerializer();
-        try (final InputStream is = this.getClass().getResourceAsStream(file)) {
-            try {
-                serializer.deserialize(is);
-                fail("Should fail");
-            } catch (SerializationException e) {
-                assertEquals("Unable to find a serializer compatible with the input.", e.getMessage());
-            }
-        }
-    }
-
-    @Test
-    public void testDeserializeJsonNoVersion() throws IOException {
-        final String file = "/serialization/json/no-version.snapshot";
-        final VersionedProcessGroupSerializer serializer = new VersionedProcessGroupSerializer();
-        try (final InputStream is = this.getClass().getResourceAsStream(file)) {
-            try {
-                serializer.deserialize(is);
-                fail("Should fail");
-            } catch (SerializationException e) {
-                assertEquals("Unable to find a serializer compatible with the input.", e.getMessage());
-            }
-        }
-    }
-
-    @Test
-    public void testDeserializeVer1() throws IOException {
-        final String file = "/serialization/ver1.snapshot";
-        final VersionedProcessGroupSerializer serializer = new VersionedProcessGroupSerializer();
-        final VersionedProcessGroup processGroup;
-        try (final InputStream is = this.getClass().getResourceAsStream(file)) {
-            processGroup = serializer.deserialize(is);
-        }
-        System.out.printf("processGroup=" + processGroup);
-    }
-
-    @Test
-    public void testDeserializeVer2() throws IOException {
-        final String file = "/serialization/ver2.snapshot";
-        final VersionedProcessGroupSerializer serializer = new VersionedProcessGroupSerializer();
-        final VersionedProcessGroup processGroup;
-        try (final InputStream is = this.getClass().getResourceAsStream(file)) {
-            processGroup = serializer.deserialize(is);
-        }
-        System.out.printf("processGroup=" + processGroup);
-    }
-
-    @Test
-    public void testDeserializeVer3() throws IOException {
-        final String file = "/serialization/ver3.snapshot";
-        final VersionedProcessGroupSerializer serializer = new VersionedProcessGroupSerializer();
-        try (final InputStream is = this.getClass().getResourceAsStream(file)) {
-            try {
-                serializer.deserialize(is);
-                fail("Should fail");
-            } catch (SerializationException e) {
-                assertEquals("Unable to find a serializer compatible with the input.", e.getMessage());
-            }
-        }
-    }
-
-}
diff --git a/nifi-registry-core/nifi-registry-framework/src/test/java/org/apache/nifi/registry/service/TestRegistryService.java b/nifi-registry-core/nifi-registry-framework/src/test/java/org/apache/nifi/registry/service/TestRegistryService.java
index baeb949..4e43645 100644
--- a/nifi-registry-core/nifi-registry-framework/src/test/java/org/apache/nifi/registry/service/TestRegistryService.java
+++ b/nifi-registry-core/nifi-registry-framework/src/test/java/org/apache/nifi/registry/service/TestRegistryService.java
@@ -31,8 +31,8 @@
 import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata;
 import org.apache.nifi.registry.flow.VersionedProcessGroup;
 import org.apache.nifi.registry.flow.VersionedProcessor;
-import org.apache.nifi.registry.serialization.Serializer;
-import org.apache.nifi.registry.serialization.VersionedProcessGroupSerializer;
+import org.apache.nifi.registry.serialization.FlowContent;
+import org.apache.nifi.registry.serialization.FlowContentSerializer;
 import org.apache.nifi.registry.service.alias.RegistryUrlAliasService;
 import org.apache.nifi.registry.service.extension.ExtensionService;
 import org.apache.nifi.registry.service.extension.StandardExtensionService;
@@ -77,7 +77,7 @@
     private MetadataService metadataService;
     private FlowPersistenceProvider flowPersistenceProvider;
     private BundlePersistenceProvider bundlePersistenceProvider;
-    private Serializer<VersionedProcessGroup> snapshotSerializer;
+    private FlowContentSerializer flowContentSerializer;
     private ExtensionService extensionService;
     private Validator validator;
     private RegistryUrlAliasService registryUrlAliasService;
@@ -89,7 +89,7 @@
         metadataService = mock(MetadataService.class);
         flowPersistenceProvider = mock(FlowPersistenceProvider.class);
         bundlePersistenceProvider = mock(BundlePersistenceProvider.class);
-        snapshotSerializer = mock(VersionedProcessGroupSerializer.class);
+        flowContentSerializer = mock(FlowContentSerializer.class);
         extensionService = mock(StandardExtensionService.class);
         registryUrlAliasService = mock(RegistryUrlAliasService.class);
 
@@ -97,7 +97,7 @@
         validator = validatorFactory.getValidator();
 
         registryService = new RegistryService(metadataService, flowPersistenceProvider, bundlePersistenceProvider,
-                snapshotSerializer, extensionService, validator, registryUrlAliasService);
+                flowContentSerializer, extensionService, validator, registryUrlAliasService);
     }
 
     // ---------------------- Test Bucket methods ---------------------------------------------
@@ -800,7 +800,7 @@
         assertNotNull(createdSnapshot.getFlow());
         assertNotNull(createdSnapshot.getBucket());
 
-        verify(snapshotSerializer, times(1)).serialize(eq(snapshot.getFlowContents()), any(OutputStream.class));
+        verify(flowContentSerializer, times(1)).serializeFlowContent(any(FlowContent.class), any(OutputStream.class));
         verify(flowPersistenceProvider, times(1)).saveFlowContent(any(), any());
         verify(metadataService, times(1)).createFlowSnapshot(any(FlowSnapshotEntity.class));
     }
@@ -1150,8 +1150,10 @@
                 existingSnapshot.getVersion()
         )).thenReturn(new byte[10]);
 
-        final VersionedFlowSnapshot snapshotToDeserialize = createSnapshot();
-        when(snapshotSerializer.deserialize(any(InputStream.class))).thenReturn(snapshotToDeserialize.getFlowContents());
+        final FlowContent flowContent = new FlowContent();
+        flowContent.setFlowSnapshot(createSnapshot());
+        when(flowContentSerializer.readDataModelVersion(any(InputStream.class))).thenReturn(3);
+        when(flowContentSerializer.deserializeFlowContent(eq(3), any(InputStream.class))).thenReturn(flowContent);
 
         final VersionedFlowSnapshot returnedSnapshot = registryService.getFlowSnapshot(
                 existingBucket.getId(), existingSnapshot.getFlowId(), existingSnapshot.getVersion());
@@ -1253,7 +1255,9 @@
 
         final VersionedProcessGroup pgA = createVersionedProcessGroupA();
         final VersionedProcessGroup pgB = createVersionedProcessGroupB();
-        when(snapshotSerializer.deserialize(any())).thenReturn(pgA, pgB);
+        when(flowContentSerializer.readDataModelVersion(any(InputStream.class))).thenReturn(2);
+        when(flowContentSerializer.isProcessGroupVersion(eq(2))).thenReturn(true);
+        when(flowContentSerializer.deserializeProcessGroup(eq(2),any())).thenReturn(pgA, pgB);
 
         final VersionedFlowDifference diff = registryService.getFlowDiff(
                 "bucketIdentifier", "flowIdentifier", 1, 2);
@@ -1274,7 +1278,9 @@
 
         final VersionedProcessGroup pgA = createVersionedProcessGroupA();
         final VersionedProcessGroup pgB = createVersionedProcessGroupB();
-        when(snapshotSerializer.deserialize(any())).thenReturn(pgA, pgB);
+        when(flowContentSerializer.readDataModelVersion(any(InputStream.class))).thenReturn(2);
+        when(flowContentSerializer.isProcessGroupVersion(eq(2))).thenReturn(true);
+        when(flowContentSerializer.deserializeProcessGroup(eq(2),any())).thenReturn(pgA, pgB);
 
         // getFlowDiff orders the changes in ascending order of version number regardless of param order
         final VersionedFlowDifference diff = registryService.getFlowDiff(
diff --git a/nifi-registry-core/nifi-registry-framework/src/test/resources/serialization/ver3.snapshot b/nifi-registry-core/nifi-registry-framework/src/test/resources/serialization/ver3.snapshot
index 574fe56..3611014 100644
--- a/nifi-registry-core/nifi-registry-framework/src/test/resources/serialization/ver3.snapshot
+++ b/nifi-registry-core/nifi-registry-framework/src/test/resources/serialization/ver3.snapshot
@@ -1,6 +1,28 @@
 {
-  "header": {
-    "dataModelVersion": 3
-  },
-  "content": {}
+    "header" : {
+        "dataModelVersion" : "3"
+    },
+    "content" : {
+        "flowSnapshot" : {
+            "flowContents" : {
+                "componentType" : "PROCESS_GROUP",
+                "connections" : [ ],
+                "controllerServices" : [ ],
+                "funnels" : [ ],
+                "identifier" : "pg1",
+                "inputPorts" : [ ],
+                "labels" : [ ],
+                "name" : "My Process Group",
+                "outputPorts" : [ ],
+                "processGroups" : [ ],
+                "processors" : [ {
+                    "componentType" : "PROCESSOR",
+                    "identifier" : "processor1",
+                    "name" : "My Processor 1"
+                } ],
+                "remoteProcessGroups" : [ ],
+                "variables" : { }
+            }
+        }
+    }
 }
\ No newline at end of file
diff --git a/nifi-registry-core/nifi-registry-framework/src/test/resources/serialization/ver9999.snapshot b/nifi-registry-core/nifi-registry-framework/src/test/resources/serialization/ver9999.snapshot
new file mode 100644
index 0000000..743401b
--- /dev/null
+++ b/nifi-registry-core/nifi-registry-framework/src/test/resources/serialization/ver9999.snapshot
@@ -0,0 +1,6 @@
+{
+    "header": {
+        "dataModelVersion": 9999
+    },
+    "content": {}
+}
\ No newline at end of file
diff --git a/nifi-registry-core/nifi-registry-web-api/src/test/java/org/apache/nifi/registry/web/api/UnsecuredNiFiRegistryClientIT.java b/nifi-registry-core/nifi-registry-web-api/src/test/java/org/apache/nifi/registry/web/api/UnsecuredNiFiRegistryClientIT.java
index e16fa79..d5e0bc7 100644
--- a/nifi-registry-core/nifi-registry-web-api/src/test/java/org/apache/nifi/registry/web/api/UnsecuredNiFiRegistryClientIT.java
+++ b/nifi-registry-core/nifi-registry-web-api/src/test/java/org/apache/nifi/registry/web/api/UnsecuredNiFiRegistryClientIT.java
@@ -60,9 +60,12 @@
 import org.apache.nifi.registry.extension.repo.ExtensionRepoVersion;
 import org.apache.nifi.registry.extension.repo.ExtensionRepoVersionSummary;
 import org.apache.nifi.registry.field.Fields;
+import org.apache.nifi.registry.flow.ExternalControllerServiceReference;
 import org.apache.nifi.registry.flow.VersionedFlow;
 import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
 import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata;
+import org.apache.nifi.registry.flow.VersionedParameter;
+import org.apache.nifi.registry.flow.VersionedParameterContext;
 import org.apache.nifi.registry.flow.VersionedProcessGroup;
 import org.apache.nifi.registry.flow.VersionedProcessor;
 import org.apache.nifi.registry.flow.VersionedPropertyDescriptor;
@@ -86,12 +89,15 @@
 import java.nio.charset.StandardCharsets;
 import java.security.NoSuchAlgorithmException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.UUID;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -270,6 +276,9 @@
         final VersionedFlow snapshotFlow = flow1;
 
         final VersionedFlowSnapshot snapshot1 = createSnapshot(snapshotClient, snapshotFlow, 1);
+        assertNotNull(snapshot1);
+        assertNotNull(snapshot1.getSnapshotMetadata());
+        assertEquals(snapshotFlow.getIdentifier(), snapshot1.getSnapshotMetadata().getFlowIdentifier());
         LOGGER.info("Created snapshot # 1 with version " + snapshot1.getSnapshotMetadata().getVersion());
 
         final VersionedFlowSnapshot snapshot2 = createSnapshot(snapshotClient, snapshotFlow, 2);
@@ -805,6 +814,78 @@
 
     }
 
+    @Test
+    public void testFlowSnapshotsWithParameterContextAndEncodingVersion() throws IOException, NiFiRegistryException {
+        // Create a bucket
+        final Bucket bucket = new Bucket();
+        bucket.setName("Test Bucket");
+
+        final Bucket createdBucket = client.getBucketClient().create(bucket);
+        assertNotNull(createdBucket);
+
+        // Create the flow
+        final VersionedFlow flow = new VersionedFlow();
+        flow.setName("My Flow");
+        flow.setBucketIdentifier(createdBucket.getIdentifier());
+
+        final VersionedFlow createdFlow = client.getFlowClient().create(flow);
+        assertNotNull(createdFlow);
+
+        // Create a param context
+        final VersionedParameter param1 = new VersionedParameter();
+        param1.setName("Param 1");
+        param1.setValue("Param 1 Value");
+        param1.setDescription("Description");
+
+        final VersionedParameter param2 = new VersionedParameter();
+        param2.setName("Param 2");
+        param2.setValue("Param 2 Value");
+        param2.setDescription("Description");
+        param2.setSensitive(true);
+
+        final VersionedParameterContext context1 = new VersionedParameterContext();
+        context1.setName("Parameter Context 1");
+        context1.setParameters(new HashSet<>(Arrays.asList(param1, param2)));
+
+        final Map<String,VersionedParameterContext> contexts = new HashMap<>();
+        contexts.put(context1.getName(), context1);
+
+        // Create an external controller service reference
+        final ExternalControllerServiceReference serviceReference = new ExternalControllerServiceReference();
+        serviceReference.setName("External Service 1");
+        serviceReference.setIdentifier(UUID.randomUUID().toString());
+
+        final Map<String,ExternalControllerServiceReference> serviceReferences = new HashMap<>();
+        serviceReferences.put(serviceReference.getIdentifier(), serviceReference);
+
+        // Create the snapshot
+        final VersionedFlowSnapshot snapshot = buildSnapshot(createdFlow, 1);
+        snapshot.setFlowEncodingVersion("2.0.0");
+        snapshot.setParameterContexts(contexts);
+        snapshot.setExternalControllerServices(serviceReferences);
+
+        final VersionedFlowSnapshot createdSnapshot = client.getFlowSnapshotClient().create(snapshot);
+        assertNotNull(createdSnapshot);
+        assertNotNull(createdSnapshot.getFlowEncodingVersion());
+        assertNotNull(createdSnapshot.getParameterContexts());
+        assertNotNull(createdSnapshot.getExternalControllerServices());
+        assertEquals(snapshot.getFlowEncodingVersion(), createdSnapshot.getFlowEncodingVersion());
+        assertEquals(1, createdSnapshot.getParameterContexts().size());
+        assertEquals(1, createdSnapshot.getExternalControllerServices().size());
+
+        // Retrieve the snapshot
+        final VersionedFlowSnapshot retrievedSnapshot = client.getFlowSnapshotClient().get(
+                createdSnapshot.getSnapshotMetadata().getFlowIdentifier(),
+                createdSnapshot.getSnapshotMetadata().getVersion());
+        assertNotNull(retrievedSnapshot);
+        assertNotNull(retrievedSnapshot.getFlowEncodingVersion());
+        assertNotNull(retrievedSnapshot.getParameterContexts());
+        assertNotNull(retrievedSnapshot.getExternalControllerServices());
+        assertEquals(snapshot.getFlowEncodingVersion(), retrievedSnapshot.getFlowEncodingVersion());
+        assertEquals(1, retrievedSnapshot.getParameterContexts().size());
+        assertEquals(1, retrievedSnapshot.getExternalControllerServices().size());
+    }
+
     private void checkExtensionMetadata(Collection<ExtensionMetadata> extensions) {
         extensions.forEach(e -> {
             assertNotNull(e.getBundleInfo());