QPID-4999 : [Java Broker] Strip selector arguments from persistent bindings to non-topic exchanges created by buggy old clients

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

10 files changed, 334 insertions, 47 deletions

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 2b9c5ad290..647e19d659 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
@@ -531,6 +531,10 @@ public class MemoryConfigurationEntryStore implements ConfigurationEntryStore
                 if (fieldValues != null)
                 {
                     Object[] array = fieldValues.toArray(new Object[fieldValues.size()]);
+                    if (attributes == null)
+                    {
+                        attributes = new HashMap<String, Object>();
+                    }
                     attributes.put(fieldName, array);
                 }
             }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FilterSupport.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FilterSupport.java
index e0d1ac8695..81a89c9b4b 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FilterSupport.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FilterSupport.java
@@ -91,12 +91,20 @@ public class FilterSupport
     }
 
 
-    static boolean argumentsContainFilter(final Map<String, Object> args)
+    public static boolean argumentsContainFilter(final Map<String, Object> args)
     {
         return argumentsContainNoLocal(args) || argumentsContainJMSSelector(args);
     }
 
 
+    public static void removeFilters(final Map<String, Object> args)
+    {
+        args.remove(AMQPFilterTypes.JMS_SELECTOR.toString());
+        args.remove(AMQPFilterTypes.NO_LOCAL.toString());
+    }
+
+
+
     static boolean argumentsContainNoLocal(final Map<String, Object> args)
     {
         return args != null
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/model/VirtualHost.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/model/VirtualHost.java
index 08e9c5ca5c..9e32d303fb 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/model/VirtualHost.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/model/VirtualHost.java
@@ -119,6 +119,7 @@ public interface VirtualHost extends ConfiguredObject
                             QUEUE_ALERT_THRESHOLD_QUEUE_DEPTH_BYTES,
                             QUEUE_ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES,
                             CONFIG_PATH));
+    int CURRENT_CONFIG_VERSION = 1;
 
     //children
     Collection<VirtualHostAlias> getAliases();
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/AbstractJDBCMessageStore.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/AbstractJDBCMessageStore.java
index 905c83f7ed..e97c0d662d 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/AbstractJDBCMessageStore.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/AbstractJDBCMessageStore.java
@@ -55,6 +55,7 @@ import org.codehaus.jackson.map.ObjectMapper;
 abstract public class AbstractJDBCMessageStore implements MessageStore, DurableConfigurationStore
 {
     private static final String DB_VERSION_TABLE_NAME = "QPID_DB_VERSION";
+    private static final String CONFIGURATION_VERSION_TABLE_NAME = "QPID_CONFIG_VERSION";
 
     private static final String QUEUE_ENTRY_TABLE_NAME = "QPID_QUEUE_ENTRIES";
 
@@ -68,9 +69,10 @@ abstract public class AbstractJDBCMessageStore implements MessageStore, DurableC
     private static final String XID_ACTIONS_TABLE_NAME = "QPID_XID_ACTIONS";
 
     private static final String CONFIGURED_OBJECTS_TABLE_NAME = "QPID_CONFIGURED_OBJECTS";
+    private static final int DEFAULT_CONFIG_VERSION = 0;
 
     public static String[] ALL_TABLES = new String[] { DB_VERSION_TABLE_NAME, LINKS_TABLE_NAME, BRIDGES_TABLE_NAME, XID_ACTIONS_TABLE_NAME,
-        XID_TABLE_NAME, QUEUE_ENTRY_TABLE_NAME, MESSAGE_CONTENT_TABLE_NAME, META_DATA_TABLE_NAME, CONFIGURED_OBJECTS_TABLE_NAME };
+        XID_TABLE_NAME, QUEUE_ENTRY_TABLE_NAME, MESSAGE_CONTENT_TABLE_NAME, META_DATA_TABLE_NAME, CONFIGURED_OBJECTS_TABLE_NAME, CONFIGURATION_VERSION_TABLE_NAME };
 
     private static final int DB_VERSION = 6;
 
