QPID-4803: Ensure the modelVersion and storeVersion attributes are saved to the configuration store and validated at startup

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1478732 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 79 insertions, 16 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/startup/BrokerRecoverer.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/startup/BrokerRecoverer.java
index 35c96bc993..4b7b9e3254 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/startup/BrokerRecoverer.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/startup/BrokerRecoverer.java
@@ -25,6 +25,7 @@ import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.regex.Pattern;
 
 import org.apache.qpid.server.BrokerOptions;
 import org.apache.qpid.server.configuration.ConfigurationEntry;
@@ -39,6 +40,7 @@ import org.apache.qpid.server.model.AuthenticationProvider;
 import org.apache.qpid.server.model.Broker;
 import org.apache.qpid.server.model.ConfiguredObject;
 import org.apache.qpid.server.model.KeyStore;
+import org.apache.qpid.server.model.Model;
 import org.apache.qpid.server.model.TrustStore;
 import org.apache.qpid.server.model.adapter.AccessControlProviderFactory;
 import org.apache.qpid.server.model.adapter.AuthenticationProviderFactory;
@@ -46,10 +48,13 @@ import org.apache.qpid.server.model.adapter.BrokerAdapter;
 import org.apache.qpid.server.model.adapter.GroupProviderFactory;
 import org.apache.qpid.server.model.adapter.PortFactory;
 import org.apache.qpid.server.stats.StatisticsGatherer;
+import org.apache.qpid.server.util.MapValueConverter;
 import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
 
 public class BrokerRecoverer implements ConfiguredObjectRecoverer<Broker>
 {
+    private static final Pattern MODEL_VERSION_PATTERN = Pattern.compile("^\\d+\\.\\d+$");
+
     private final StatisticsGatherer _statisticsGatherer;
     private final VirtualHostRegistry _virtualHostRegistry;
     private final LogRecorder _logRecorder;
@@ -80,8 +85,14 @@ public class BrokerRecoverer implements ConfiguredObjectRecoverer<Broker>
     @Override
     public Broker create(RecovererProvider recovererProvider, ConfigurationEntry entry, ConfiguredObject... parents)
     {
+        Map<String, Object> attributes = entry.getAttributes();
+        validateAttributes(attributes);
+
+        Map<String, Object> attributesCopy = new HashMap<String, Object>(attributes);
+        attributesCopy.put(Broker.MODEL_VERSION, Model.MODEL_VERSION);
+
         StoreConfigurationChangeListener storeChangeListener = new StoreConfigurationChangeListener(entry.getStore());
-        BrokerAdapter broker = new BrokerAdapter(entry.getId(), entry.getAttributes(), _statisticsGatherer, _virtualHostRegistry,
+        BrokerAdapter broker = new BrokerAdapter(entry.getId(), attributesCopy, _statisticsGatherer, _virtualHostRegistry,
                 _logRecorder, _rootMessageLogger, _authenticationProviderFactory, _groupProviderFactory, _accessControlProviderFactory,
                 _portFactory, _taskExecutor, entry.getStore(), _brokerOptions);
 
@@ -117,6 +128,37 @@ public class BrokerRecoverer implements ConfiguredObjectRecoverer<Broker>
         return broker;
     }
 
+    private void validateAttributes(Map<String, Object> attributes)
+    {
+        String modelVersion = null;
+        if (attributes.containsKey(Broker.MODEL_VERSION))
+        {
+            modelVersion = MapValueConverter.getStringAttribute(Broker.MODEL_VERSION, attributes, null);
+        }
+
+        if (modelVersion == null)
+        {
+            throw new IllegalConfigurationException("Broker " + Broker.MODEL_VERSION + " must be specified");
+        }
+
+        if (!MODEL_VERSION_PATTERN.matcher(modelVersion).matches())
+        {
+            throw new IllegalConfigurationException("Broker " + Broker.MODEL_VERSION + " is specified in incorrect format: "
+                    + modelVersion);
+        }
+
+        int versionSeparatorPosition = modelVersion.indexOf(".");
+        String majorVersionPart = modelVersion.substring(0, versionSeparatorPosition);
+        int majorModelVersion = Integer.parseInt(majorVersionPart);
+        int minorModelVersion = Integer.parseInt(modelVersion.substring(versionSeparatorPosition + 1));
+
+        if (majorModelVersion != Model.MODEL_MAJOR_VERSION || minorModelVersion > Model.MODEL_MINOR_VERSION)
+        {
+            throw new IllegalConfigurationException("The model version '" + modelVersion
+                    + "' in configuration is incompatible with the broker model version '" + Model.MODEL_VERSION + "'");
+        }
+    }
+
     private void recoverType(RecovererProvider recovererProvider,
                              StoreConfigurationChangeListener storeChangeListener,
                              BrokerAdapter broker,
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/store/MemoryConfigurationEntryStore.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/store/MemoryConfigurationEntryStore.java
index 24e0e3bbff..2b9c5ad290 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/store/MemoryConfigurationEntryStore.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/store/MemoryConfigurationEntryStore.java
@@ -66,7 +66,7 @@ public class MemoryConfigurationEntryStore implements ConfigurationEntryStore
     private static final String ID = "id";
     private static final String TYPE = "@type";
 
-    private static final int STORE_VERSION = 1;
+    static final int STORE_VERSION = 1;
 
     private final ObjectMapper _objectMapper;
     private final Map<UUID, ConfigurationEntry> _entries;
@@ -268,6 +268,24 @@ public class MemoryConfigurationEntryStore implements ConfigurationEntryStore
         {
             is = url.openStream();
             JsonNode node = loadJsonNodes(is, _objectMapper);
+
+            int storeVersion = 0;
+            JsonNode storeVersionNode = node.get(Broker.STORE_VERSION);
+            if (storeVersionNode == null || storeVersionNode.isNull())
+            {
+                throw new IllegalConfigurationException("Broker " + Broker.STORE_VERSION + " attribute must be specified");
+            }
+            else
+            {
+                storeVersion = storeVersionNode.getIntValue();
+            }
+
+            if (storeVersion != STORE_VERSION)
+            {
+                throw new IllegalConfigurationException("The data of version " + storeVersion
+                        + " can not be loaded by store of version " + STORE_VERSION);
+            }
+
             ConfigurationEntry brokerEntry = toEntry(node, Broker.class, _entries);
             _rootId = brokerEntry.getId();
         }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/model/Model.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/model/Model.java
index bccb6b48ee..dab92c50fa 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/model/Model.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/model/Model.java
@@ -34,6 +34,7 @@ public class Model
      */
     public static final int MODEL_MAJOR_VERSION = 1;
     public static final int MODEL_MINOR_VERSION = 0;
+    public static final String MODEL_VERSION = MODEL_MAJOR_VERSION + "." + MODEL_MINOR_VERSION;
 
     private static final Model MODEL_INSTANCE = new Model();
 
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/BrokerAdapter.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/BrokerAdapter.java
index 3a94cf22f2..adc30eb944 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/BrokerAdapter.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/BrokerAdapter.java
@@ -100,6 +100,8 @@ public class BrokerAdapter extends AbstractAdapter implements Broker, Configurat
         put(VIRTUALHOST_STORE_TRANSACTION_IDLE_TIMEOUT_WARN, Long.class);
         put(VIRTUALHOST_STORE_TRANSACTION_OPEN_TIMEOUT_CLOSE, Long.class);
         put(VIRTUALHOST_STORE_TRANSACTION_OPEN_TIMEOUT_WARN, Long.class);
+        put(MODEL_VERSION, String.class);
+        put(STORE_VERSION, String.class);
     }});
 
     public static final int DEFAULT_STATISTICS_REPORTING_PERIOD = 0;