@@ -80,6 +82,12 @@ abstract public class AbstractJDBCMessageStore implements MessageStore, DurableC
     private static final String CREATE_DB_VERSION_TABLE = "CREATE TABLE "+ DB_VERSION_TABLE_NAME + " ( version int not null )";
     private static final String INSERT_INTO_DB_VERSION = "INSERT INTO "+ DB_VERSION_TABLE_NAME + " ( version ) VALUES ( ? )";
 
+    private static final String CREATE_CONFIG_VERSION_TABLE = "CREATE TABLE "+ CONFIGURATION_VERSION_TABLE_NAME + " ( version int not null )";
+    private static final String INSERT_INTO_CONFIG_VERSION = "INSERT INTO "+ CONFIGURATION_VERSION_TABLE_NAME + " ( version ) VALUES ( ? )";
+    private static final String SELECT_FROM_CONFIG_VERSION = "SELECT version FROM " + CONFIGURATION_VERSION_TABLE_NAME;
+    private static final String UPDATE_CONFIG_VERSION = "UPDATE " + CONFIGURATION_VERSION_TABLE_NAME + " SET version = ?";
+
+
     private static final String INSERT_INTO_QUEUE_ENTRY = "INSERT INTO " + QUEUE_ENTRY_TABLE_NAME + " (queue_id, message_id) values (?,?)";
     private static final String DELETE_FROM_QUEUE_ENTRY = "DELETE FROM " + QUEUE_ENTRY_TABLE_NAME + " WHERE queue_id = ? AND message_id =?";
     private static final String SELECT_FROM_QUEUE_ENTRY = "SELECT queue_id, message_id FROM " + QUEUE_ENTRY_TABLE_NAME + " ORDER BY queue_id, message_id";
@@ -223,6 +231,7 @@ abstract public class AbstractJDBCMessageStore implements MessageStore, DurableC
         Connection conn = newAutoCommitConnection();
 
         createVersionTable(conn);
+        createConfigVersionTable(conn);
         createConfiguredObjectsTable(conn);
         createQueueEntryTable(conn);
         createMetaDataTable(conn);
@@ -259,7 +268,33 @@ abstract public class AbstractJDBCMessageStore implements MessageStore, DurableC
                 pstmt.close();
             }
         }
+    }
 
+    private void createConfigVersionTable(final Connection conn) throws SQLException
+    {
+        if(!tableExists(CONFIGURATION_VERSION_TABLE_NAME, conn))
+        {
+            Statement stmt = conn.createStatement();
+            try
+            {
+                stmt.execute(CREATE_CONFIG_VERSION_TABLE);
+            }
+            finally
+            {
+                stmt.close();
+            }
+
+            PreparedStatement pstmt = conn.prepareStatement(INSERT_INTO_CONFIG_VERSION);
+            try
+            {
+                pstmt.setInt(1, DEFAULT_CONFIG_VERSION);
+                pstmt.execute();
+            }
+            finally
+            {
+                pstmt.close();
+            }
+        }
     }
 
     private void createConfiguredObjectsTable(final Connection conn) throws SQLException
@@ -279,6 +314,8 @@ abstract public class AbstractJDBCMessageStore implements MessageStore, DurableC
         }
     }
 
+
+
     private void createQueueEntryTable(final Connection conn) throws SQLException
     {
         if(!tableExists(QUEUE_ENTRY_TABLE_NAME, conn))
@@ -457,10 +494,10 @@ abstract public class AbstractJDBCMessageStore implements MessageStore, DurableC
     {
         try
         {
-            recoveryHandler.beginConfigurationRecovery(this);
+            recoveryHandler.beginConfigurationRecovery(this, getConfigVersion());
             loadConfiguredObjects(recoveryHandler);
 
-            recoveryHandler.completeConfigurationRecovery();
+            setConfigVersion(recoveryHandler.completeConfigurationRecovery());
         }
         catch (SQLException e)
         {
@@ -468,6 +505,67 @@ abstract public class AbstractJDBCMessageStore implements MessageStore, DurableC
         }
     }
 
+    private void setConfigVersion(int version) throws SQLException
+    {
+        Connection conn = newAutoCommitConnection();
+        try
+        {
+
+            PreparedStatement stmt = conn.prepareStatement(UPDATE_CONFIG_VERSION);
+            try
+            {
+                stmt.setInt(1, version);
+                stmt.execute();
+
+            }
+            finally
+            {
+                stmt.close();
+            }
+        }
+        finally
+        {
+            conn.close();
+        }
+    }
+
+    private int getConfigVersion() throws SQLException
+    {
+        Connection conn = newAutoCommitConnection();
+        try
+        {
+
+            Statement stmt = conn.createStatement();
+            try
+            {
+                ResultSet rs = stmt.executeQuery(SELECT_FROM_CONFIG_VERSION);
+                try
+                {
+
+                    if(rs.next())
+                    {
+                        return rs.getInt(1);
+                    }
+                    return DEFAULT_CONFIG_VERSION;
+                }
+                finally
+                {
+                    rs.close();
+                }
+
+            }
+            finally
+            {
+                stmt.close();
+            }
+        }
+        finally
+        {
+            conn.close();
+        }
+
+    }
+
     @Override
     public void close() throws Exception
     {
@@ -1837,52 +1935,89 @@ abstract public class AbstractJDBCMessageStore implements MessageStore, DurableC
                 Connection conn = newAutoCommitConnection();
                 try
                 {
-                    PreparedStatement stmt = conn.prepareStatement(FIND_CONFIGURED_OBJECT);
-                    try
+                    updateConfiguredObject(configuredObject, conn);
+                }
+                finally
+                {
+                    conn.close();
+                }
+            }
+            catch (SQLException e)
+            {
+                throw new AMQStoreException("Error updating configured object " + configuredObject + " in database: " + e.getMessage(), e);
+            }
+        }
+    }
+
+    @Override
+    public void update(ConfiguredObjectRecord... records) throws AMQStoreException
+    {
+        if (_stateManager.isInState(State.ACTIVE) || _stateManager.isInState(State.ACTIVATING))
+        {
+            try
+            {
+                Connection conn = newConnection();
+                try
+                {
+                    for(ConfiguredObjectRecord record : records)
                     {
-                        stmt.setString(1, configuredObject.getId().toString());
-                        ResultSet rs = stmt.executeQuery();
+                        updateConfiguredObject(record, conn);
+                    }
+                    conn.commit();
+                }
+                finally
+                {
+                    conn.close();
+                }
+            }
+            catch (SQLException e)
+            {
+                throw new AMQStoreException("Error updating configured objects in database: " + e.getMessage(), e);
+            }
+
+        }
+
+    }
+
+    private void updateConfiguredObject(ConfiguredObjectRecord configuredObject, Connection conn)
+            throws SQLException, AMQStoreException
+    {
+            PreparedStatement stmt = conn.prepareStatement(FIND_CONFIGURED_OBJECT);
+            try
+            {
+                stmt.setString(1, configuredObject.getId().toString());
+                ResultSet rs = stmt.executeQuery();
+                try
+                {
+                    if (rs.next())
+                    {
+                        PreparedStatement stmt2 = conn.prepareStatement(UPDATE_CONFIGURED_OBJECTS);
                         try
                         {
-                            if (rs.next())
+                            stmt2.setString(1, configuredObject.getType());
+                            if (configuredObject.getAttributes() != null)
                             {
-                                PreparedStatement stmt2 = conn.prepareStatement(UPDATE_CONFIGURED_OBJECTS);
-                                try
-                                {
-                                    stmt2.setString(1, configuredObject.getType());
-                                    if (configuredObject.getAttributes() != null)
-                                    {
-                                        byte[] attributesAsBytes = (new ObjectMapper()).writeValueAsBytes(
-                                                configuredObject.getAttributes());
-                                        ByteArrayInputStream bis = new ByteArrayInputStream(attributesAsBytes);
-                                        stmt2.setBinaryStream(2, bis, attributesAsBytes.length);
-                                    }
-                                    else
-                                    {
-                                        stmt2.setNull(2, Types.BLOB);
-                                    }
-                                    stmt2.setString(3, configuredObject.getId().toString());
-                                    stmt2.execute();
-                                }
-                                finally
-                                {
-                                    stmt2.close();
-                                }
+                                byte[] attributesAsBytes = (new ObjectMapper()).writeValueAsBytes(
+                                        configuredObject.getAttributes());
+                                ByteArrayInputStream bis = new ByteArrayInputStream(attributesAsBytes);
+                                stmt2.setBinaryStream(2, bis, attributesAsBytes.length);
+                            }
+                            else
+                            {
+                                stmt2.setNull(2, Types.BLOB);
                             }
+                            stmt2.setString(3, configuredObject.getId().toString());
+                            stmt2.execute();
                         }
                         finally
                         {
-                            rs.close();
+                            stmt2.close();
                         }
                     }
-                    finally
-                    {
-                        stmt.close();
-                    }
                 }
                 finally
                 {
-                    conn.close();
+                    rs.close();
                 }
             }
             catch (JsonMappingException e)
@@ -1897,11 +2032,11 @@ abstract public class AbstractJDBCMessageStore implements MessageStore, DurableC
             {
                 throw new AMQStoreException("Error updating configured object " + configuredObject + " in database: " + e.getMessage(), e);
             }
-            catch (SQLException e)
+            finally
             {
-                throw new AMQStoreException("Error updating configured object " + configuredObject + " in database: " + e.getMessage(), e);
+                stmt.close();
             }
-        }
+
     }
 
     private ConfiguredObjectRecord loadConfiguredObject(final UUID id) throws AMQStoreException
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/ConfigurationRecoveryHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/ConfigurationRecoveryHandler.java
index 2a4aed5373..a3534d3fa5 100755
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/ConfigurationRecoveryHandler.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/ConfigurationRecoveryHandler.java
@@ -28,10 +28,14 @@ import java.util.UUID;
 
 public interface ConfigurationRecoveryHandler
 {
-    void beginConfigurationRecovery(DurableConfigurationStore store);
+    void beginConfigurationRecovery(DurableConfigurationStore store, int configVersion);
 
     void configuredObject(UUID id, String type, Map<String, Object> attributes);
 
-    void completeConfigurationRecovery();
+    /**
+     *
+     * @return the model version of the configuration
+     */
+    int completeConfigurationRecovery();
 
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/ConfiguredObjectRecord.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/ConfiguredObjectRecord.java
index 5c8f452dc7..44490385d9 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/ConfiguredObjectRecord.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/ConfiguredObjectRecord.java
@@ -60,4 +60,29 @@ public class ConfiguredObjectRecord
         return "ConfiguredObjectRecord [id=" + _id + ", type=" + _type + ", attributes=" + _attributes + "]";
     }
 