@@ -775,7 +777,7 @@ public class BrokerAdapter extends AbstractAdapter implements Broker, Configurat
         }
         else if (MODEL_VERSION.equals(name))
         {
-            return Model.MODEL_MAJOR_VERSION + "." + Model.MODEL_MINOR_VERSION;
+            return Model.MODEL_VERSION;
         }
         else if (STORE_VERSION.equals(name))
         {
@@ -1132,23 +1134,22 @@ public class BrokerAdapter extends AbstractAdapter implements Broker, Configurat
         Map<String, Object> convertedAttributes = MapValueConverter.convert(attributes, ATTRIBUTE_TYPES);
         validateAttributes(convertedAttributes);
 
-        Collection<String> names = AVAILABLE_ATTRIBUTES;
-        for (String name : names)
-        {
-            if (convertedAttributes.containsKey(name))
-            {
-                Object desired = convertedAttributes.get(name);
-                Object expected = getAttribute(name);
-                if (changeAttribute(name, expected, desired))
-                {
-                    attributeSet(name, expected, desired);
-                }
-            }
-        }
+        super.changeAttributes(convertedAttributes);
     }
 
     private void validateAttributes(Map<String, Object> convertedAttributes)
     {
+        if (convertedAttributes.containsKey(MODEL_VERSION) && !Model.MODEL_VERSION.equals(convertedAttributes.get(MODEL_VERSION)))
+        {
+            throw new IllegalConfigurationException("Cannot change the model version");
+        }
+
+        if (convertedAttributes.containsKey(STORE_VERSION)
+                && !new Integer(_brokerStore.getVersion()).equals(convertedAttributes.get(STORE_VERSION)))
+        {
+            throw new IllegalConfigurationException("Cannot change the store version");
+        }
+
         String defaultVirtualHost = (String) convertedAttributes.get(DEFAULT_VIRTUAL_HOST);
         if (defaultVirtualHost != null)
         {
diff --git a/qpid/java/broker/src/main/resources/initial-config.json b/qpid/java/broker/src/main/resources/initial-config.json
index e510c34178..02fe942f55 100644
--- a/qpid/java/broker/src/main/resources/initial-config.json
+++ b/qpid/java/broker/src/main/resources/initial-config.json
@@ -21,6 +21,7 @@
 {
   "name": "Broker",
   "storeVersion": 1,
+  "modelVersion": "1.0",
   "defaultVirtualHost" : "default",
   "authenticationproviders" : [ {
     "name" : "passwordFile",