+    @Override
+    public boolean equals(Object o)
+    {
+        if(this == o)
+        {
+            return true;
+        }
+        if(o == null || getClass() != o.getClass())
+        {
+            return false;
+        }
+
+        ConfiguredObjectRecord that = (ConfiguredObjectRecord) o;
+
+        return _type.equals(that._type) && _id.equals(that._id) && _attributes.equals(that._attributes);
+    }
+
+    @Override
+    public int hashCode()
+    {
+        int result = _id.hashCode();
+        result = 31 * result + _type.hashCode();
+        result = 31 * result + _attributes.hashCode();
+        return result;
+    }
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/DurableConfigurationStore.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/DurableConfigurationStore.java
index fdde21ba89..7ce761af18 100755
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/DurableConfigurationStore.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/DurableConfigurationStore.java
@@ -91,4 +91,7 @@ public interface DurableConfigurationStore
     void update(UUID id, String type, Map<String, Object> attributes) throws AMQStoreException;
 
 
+    public void update(ConfiguredObjectRecord... records) throws AMQStoreException;
+
+
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MessageStore.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MessageStore.java
index 5fc6bad368..c605e1b599 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MessageStore.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MessageStore.java
@@ -20,8 +20,6 @@
  */
 package org.apache.qpid.server.store;
 
-import org.apache.commons.configuration.Configuration;
-
 /**
  * MessageStore defines the interface to a storage area, which can be used to preserve the state of messages.
  *
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/NullMessageStore.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/NullMessageStore.java
index 77cde80af9..078a0d3752 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/NullMessageStore.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/NullMessageStore.java
@@ -21,6 +21,7 @@ package org.apache.qpid.server.store;
 
 import java.util.Map;
 import java.util.UUID;
+import org.apache.qpid.AMQStoreException;
 import org.apache.qpid.server.model.VirtualHost;
 
 public abstract class NullMessageStore implements MessageStore, DurableConfigurationStore
@@ -38,6 +39,11 @@ public abstract class NullMessageStore implements MessageStore, DurableConfigura
     }
 
     @Override
+    public void update(ConfiguredObjectRecord... records) throws AMQStoreException
+    {
+    }
+
+    @Override
     public void remove(UUID id, String type)
     {
     }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java
index f364d93d98..605bbe5f45 100755
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java
@@ -24,6 +24,7 @@ import java.io.DataInputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.TreeMap;
 import java.util.UUID;
@@ -35,6 +36,8 @@ import org.apache.qpid.framing.FieldTable;
 import org.apache.qpid.server.exchange.Exchange;
 import org.apache.qpid.server.exchange.ExchangeFactory;
 import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.exchange.FilterSupport;
+import org.apache.qpid.server.exchange.TopicExchange;
 import org.apache.qpid.server.logging.actors.CurrentActor;
 import org.apache.qpid.server.logging.messages.ConfigStoreMessages;
 import org.apache.qpid.server.logging.messages.TransactionLogMessages;
@@ -49,6 +52,7 @@ import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.queue.AMQQueueFactory;
 import org.apache.qpid.server.queue.QueueEntry;
 import org.apache.qpid.server.store.ConfigurationRecoveryHandler;
+import org.apache.qpid.server.store.ConfiguredObjectRecord;
 import org.apache.qpid.server.store.DurableConfigurationStore;
 import org.apache.qpid.server.store.MessageStore;
 import org.apache.qpid.server.store.MessageStoreRecoveryHandler;
@@ -63,6 +67,8 @@ import org.apache.qpid.transport.Xid;
 import org.apache.qpid.transport.util.Functions;
 import org.apache.qpid.util.ByteBufferInputStream;
 
+import static org.apache.qpid.server.model.VirtualHost.CURRENT_CONFIG_VERSION;
+
 public class VirtualHostConfigRecoveryHandler implements ConfigurationRecoveryHandler,
                                                         MessageStoreRecoveryHandler,
                                                         MessageStoreRecoveryHandler.StoredMessageRecoveryHandler,
@@ -85,6 +91,8 @@ public class VirtualHostConfigRecoveryHandler implements ConfigurationRecoveryHa
 
     private MessageStoreLogSubject _logSubject;
     private MessageStore _store;
+    private int _currentConfigVersion;
+    private DurableConfigurationStore _configStore;
 
     public VirtualHostConfigRecoveryHandler(VirtualHost virtualHost,
                                             ExchangeRegistry exchangeRegistry,
@@ -96,10 +104,11 @@ public class VirtualHostConfigRecoveryHandler implements ConfigurationRecoveryHa
     }
 
     @Override
-    public void beginConfigurationRecovery(DurableConfigurationStore store)
+    public void beginConfigurationRecovery(DurableConfigurationStore store, int configVersion)
     {
         _logSubject = new MessageStoreLogSubject(_virtualHost,store.getClass().getSimpleName());
-
+        _configStore = store;
+        _currentConfigVersion = configVersion;
         CurrentActor.get().message(_logSubject, ConfigStoreMessages.RECOVERY_START());
     }
 
@@ -482,8 +491,20 @@ public class VirtualHostConfigRecoveryHandler implements ConfigurationRecoveryHa
     }
 
     @Override
-    public void completeConfigurationRecovery()
+    public int completeConfigurationRecovery()
     {
+        if(CURRENT_CONFIG_VERSION !=_currentConfigVersion)
+        {
+            try
+            {
+                upgrade();
+            }
+            catch (AMQStoreException e)
+            {
+                throw new IllegalArgumentException("Unable to upgrade configuration from version " + _currentConfigVersion + " to version " + CURRENT_CONFIG_VERSION);
+            }
+        }
+
         Map<UUID, Map<String, Object>> exchangeObjects =
                 _configuredObjects.remove(org.apache.qpid.server.model.Exchange.class.getName());
 
@@ -511,6 +532,88 @@ public class VirtualHostConfigRecoveryHandler implements ConfigurationRecoveryHa
 
 
         CurrentActor.get().message(_logSubject, ConfigStoreMessages.RECOVERY_COMPLETE());
+
+        return CURRENT_CONFIG_VERSION;
+    }
+
+    private void upgrade() throws AMQStoreException
+    {
+
+        Map<UUID, String> updates = new HashMap<UUID, String>();
+
+        final String bindingType = Binding.class.getName();
+
+        switch(_currentConfigVersion)
+        {
+            case 0:
+                Map<UUID, Map<String, Object>> bindingObjects =
+                                    _configuredObjects.get(bindingType);
+                if(bindingObjects != null)
+                {
+                    for(Map.Entry<UUID, Map<String,Object>> bindingEntry : bindingObjects.entrySet())
+                    {
+                        Map<String, Object> binding = bindingEntry.getValue();
+
+                        if(hasSelectorArguments(binding) && !isTopicExchange(binding))
+                        {
+                            binding = new LinkedHashMap<String, Object>(binding);
+                            removeSelectorArguments(binding);
+                            bindingEntry.setValue(binding);
+
+                            updates.put(bindingEntry.getKey(), bindingType);
+                        }
+                    }
+                }
+            case CURRENT_CONFIG_VERSION:
+                if(!updates.isEmpty())
+                {
+                    ConfiguredObjectRecord[] updateRecords = new ConfiguredObjectRecord[updates.size()];
+                    int i = 0;
+                    for(Map.Entry<UUID, String> update : updates.entrySet())
+                    {
+                        updateRecords[i++] = new ConfiguredObjectRecord(update.getKey(), update.getValue(), _configuredObjects.get(update.getValue()).get(update.getKey()));
+                    }
+                    _configStore.update(updateRecords);
+                }
+                break;
+            default:
+                throw new IllegalStateException("Unknown configuration model version: " + _currentConfigVersion + ". Are you attempting to run an older instance against an upgraded configuration?");
+        }
+    }
+
+    private void removeSelectorArguments(Map<String, Object> binding)
+    {
+        @SuppressWarnings("unchecked")
+        Map<String, Object> arguments = new LinkedHashMap<String, Object>((Map<String,Object>)binding.get(Binding.ARGUMENTS));
+
+        FilterSupport.removeFilters(arguments);
+        binding.put(Binding.ARGUMENTS, arguments);
+    }
+
+    private boolean isTopicExchange(Map<String, Object> binding)
+    {
+        UUID exchangeId = UUID.fromString((String)binding.get(Binding.EXCHANGE));
+        final
+        Map<UUID, Map<String, Object>> exchanges =
+                _configuredObjects.get(org.apache.qpid.server.model.Exchange.class.getName());
+
+        if(exchanges != null && exchanges.containsKey(exchangeId))
+        {
+            return "topic".equals(exchanges.get(exchangeId).get(org.apache.qpid.server.model.Exchange.TYPE));
+        }
+        else
+        {
+            return _exchangeRegistry.getExchange(exchangeId) != null
+                   && _exchangeRegistry.getExchange(exchangeId).getType() == TopicExchange.TYPE;
+        }
+
+    }
+
+    private boolean hasSelectorArguments(Map<String, Object> binding)
+    {
+        @SuppressWarnings("unchecked")
+        Map<String, Object> arguments = (Map<String, Object>) binding.get(Binding.ARGUMENTS);
+        return (arguments != null) && FilterSupport.argumentsContainFilter(arguments);
     }
 
     private void recoverExchanges(Map<UUID, Map<String, Object>> exchangeObjects)