Repository: EnMasseProject/enmasse
Pull request #3193 (2019-09-16 09:22:58)

Remove infraType label to prevent accidental delete
Use ownerreferences instead to make sure address configmaps are deleted
when an address space configmap is deleted.

k-wall: Thinking about the upgrade between versions of enmasse, is it possible to retrospectively add the `NewOwnerReference` to existing address resources.
lulf: @k-wall Good point, that should be doable. We anyway need the uid of the 'parent', so some more work needed still
lulf: @enmasse-ci run tests profile=shared
enmasse-ci: ### :x: Test Summary :x:
**TEST_PROFILE**: shared
**TEST_CASE:** 
**TOTAL:** 233
**PASS:** 171
**FAIL:** 4
**SKIP:** 58
### :heavy_exclamation_mark: Test Failures :heavy_exclamation_mark:
- testFilterConnectionsByHostname in io.enmasse.systemtest.shared.brokered.web.FirefoxWebConsoleTest
- previousPersistentSessionDiscarded in io.enmasse.systemtest.shared.standard.mqtt.SessionTest
- testFilterConnectionsByHostname in io.enmasse.systemtest.shared.standard.web.FirefoxWebConsoleTest
- testSortConnectionsBySenders in io.enmasse.systemtest.shared.standard.web.FirefoxWebConsoleTest
k-wall: Changes in themselves look good, however as I mentioned in another channel, I think this work should include an upgrader to set the owner references on existing addresses.  Approaches that minimise difference/favour uniformity tend to have fewer bugs in the long run.

I know we are talking about AddressSpace/Address/CRD refactoring, but I don't think this is contemplated for 0.30. I'm in favour of a upgrader for this change.
lulf: @k-wall I have added conversion of configmaps to both standard and brokered address space. For brokered, the logic for replacing is a bit different in standard, but in brokered agent will always try to update configmaps at startup, so updating the ownerref while doing that seems to work.

I have also made it so that version is always set if address is created/replaced both via api server and console.

Patch:
diff --git a/address-space-controller/src/main/java/io/enmasse/controller/AddressSpaceController.java b/address-space-controller/src/main/java/io/enmasse/controller/AddressSpaceController.java
index 2873c884f..0d0b28ef1 100644
--- a/address-space-controller/src/main/java/io/enmasse/controller/AddressSpaceController.java
+++ b/address-space-controller/src/main/java/io/enmasse/controller/AddressSpaceController.java
@@ -65,7 +65,7 @@ public class AddressSpaceController {
         schemaApi.watchSchema(schemaProvider, options.getResyncInterval());
         Kubernetes kubernetes = new KubernetesHelper(controllerClient.getNamespace(), controllerClient, options.getTemplateDir(), isOpenShift);
 
-        AddressSpaceApi addressSpaceApi = new ConfigMapAddressSpaceApi(controllerClient);
+        AddressSpaceApi addressSpaceApi = new ConfigMapAddressSpaceApi(controllerClient, options.getVersion());
         EventLogger eventLogger = options.isEnableEventLogger() ? new KubeEventLogger(controllerClient, controllerClient.getNamespace(), Clock.systemUTC(), "address-space-controller")
                 : new LogEventLogger();
 
diff --git a/agent/bin/agent.js b/agent/bin/agent.js
index b459589a0..276b98c7f 100644
--- a/agent/bin/agent.js
+++ b/agent/bin/agent.js
@@ -29,6 +29,23 @@ function bind_event(source, event, target, method) {
     source.on(event, target[method || event].bind(target));
 }
 
+function create_ownerreference(namespace, address_space_name) {
+    // If not defined, that is ok. The address source handles the case where ownerreference is not set
+    if (namespace === undefined || address_space_name === undefined) {
+        return Promise.resolve(undefined)
+    }
+    return kubernetes.get('configmaps/' + namespace + '.' + address_space_name, {}).then(function (configmap) {
+        return {
+            apiVersion: "v1",
+            kind: "ConfigMap",
+            blockOwnerDeletion: true,
+            controller: true,
+            name: configmap.metadata.name,
+            uid: configmap.metadata.uid
+        }
+    });
+}
+
 function start(env) {
     kubernetes.is_openshift().then((openshift) => {
         kubernetes.get_messaging_route_hostname(env).then(function (result) {
@@ -69,7 +86,10 @@ function start(env) {
                     ragent.start_listening(env);
                     ragent.listen_health({HEALTH_PORT:8888});
                 }
-                address_source.start();
+
+                create_ownerreference(env.ADDRESS_SPACE_NAMESPACE, env.ADDRESS_SPACE).then(function (ownerReference) {
+                    address_source.start(ownerReference);
+                });
 
                 process.on('SIGTERM', function () {
                     log.info('Shutdown started');
diff --git a/agent/lib/internal_address_source.js b/agent/lib/internal_address_source.js
index 8256c305e..7ddb06034 100644
--- a/agent/lib/internal_address_source.js
+++ b/agent/lib/internal_address_source.js
@@ -146,8 +146,9 @@ function AddressSource(config) {
     events.EventEmitter.call(this);
 }
 
-AddressSource.prototype.start = function() {
+AddressSource.prototype.start = function(ownerReference) {
     var options = myutils.merge({selector: this.selector}, this.config);
+    this.ownerReference = ownerReference;
     this.watcher = kubernetes.watch('configmaps', options);
     this.watcher.on('updated', this.updated.bind(this));
     this.readiness = {};
@@ -215,13 +216,24 @@ AddressSource.prototype.updated = function (objects) {
     }
 };
 
+AddressSource.prototype.get_addressspace_configmap_name = function() {
+    return this.config.ADDRESS_SPACE_NAMESPACE + "." + this.config.ADDRESS_SPACE;
+}
+
 AddressSource.prototype.update_status = function (record, ready) {
+    var self = this;
     function update(configmap) {
         var def = JSON.parse(configmap.data['config.json']);
         if (def.status === undefined) {
             def.status = {};
         }
-        if (def.status.isReady !== ready) {
+        var updateOwnerRef = false;
+        if (configmap.metadata.ownerReferences === undefined && self.ownerReference !== undefined) {
+            def.metadata.annotations = myutils.merge(def.metadata.annotations, {"enmasse.io/version": self.config.VERSION});
+            configmap.metadata.ownerReferences = [self.ownerReference];
+            updateOwnerRef = true;
+        }
+        if (def.status.isReady !== ready || updateOwnerRef) {
             def.status.isReady = ready;
             def.status.phase = ready ? 'Active' : 'Pending';
             configmap.data['config.json'] = JSON.stringify(def);
diff --git a/api-model/src/main/java/io/enmasse/config/AnnotationKeys.java b/api-model/src/main/java/io/enmasse/config/AnnotationKeys.java
index b459f6f75..04af63ab4 100644
--- a/api-model/src/main/java/io/enmasse/config/AnnotationKeys.java
+++ b/api-model/src/main/java/io/enmasse/config/AnnotationKeys.java
@@ -31,4 +31,5 @@ public interface AnnotationKeys {
     String OPENSHIFT_SERVING_CERT_SECRET_NAME = "service.alpha.openshift.io/serving-cert-secret-name";
     String APPLIED_PLAN = "enmasse.io/applied-plan";
     String GENERATION = "enmasse.io/generation";
+    String VERSION = "enmasse.io/version";
 }
diff --git a/api-model/src/main/java/io/enmasse/model/validation/AddressSpaceConnectorAddressRulePatternValidator.java b/api-model/src/main/java/io/enmasse/model/validation/AddressSpaceConnectorAddressRulePatternValidator.java
index ab33d2258..2b875f7bc 100644
--- a/api-model/src/main/java/io/enmasse/model/validation/AddressSpaceConnectorAddressRulePatternValidator.java
+++ b/api-model/src/main/java/io/enmasse/model/validation/AddressSpaceConnectorAddressRulePatternValidator.java
@@ -13,7 +13,7 @@ import java.util.regex.Pattern;
 
 public class AddressSpaceConnectorAddressRulePatternValidator implements ConstraintValidator<AddressSpaceConnectorAddressRulePattern, AddressSpaceSpecConnectorAddressRule> {
 
-    private static final Pattern PATTERN = Pattern.compile("([^/]+(/?[^/]+)*)*");
+    private static final Pattern PATTERN = Pattern.compile("([^/#*]+|\\*|#)(/([^/#*]+|\\*|#))*");
 
     @Override
     public boolean isValid(AddressSpaceSpecConnectorAddressRule rule, ConstraintValidatorContext context) {
diff --git a/api-server/src/main/java/io/enmasse/api/server/ApiServer.java b/api-server/src/main/java/io/enmasse/api/server/ApiServer.java
index 54600c136..11aa9e046 100644
--- a/api-server/src/main/java/io/enmasse/api/server/ApiServer.java
+++ b/api-server/src/main/java/io/enmasse/api/server/ApiServer.java
@@ -65,7 +65,7 @@ public class ApiServer extends AbstractVerticle {
         CachingSchemaProvider schemaProvider = new CachingSchemaProvider();
         schemaApi.watchSchema(schemaProvider, options.getResyncInterval());
 
-        AddressSpaceApi addressSpaceApi = new ConfigMapAddressSpaceApi(client);
+        AddressSpaceApi addressSpaceApi = new ConfigMapAddressSpaceApi(client, options.getVersion());
 
         AuthApi authApi = new KubeAuthApi(client, client.getConfiguration().getOauthToken());
 
diff --git a/documentation/assemblies/assembly-managing-address-spaces.adoc b/documentation/assemblies/assembly-managing-address-spaces.adoc
index c11231276..8c4241162 100644
--- a/documentation/assemblies/assembly-managing-address-spaces.adoc
+++ b/documentation/assemblies/assembly-managing-address-spaces.adoc
@@ -31,6 +31,8 @@ include::../modules/ref-example-address-space-output.adoc[leveloffset=+1]
 
 include::../modules/ref-address-space-example-exporting-endpoints.adoc[leveloffset=+1]
 
+include::../modules/ref-address-space-example-connectors.adoc[leveloffset=+1]
+
 include::../modules/proc-create-address-space-cli.adoc[leveloffset=+1]
 
 include::../modules/proc-create-address-space-console.adoc[leveloffset=+1]
diff --git a/documentation/assemblies/assembly-managing-addresses.adoc b/documentation/assemblies/assembly-managing-addresses.adoc
index 2c830a61a..5eee31069 100644
--- a/documentation/assemblies/assembly-managing-addresses.adoc
+++ b/documentation/assemblies/assembly-managing-addresses.adoc
@@ -16,6 +16,8 @@ include::../modules/con-tenant-address-plans.adoc[leveloffset=+1]
 
 include::../modules/ref-address-example.adoc[leveloffset=+1]
 
+include::../modules/ref-address-example-forwarders.adoc[leveloffset=+2]
+
 include::../modules/proc-list-available-address-plans.adoc[leveloffset=+1]
 
 include::../modules/proc-create-address-cli.adoc[leveloffset=+1]
diff --git a/documentation/common/attributes.adoc b/documentation/common/attributes.adoc
index 5d38457cc..c9cc8e613 100644
--- a/documentation/common/attributes.adoc
+++ b/documentation/common/attributes.adoc
@@ -36,7 +36,7 @@
 :cpp: C++
 :BaseProductVersion: master
 :ConsoleName: {ProductName} Console
-
+:DataGridProductName: Infinispan
 
 // Component attributes
 :BrokerName: Apache ActiveMQ Artemis
diff --git a/documentation/design/proposals/bridging_external.adoc b/documentation/design/proposals/bridging_external.adoc
index 26e9c13a3..eb8603d38 100644
--- a/documentation/design/proposals/bridging_external.adoc
+++ b/documentation/design/proposals/bridging_external.adoc
@@ -130,20 +130,20 @@ For each forwarder on an address, the agent will create a connector from the bro
 * Set up an AMQP broker (i.e. Artemis) on a remote host
 * Configure an address space with connectors and create addresses with forwarders for the 4 different use cases listed initially
 
-List of functionalities/behaviours to test:
-  - [tested] Sending messages to a remote AMQP endpoint via a local address space - by creating a connector and using prefixing
-  - [tested] Receiving messages from a remote AMQP endpoint via a local address space - by creating a connector and using prefixing
-  - [tested] config a connector to refer to a host that does not exist. addressspace should report ready true and the connector's status should report the failure.
-  - invalid connector names (ie: using / in connector name)
-  - invalid patterns in address rule in connector (ie: queue*)
-  - Restart broker to ensure router reattached to broker and you can send/recv messages
-  - use TLS
-  - Using mutual TLS (SASL EXTERNAL) instead of credentials
-  - [tested] Forwarding messages from a local queue in a local address space to a destination on a remote AMQP endpoint
-  - [tested] Forwarding messages to a local queue in a local address space from a destination on a remote AMQP endpoint
-  - forward to FULL remote queue
-  - forward to FULL local queue
-  - try to forward messages when the remote broker is unavailable , and check how the messages are automatically forwarded when the remote broker comes back up
+=== List of functionalities/behaviours to test:
+* [tested] Sending messages to a remote AMQP endpoint via a local address space - by creating a connector and using prefixing
+* [tested] Receiving messages from a remote AMQP endpoint via a local address space - by creating a connector and using prefixing
+* [tested] config a connector to refer to a host that does not exist. addressspace should report ready true and the connector's status should report the failure.
+* [tested] invalid connector names (ie: using / in connector name)
+* [tested] invalid patterns in address rule in connector (ie: queue*)
+* [tested] Restart broker to ensure router reattached to broker and you can send/recv messages
+* use TLS
+* Using mutual TLS (SASL EXTERNAL) instead of credentials
+* [tested] Forwarding messages from a local queue in a local address space to a destination on a remote AMQP endpoint
+* [tested] Forwarding messages to a local queue in a local address space from a destination on a remote AMQP endpoint
+* forward to FULL remote queue
+* forward to FULL local queue
+* [tested] try to forward messages when the remote broker is unavailable , and check how the messages are automatically forwarded when the remote broker comes back up
 
 == Documentation
 
diff --git a/documentation/modules/proc-iot-creating-device.adoc b/documentation/modules/proc-iot-creating-device.adoc
index 1779a47ee..e4451cb0c 100644
--- a/documentation/modules/proc-iot-creating-device.adoc
+++ b/documentation/modules/proc-iot-creating-device.adoc
@@ -10,9 +10,11 @@ After installing the IoT services and creating an IoT project, you can create an
 
 ifeval::["{cmdcli}" == "oc"]
 :registry: $(oc -n enmasse-infra get routes device-registry --template='{{ .spec.host }}')
+:headers: -H 'Content-Type: application/json' -H "Authorization: Bearer ${TOKEN}"
 endif::[]
 ifeval::["{cmdcli}" == "kubectl"]
 :registry: $(kubectl -n enmasse-infra get service iot-device-registry-external -o jsonpath={.status.loadBalancer.ingress[0].hostname}):31443
+:headers: -H 'Content-Type: application/json'
 endif::[]
 
 .Prerequisites
@@ -25,6 +27,16 @@ endif::[]
 export REGISTRY_HOST={registry}
 ----
 
+ifeval::["{cmdcli}" == "oc"]
+* Export current user session token
++
+[options="nowrap",subs="attributes"]
+----
+export TOKEN=$(oc whoami --show-token)
+----
+This token is used to authenticate against device registry management API.
+endif::[]
+
 [id='iot-proc-creating-device-register-new-device-{context}']
 == Registering a new device
 
@@ -34,14 +46,14 @@ export REGISTRY_HOST={registry}
 +
 [options="nowrap",subs="attributes"]
 ----
-curl --insecure -X POST -i -H 'Content-Type: application/json' https://$REGISTRY_HOST/v1/devices/myapp.iot/4711
+curl --insecure -X POST -i {headers} https://$REGISTRY_HOST/v1/devices/myapp.iot/4711
 ----
 . If you need to provide additional registration information, you can provide
   it as follows:
 +
 [options="nowrap",subs="attributes"]
 ----
-curl --insecure -X POST -i -H 'Content-Type: application/json' --data-binary '{
+curl --insecure -X POST -i {headers} --data-binary '{
 	"via": ["gateway1"]
 }' https://$REGISTRY_HOST/v1/devices/myapp.iot/4711
 ----
@@ -55,7 +67,7 @@ curl --insecure -X POST -i -H 'Content-Type: application/json' --data-binary '{
 +
 [options="nowrap",subs="attributes"]
 ----
-curl --insecure -X PUT -i -H 'Content-Type: application/json' --data-binary '[{
+curl --insecure -X PUT -i {headers} --data-binary '[{
 	"type": "hashed-password",
 	"auth-id": "sensor1",
 	"secrets": [{
diff --git a/documentation/modules/proc-iot-installing-services.adoc b/documentation/modules/proc-iot-installing-services.adoc
index cca1e4ad1..a8afa85c3 100644
--- a/documentation/modules/proc-iot-installing-services.adoc
+++ b/documentation/modules/proc-iot-installing-services.adoc
@@ -52,6 +52,16 @@ CLUSTER=x.x.x.x.nip.io install/components/iot/examples/k8s-tls/create
 ----
 ====
 
+ifeval::["{cmdcli}" == "oc"]
+. Install {DataGridProductName} server
++
+[options="nowrap",subs="attributes"]
+----
+{cmdcli} apply -f install/components/iot/examples/infinispan/common
+{cmdcli} apply -f install/components/iot/examples/infinispan/manual
+----
+endif::[]
+
 . Install an example IoT infrastructure configuration:
 +
 [options="nowrap",subs="attributes"]
diff --git a/documentation/modules/ref-address-example-forwarders.adoc b/documentation/modules/ref-address-example-forwarders.adoc
new file mode 100644
index 000000000..b6920a22e
--- /dev/null
+++ b/documentation/modules/ref-address-example-forwarders.adoc
@@ -0,0 +1,63 @@
+// Module included in the following assemblies:
+//
+// assembly-managing-addresses.adoc
+
+[id='ref-address-example-forwarders-{context}']
+= Address forwarding examples
+
+You can use forwarders to:
+* automatically forward messages from a "local" address to a remote AMQP address outside of {ProductName}, or 
+* forward messages from a remote AMQP address to a "local" address. 
+To create a forwarder you must first configure a connector to the remote AMQP endpoint for the address space. For more information about connectors, see link:{BookUrlBase}{BaseProductVersion}{BookNameUrl}#ref-address-space-example-connectors-messaging[Address space connector examples].
+
+Forwarding is only supported in the `standard` address space type, and only for the `queue` and `subscription` address types. The `subscription` address type only supports forwarding in the `out` direction.
+
+These examples expect that a connector `remote1` has been configured for the address space.
+
+== Forwarding messages from a local queue to a remote endpoint
+
+In this example, messages in `myqueue` are forwarded to the remote endpoint with an  address of `clients/me/1`.
+
+that we can forward messages in  to.
+
+[source,yaml,options="nowrap"]
+----
+apiVersion: enmasse.io/v1beta1
+kind: Address
+metadata:
+  name: myspace.myqueue
+spec:
+  address: myqueue
+  type: queue
+  plan: standard-small-queue
+  forwarders:
+  - name: f1 <1>
+    remoteAddress: remote1/clients/me/1 <2>
+    direction: out <3>
+----
+<1> (Required) Specifies the name of the forwarder, which is used to ensure a unique identity.
+<2> (Required) Specifies the remote address to forward messages to. The address must be prefixed with the connector name and must be identical to the address matching patterns defined on the connector.
+<3> (Required) Specifies the direction of message flow, which is either `out` or `in`. A value of `out` forwards messages to the remote endpoint. A value of `in` forwards messages from the remote endpoint.
+
+== Forwarding messages from a remote endpoint to a local queue
+
+In this example, the remote endpoint has an address `prices/milk` that we can receive messages from and move to `myqueue`.
+
+[source,yaml,options="nowrap"]
+----
+apiVersion: enmasse.io/v1beta1
+kind: Address
+metadata:
+  name: myspace.myqueue
+spec:
+  address: myqueue
+  type: queue
+  plan: standard-small-queue
+  forwarders:
+  - name: f1 <1>
+    remoteAddress: remote1/prices/milk <2>
+    direction: in <3>
+----
+<1> (Required) Specifies the name of the forwarder, which is used to ensure a unique identity.
+<2> (Required) Specifies the remote address to forward messages to. The address must be prefixed with the connector name and must be identical to the address matching patterns defined on the connector.
+<3> (Required) Specifies the direction of message flow, which is either `out` or `in`. A value of `out` forwards messages to the remote endpoint. A value of `in` forwards messages from the remote endpoint.
diff --git a/documentation/modules/ref-address-space-example-connectors.adoc b/documentation/modules/ref-address-space-example-connectors.adoc
new file mode 100644
index 000000000..6e675bf86
--- /dev/null
+++ b/documentation/modules/ref-address-space-example-connectors.adoc
@@ -0,0 +1,90 @@
+// Module included in the following assemblies:
+//
+// assembly-managing-address-spaces.adoc
+
+[id='ref-address-space-example-connectors-{context}']
+= Address space connector examples
+
+Address space connectors are most commonly used in two ways. One use case is to map addresses on a remote AMQP endpoint into an address space. For example, a broker is running on the host `messaging.example.com` that you want to access by connecting using the {ProductName} endpoints. Another use case is to enable address forwarding. For more information about address forwarding, see link:{BookUrlBase}{BaseProductVersion}{BookNameUrl}#ref-address-example-forwarders-messaging[Address forwarding examples].
+
+The following address space examples show how you can configure connectors.
+
+== Connector using SASL PLAIN
+
+SASL PLAIN can be used when you do not want to use mutual TLS for authentication. Not enabling TLS is not recommended, since any usernames and passwords are then sent as clear text. 
+
+[source,yaml,options="nowrap"]
+----
+apiVersion: enmasse.io/v1beta1
+kind: AddressSpace
+metadata:
+  name: myspace
+spec:
+  type: standard
+  plan: standard-unlimited
+  connectors:
+  - name: remote1 <1>
+    endpointHosts: <2>
+    - host: messaging.example.com
+      port: 5672
+    - host: messaging2.example.com
+    tls: {} <3>
+    credentials: <4>
+      username:
+        value: test
+      password:
+        valueFromSecret:
+          name: password-secret
+          key: password.txt
+    addresses: <5>
+    - name: p1
+      pattern: "prices/*"
+    - name: p2
+      pattern: "clients/*/1"
+----
+<1> (Required) Specifies the name of the connector. All remote addresses are prefixed with the connector name and a forward slash, `/`.
+<2> (Required) Specifies a list of endpoints for this connector. This list must contain at least one entry, and any additional entries are used for failover. If not otherwise specified, the `port` field value is set to the registered IANA port for AMQP (or AMQPS if TLS is enabled).
+<3> (Optional) Enable TLS. The connector trusts global root CAs by default. To use a custom CA, specify a value for the `caCert` field.
+<4> (Optional) Specifies the username and password credentials to use for this connector. The values can be specified inline or by referencing a secret along with an optional key specifying the location within the secret. The secret must be readable by the {ProductName} `address-space-controller` service account.
+<5> (Required) Specifies a list of patterns matching addresses to be exposed on the remote endpoint. The pattern consists of one or more tokens separated by a forward slash, `/`. A token can be one of the following: a * character, a # character, or a sequence of characters that do not include /, *, or #. The * token matches any single token. The # token matches zero or more tokens. * has higher precedence than #, and exact match has the highest precedence.
+
+== Connector using mutual TLS
+
+Configuring a client TLS certificate enables SASL EXTERNAL to be used for authentication. The certificates can be specified inline or using a secret reference.
+
+[source,yaml,options="nowrap"]
+----
+apiVersion: enmasse.io/v1beta1
+kind: AddressSpace
+metadata:
+  name: myspace
+spec:
+  type: standard
+  plan: standard-unlimited
+  connectors:
+  - name: remote1 <1>
+    endpointHosts: <2>
+    - host: messaging.example.com
+      port: 5671
+    tls: 
+      caCert: <3>
+        valueFromSecret:
+          name: remote-certs
+          key: ca.crt
+      clientCert: <4>
+        valueFromSecret:
+          name: remote-certs
+          key: tls.crt
+      clientKey: <5>
+        valueFromSecret:
+          name: remote-certs
+          key: tls.key
+    addresses:
+    - name: p1
+      pattern: "*"
+----
+<1> (Required) Specifies the name of the connector. All remote addresses are prefixed with the connector name and a forward slash, `/`.
+<2> (Required) Specifies a list of endpoints for this connector. This list must contain at least one entry, and any additional entries are used for failover. If not otherwise specified, the `port` field value is set to the registered IANA port for AMQP (or AMQPS if TLS is enabled).
+<3> (Optional) Specifies the CA certificate to trust for the remote connection. The referenced secret must be readable by the {ProductName} `address-space-controller` service account.
+<4> (Optional) Specifies the client certificate to use for mutual TLS authentication. The referenced secret must be readable by the {ProductName} `address-space-controller` service account.
+<5> (Optional) Specifies the client private key to use for mutual TLS authentication. The referenced secret must be readable by the {ProductName} `address-space-controller` service account.
diff --git a/k8s-api/src/main/java/io/enmasse/k8s/api/ConfigMapAddressApi.java b/k8s-api/src/main/java/io/enmasse/k8s/api/ConfigMapAddressApi.java
index 00808fb5e..3449195d7 100644
--- a/k8s-api/src/main/java/io/enmasse/k8s/api/ConfigMapAddressApi.java
+++ b/k8s-api/src/main/java/io/enmasse/k8s/api/ConfigMapAddressApi.java
@@ -16,6 +16,7 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import io.fabric8.kubernetes.api.model.OwnerReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -52,10 +53,14 @@ public class ConfigMapAddressApi implements AddressApi, ListerWatcher<ConfigMap,
     private final String infraUuid;
     private final WorkQueue<ConfigMap> cache = new EventCache<>(new HasMetadataFieldExtractor<>());
     private ObjectMapper mapper = new ObjectMapper();
+    private final OwnerReference ownerReference;
+    private final String version;
 
-    public ConfigMapAddressApi(NamespacedKubernetesClient client, String infraUuid) {
+    public ConfigMapAddressApi(NamespacedKubernetesClient client, String infraUuid, OwnerReference ownerReference, String version) {
         this.client = client;
         this.infraUuid = infraUuid;
+        this.ownerReference = ownerReference;
+        this.version = version;
     }
 
     @Override
@@ -173,17 +178,28 @@ public class ConfigMapAddressApi implements AddressApi, ListerWatcher<ConfigMap,
     }
 
     private ConfigMap create(Address address) {
+        String addressSpaceName = Address.extractAddressSpace(address);
         ConfigMapBuilder builder = new ConfigMapBuilder()
                 .editOrNewMetadata()
                 .withName(getConfigMapName(address.getMetadata().getNamespace(), address.getMetadata().getName()))
                 .addToLabels(address.getMetadata().getLabels())
                 .addToLabels(LabelKeys.TYPE, "address-config")
                 .addToLabels(LabelKeys.INFRA_UUID, infraUuid)
-                .addToLabels(LabelKeys.INFRA_TYPE, "any")
+
                 .addToAnnotations(address.getMetadata().getAnnotations())
                 // TODO: Support other ways of doing this
-                .addToAnnotations(AnnotationKeys.ADDRESS_SPACE, Address.extractAddressSpace(address))
+                .addToAnnotations(AnnotationKeys.ADDRESS_SPACE, addressSpaceName)
+                .endMetadata();
+
+        if (ownerReference != null) {
+            builder.editMetadata()
+                .addToOwnerReferences(ownerReference)
                 .endMetadata();
+        }
+
+        if (version != null) {
+            address.putAnnotation(AnnotationKeys.VERSION, version);
+        }
 
         if (address.getMetadata().getResourceVersion() != null) {
             builder.editOrNewMetadata()
diff --git a/k8s-api/src/main/java/io/enmasse/k8s/api/ConfigMapAddressSpaceApi.java b/k8s-api/src/main/java/io/enmasse/k8s/api/ConfigMapAddressSpaceApi.java
index b10471934..f921a591b 100644
--- a/k8s-api/src/main/java/io/enmasse/k8s/api/ConfigMapAddressSpaceApi.java
+++ b/k8s-api/src/main/java/io/enmasse/k8s/api/ConfigMapAddressSpaceApi.java
@@ -17,6 +17,7 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import io.fabric8.kubernetes.api.model.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -35,9 +36,6 @@ import io.enmasse.k8s.api.cache.ListOptions;
 import io.enmasse.k8s.api.cache.ListerWatcher;
 import io.enmasse.k8s.api.cache.Reflector;
 import io.enmasse.k8s.api.cache.WorkQueue;
-import io.fabric8.kubernetes.api.model.ConfigMap;
-import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
-import io.fabric8.kubernetes.api.model.ConfigMapList;
 import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
 import io.fabric8.kubernetes.client.RequestConfig;
@@ -50,12 +48,14 @@ public class ConfigMapAddressSpaceApi implements AddressSpaceApi, ListerWatcher<
     protected final Logger log = LoggerFactory.getLogger(getClass().getName());
     private final NamespacedKubernetesClient client;
     private final ObjectMapper mapper = new ObjectMapper();
+    private final String version;
 
-    public ConfigMapAddressSpaceApi(NamespacedKubernetesClient client) {
+    public ConfigMapAddressSpaceApi(NamespacedKubernetesClient client, String version) {
         this.client = client;
+        this.version = version;
     }
 
-    private static String getConfigMapName(String namespace, String name) {
+    public static String getConfigMapName(String namespace, String name) {
         return namespace + "." + name;
     }
 
@@ -245,7 +245,15 @@ public class ConfigMapAddressSpaceApi implements AddressSpaceApi, ListerWatcher<
 
     @Override
     public AddressApi withAddressSpace(AddressSpace addressSpace) {
-        return new ConfigMapAddressApi(client, addressSpace.getAnnotation(AnnotationKeys.INFRA_UUID));
+        OwnerReference ownerReference = new OwnerReferenceBuilder()
+                .withApiVersion("v1")
+                .withKind("ConfigMap")
+                .withBlockOwnerDeletion(true)
+                .withController(true)
+                .withName(getConfigMapName(addressSpace.getMetadata().getNamespace(), addressSpace.getMetadata().getName()))
+                .withUid(addressSpace.getMetadata().getUid())
+                .build();
+        return new ConfigMapAddressApi(client, addressSpace.getAnnotation(AnnotationKeys.INFRA_UUID), ownerReference, version);
     }
 
     @SuppressWarnings("unused")
diff --git a/keycloak-user-api/src/main/java/io/enmasse/user/keycloak/KeycloakUserApi.java b/keycloak-user-api/src/main/java/io/enmasse/user/keycloak/KeycloakUserApi.java
index 875c7ff18..498123628 100644
--- a/keycloak-user-api/src/main/java/io/enmasse/user/keycloak/KeycloakUserApi.java
+++ b/keycloak-user-api/src/main/java/io/enmasse/user/keycloak/KeycloakUserApi.java
@@ -24,7 +24,6 @@ import org.slf4j.LoggerFactory;
 import javax.ws.rs.WebApplicationException;
 import javax.ws.rs.core.Response;
 import java.io.IOException;
-import java.io.UnsupportedEncodingException;
 import java.net.URLDecoder;
 import java.net.URLEncoder;
 import java.nio.charset.Charset;
@@ -712,18 +711,10 @@ public class KeycloakUserApi implements UserApi, RealmApi {
     }
 
     public static String decodePart(final String part) {
-        try {
-            return URLDecoder.decode(part, StandardCharsets.UTF_8.name());
-        } catch (final UnsupportedEncodingException e) {
-            throw new RuntimeException(e);
-        }
+        return URLDecoder.decode(part, StandardCharsets.UTF_8);
     }
 
     public static String encodePart(final String part) {
-        try {
-            return URLEncoder.encode(part, StandardCharsets.UTF_8.name());
-        } catch (UnsupportedEncodingException e) {
-            throw new RuntimeException(e);
-        }
+        return URLEncoder.encode(part, StandardCharsets.UTF_8);
     }
 }
diff --git a/pkg/apis/enmasse/v1beta1/types.go b/pkg/apis/enmasse/v1beta1/types.go
index 400adefe0..361596cff 100644
--- a/pkg/apis/enmasse/v1beta1/types.go
+++ b/pkg/apis/enmasse/v1beta1/types.go
@@ -6,8 +6,7 @@
 package v1beta1
 
 import (
-	"encoding/json"
-
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -40,30 +39,19 @@ type AddressSpaceSpec struct {
 }
 
 type AuthenticationService struct {
-	Type    string          `json:"type"`
-	Details json.RawMessage `json:"details,omitempty"`
-}
+	Type string `json:"type,omitempty"`
+	Name string `json:"name,omitempty"`
 
-type Detail struct {
-	StringValue  *string
-	FloatValue   *float64
-	BooleanValue *bool
+	Overrides *AuthenticationServiceSettings `json:"overrides,omitempty"`
 }
 
-func (d *Detail) UnmarshalJSON(b []byte) error {
-	return nil
-}
+type AuthenticationServiceSettings struct {
+	Host  string `json:"host,omitempty"`
+	Port  int    `json:"port,omitempty"`
+	Realm string `json:"realm,omitempty"`
 
-func (d Detail) MarshalJSON() ([]byte, error) {
-	if d.StringValue != nil {
-		return json.Marshal(d.StringValue)
-	} else if d.FloatValue != nil {
-		return json.Marshal(d.FloatValue)
-	} else if d.BooleanValue != nil {
-		return json.Marshal(d.BooleanValue)
-	} else {
-		return []byte("null"), nil
-	}
+	CaCertSecret     *corev1.SecretReference `json:"caCertSecret,omitempty"`
+	ClientCertSecret *corev1.SecretReference `json:"clientCertSecret,omitempty"`
 }
 
 type EndpointSpec struct {
diff --git a/pkg/apis/enmasse/v1beta1/zz_generated.deepcopy.go b/pkg/apis/enmasse/v1beta1/zz_generated.deepcopy.go
index 224e0f351..55de08716 100644
--- a/pkg/apis/enmasse/v1beta1/zz_generated.deepcopy.go
+++ b/pkg/apis/enmasse/v1beta1/zz_generated.deepcopy.go
@@ -10,8 +10,7 @@
 package v1beta1
 
 import (
-	json "encoding/json"
-
+	v1 "k8s.io/api/core/v1"
 	runtime "k8s.io/apimachinery/pkg/runtime"
 )
 
@@ -262,10 +261,10 @@ func (in *AddressStatus) DeepCopy() *AddressStatus {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *AuthenticationService) DeepCopyInto(out *AuthenticationService) {
 	*out = *in
-	if in.Details != nil {
-		in, out := &in.Details, &out.Details
-		*out = make(json.RawMessage, len(*in))
-		copy(*out, *in)
+	if in.Overrides != nil {
+		in, out := &in.Overrides, &out.Overrides
+		*out = new(AuthenticationServiceSettings)
+		(*in).DeepCopyInto(*out)
 	}
 	return
 }
@@ -281,64 +280,59 @@ func (in *AuthenticationService) DeepCopy() *AuthenticationService {
 }
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *BrokerStatus) DeepCopyInto(out *BrokerStatus) {
+func (in *AuthenticationServiceSettings) DeepCopyInto(out *AuthenticationServiceSettings) {
 	*out = *in
+	if in.CaCertSecret != nil {
+		in, out := &in.CaCertSecret, &out.CaCertSecret
+		*out = new(v1.SecretReference)
+		**out = **in
+	}
+	if in.ClientCertSecret != nil {
+		in, out := &in.ClientCertSecret, &out.ClientCertSecret
+		*out = new(v1.SecretReference)
+		**out = **in
+	}
 	return
 }
 
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BrokerStatus.
-func (in *BrokerStatus) DeepCopy() *BrokerStatus {
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AuthenticationServiceSettings.
+func (in *AuthenticationServiceSettings) DeepCopy() *AuthenticationServiceSettings {
 	if in == nil {
 		return nil
 	}
-	out := new(BrokerStatus)
+	out := new(AuthenticationServiceSettings)
 	in.DeepCopyInto(out)
 	return out
 }
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *CertificateSpec) DeepCopyInto(out *CertificateSpec) {
+func (in *BrokerStatus) DeepCopyInto(out *BrokerStatus) {
 	*out = *in
 	return
 }
 
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CertificateSpec.
-func (in *CertificateSpec) DeepCopy() *CertificateSpec {
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BrokerStatus.
+func (in *BrokerStatus) DeepCopy() *BrokerStatus {
 	if in == nil {
 		return nil
 	}
-	out := new(CertificateSpec)
+	out := new(BrokerStatus)
 	in.DeepCopyInto(out)
 	return out
 }
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *Detail) DeepCopyInto(out *Detail) {
+func (in *CertificateSpec) DeepCopyInto(out *CertificateSpec) {
 	*out = *in
-	if in.StringValue != nil {
-		in, out := &in.StringValue, &out.StringValue
-		*out = new(string)
-		**out = **in
-	}
-	if in.FloatValue != nil {
-		in, out := &in.FloatValue, &out.FloatValue
-		*out = new(float64)
-		**out = **in
-	}
-	if in.BooleanValue != nil {
-		in, out := &in.BooleanValue, &out.BooleanValue
-		*out = new(bool)
-		**out = **in
-	}
 	return
 }
 
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Detail.
-func (in *Detail) DeepCopy() *Detail {
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CertificateSpec.
+func (in *CertificateSpec) DeepCopy() *CertificateSpec {
 	if in == nil {
 		return nil
 	}
-	out := new(Detail)
+	out := new(CertificateSpec)
 	in.DeepCopyInto(out)
 	return out
 }
diff --git a/pom.xml b/pom.xml
index 4d0a12f3d..1ec999444 100644
--- a/pom.xml
+++ b/pom.xml
@@ -68,7 +68,7 @@
         <junit.platform.launcher.version>1.5.0</junit.platform.launcher.version>
         <artemis.version>2.10.0</artemis.version>
         <ngwebdriver.version>1.1.4</ngwebdriver.version>
-        <paho.version>1.2.0</paho.version>
+        <paho.version>1.2.1</paho.version>
         <netty.version>4.1.34.Final</netty.version>
         <keycloak.version>4.8.3.Final</keycloak.version>
         <jboss.logging.version>3.3.0.Final</jboss.logging.version>
diff --git a/service-broker/src/main/java/io/enmasse/osb/ServiceBroker.java b/service-broker/src/main/java/io/enmasse/osb/ServiceBroker.java
index 8eefde1be..868bad302 100644
--- a/service-broker/src/main/java/io/enmasse/osb/ServiceBroker.java
+++ b/service-broker/src/main/java/io/enmasse/osb/ServiceBroker.java
@@ -69,7 +69,7 @@ public class ServiceBroker extends AbstractVerticle {
         ensureRouteExists(client.adapt(OpenShiftClient.class), options);
         ensureCredentialsExist(client, options);
 
-        AddressSpaceApi addressSpaceApi = new ConfigMapAddressSpaceApi(client);
+        AddressSpaceApi addressSpaceApi = new ConfigMapAddressSpaceApi(client, options.getVersion());
         AuthApi authApi = new KubeAuthApi(client, client.getConfiguration().getOauthToken());
 
         UserApi userApi = createUserApi(options, schemaProvider);
diff --git a/standard-controller/src/main/java/io/enmasse/controller/standard/AddressController.java b/standard-controller/src/main/java/io/enmasse/controller/standard/AddressController.java
index cac1899c2..0a1e4716d 100644
--- a/standard-controller/src/main/java/io/enmasse/controller/standard/AddressController.java
+++ b/standard-controller/src/main/java/io/enmasse/controller/standard/AddressController.java
@@ -204,10 +204,15 @@ public class AddressController implements Watcher<Address> {
         checkAndMoveMigratingBrokersToDraining(addressSet, clusterList);
         checkAndRemoveDrainingBrokers(addressSet);
 
-        deprovisionUnused(clusterList, filterByNotPhases(addressSet, EnumSet.of(Terminating)));
+        Set<Address> notTerminating = filterByNotPhases(addressSet, EnumSet.of(Terminating));
+        List<BrokerCluster> unusedClusters = determineUnusedClusters(clusterList, notTerminating);
+        deprovisionUnused(unusedClusters);
+
         long deprovisionUnused = System.nanoTime();
 
-        upgradeClusters(desiredConfig, addressResolver, clusterList, filterByNotPhases(addressSet, EnumSet.of(Terminating)));
+        List<BrokerCluster> usedClusters = new ArrayList<>(clusterList);
+        usedClusters.removeAll(unusedClusters);
+        upgradeClusters(desiredConfig, addressResolver, usedClusters, notTerminating);
 
         long upgradeClusters = System.nanoTime();
 
@@ -408,7 +413,9 @@ public class AddressController implements Watcher<Address> {
         );
     }
 
-    private void deprovisionUnused(List<BrokerCluster> clusters, Set<Address> addressSet) {
+    private List<BrokerCluster> determineUnusedClusters(List<BrokerCluster> clusters, Set<Address> addressSet) {
+        List<BrokerCluster> unused = new ArrayList<>();
+
         for (BrokerCluster cluster : clusters) {
             int numFound = 0;
             for (Address address : addressSet) {
@@ -426,15 +433,22 @@ public class AddressController implements Watcher<Address> {
             }
 
             if (numFound == 0) {
-                try {
-                    kubernetes.delete(cluster.getResources());
-                    eventLogger.log(ControllerReason.BrokerDeleted, "Deleted broker " + cluster.getClusterId(), Normal, ControllerKind.Address, cluster.getClusterId());
-                } catch (Exception e) {
-                    log.warn("Error deleting cluster {}", cluster.getClusterId(), e);
-                    eventLogger.log(ControllerReason.BrokerDeleteFailed, "Error deleting broker cluster " + cluster.getClusterId() + ": " + e.getMessage(), EventLogger.Type.Warning, ControllerKind.Address, cluster.getClusterId());
-                }
+                unused.add(cluster);
             }
         }
+        return unused;
+    }
+
+    private void deprovisionUnused(List<BrokerCluster> unused) {
+        unused.forEach(cluster -> {
+            try {
+                kubernetes.delete(cluster.getResources());
+                eventLogger.log(ControllerReason.BrokerDeleted, "Deleted broker " + cluster.getClusterId(), Normal, ControllerKind.Address, cluster.getClusterId());
+            } catch (Exception e) {
+                log.warn("Error deleting cluster {}", cluster.getClusterId(), e);
+                eventLogger.log(ControllerReason.BrokerDeleteFailed, "Error deleting broker cluster " + cluster.getClusterId() + ": " + e.getMessage(), EventLogger.Type.Warning, ControllerKind.Address, cluster.getClusterId());
+            }
+        });
     }
 
     private Set<Address> filterBy(Set<Address> addressSet, Predicate<Address> predicate) {
diff --git a/standard-controller/src/main/java/io/enmasse/controller/standard/StandardController.java b/standard-controller/src/main/java/io/enmasse/controller/standard/StandardController.java
index b3f24a58b..aa9397a91 100644
--- a/standard-controller/src/main/java/io/enmasse/controller/standard/StandardController.java
+++ b/standard-controller/src/main/java/io/enmasse/controller/standard/StandardController.java
@@ -7,16 +7,15 @@ package io.enmasse.controller.standard;
 import java.time.Clock;
 import java.util.Map;
 
+import io.enmasse.address.model.Address;
+import io.enmasse.address.model.AddressSpace;
+import io.enmasse.config.AnnotationKeys;
+import io.enmasse.k8s.api.*;
+import io.fabric8.kubernetes.api.model.OwnerReference;
+import io.fabric8.kubernetes.api.model.OwnerReferenceBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.enmasse.k8s.api.CachingSchemaProvider;
-import io.enmasse.k8s.api.ConfigMapAddressApi;
-import io.enmasse.k8s.api.EventLogger;
-import io.enmasse.k8s.api.KubeEventLogger;
-import io.enmasse.k8s.api.KubeSchemaApi;
-import io.enmasse.k8s.api.LogEventLogger;
-import io.enmasse.k8s.api.SchemaApi;
 import io.enmasse.metrics.api.Metrics;
 import io.enmasse.model.CustomResourceDefinitions;
 import io.fabric8.kubernetes.client.DefaultKubernetesClient;
@@ -90,9 +89,36 @@ public class StandardController {
 
         BrokerClientFactory brokerClientFactory = new MutualTlsBrokerClientFactory(vertx, options);
 
+        AddressSpaceApi addressSpaceApi = new ConfigMapAddressSpaceApi(kubeClient, options.getVersion());
+        AddressSpace addressSpace = addressSpaceApi.getAddressSpaceWithName(options.getAddressSpaceNamespace(), options.getAddressSpace()).orElseThrow(() ->
+                new IllegalStateException("Unable to lookup address space " + options.getAddressSpace()));
+
+        OwnerReference ownerReference = new OwnerReferenceBuilder()
+                .withApiVersion("v1")
+                .withKind("ConfigMap")
+                .withBlockOwnerDeletion(true)
+                .withController(true)
+                .withName(ConfigMapAddressSpaceApi.getConfigMapName(addressSpace.getMetadata().getNamespace(), addressSpace.getMetadata().getName()))
+                .withUid(addressSpace.getMetadata().getUid())
+                .build();
+
+        ConfigMapAddressApi addressApi = new ConfigMapAddressApi(kubeClient, options.getInfraUuid(), ownerReference, options.getVersion());
+
+        // Replace resources as part of upgrade if version is different
+        for (Address address : addressApi.listAddresses(options.getAddressSpaceNamespace())) {
+            if (!options.getVersion().equals(address.getAnnotation(AnnotationKeys.VERSION))) {
+                try {
+                    // Version will be updated by replaceAddress
+                    addressApi.replaceAddress(address);
+                } catch (Exception e) {
+                    log.warn("Error replacing {}", address.getMetadata().getName(), e);
+                }
+            }
+        }
+
         addressController = new AddressController(
                 options,
-                new ConfigMapAddressApi(kubeClient, options.getInfraUuid()),
+                addressApi,
                 kubernetes,
                 clusterGenerator,
                 eventLogger,
diff --git a/templates/address-space-controller/020-Role-address-space-admin.yaml b/templates/address-space-controller/020-Role-address-space-admin.yaml
index f9bd88942..db461d93b 100644
--- a/templates/address-space-controller/020-Role-address-space-admin.yaml
+++ b/templates/address-space-controller/020-Role-address-space-admin.yaml
@@ -12,7 +12,7 @@ rules:
     resources: [ "pods", "secrets" ]
     verbs: [ "get", "list", "watch" ]
   - apiGroups: [ "" ]
-    resources: [ "configmaps" ]
+    resources: [ "configmaps", "configmaps/finalizers" ]
     verbs: [ "create", "update", "patch", "get", "list", "watch", "delete" ]
   - apiGroups: [ "" ]
     resources: [ "events" ]
diff --git a/templates/address-space-controller/020-Role-address-space-controller.yaml b/templates/address-space-controller/020-Role-address-space-controller.yaml
index 629ae3fee..6a039a4bb 100644
--- a/templates/address-space-controller/020-Role-address-space-controller.yaml
+++ b/templates/address-space-controller/020-Role-address-space-controller.yaml
@@ -12,7 +12,7 @@ rules:
     resources: [ "pods" ]
     verbs: [ "get", "list", "patch", "update" ]
   - apiGroups: [ "" ]
-    resources: [ "configmaps" ]
+    resources: [ "configmaps", "configmaps/finalizers" ]
     verbs: [ "create", "update", "patch", "get", "list", "watch", "delete" ]
   - apiGroups: [ "" ]
     resources: [ "events" ]
diff --git a/templates/api-server/020-Role-api-server.yaml b/templates/api-server/020-Role-api-server.yaml
index 392c892a8..0dbd93e01 100644
--- a/templates/api-server/020-Role-api-server.yaml
+++ b/templates/api-server/020-Role-api-server.yaml
@@ -9,7 +9,7 @@ rules:
     resources: [ "services", "secrets" ]
     verbs: [ "get", "list", "watch" ]
   - apiGroups: [ "" ]
-    resources: [ "configmaps" ]
+    resources: [ "configmaps", "configmaps/finalizers" ]
     verbs: [ "create", "update", "patch", "get", "list", "watch", "delete" ]
   - apiGroups: [ "admin.enmasse.io" ]
     resources: [ "addressspaceplans", "addressplans", "standardinfraconfigs", "brokeredinfraconfigs", "authenticationservices", "consoleservices"]
diff --git a/templates/iot/examples/iot-config-filebased.yaml b/templates/iot/examples/iot-config-filebased.yaml
new file mode 100644
index 000000000..4bbb7f42c
--- /dev/null
+++ b/templates/iot/examples/iot-config-filebased.yaml
@@ -0,0 +1,14 @@
+kind: IoTConfig
+apiVersion: iot.enmasse.io/v1alpha1
+metadata:
+  name: default
+spec:
+  services:
+    deviceRegistry:
+      file:
+        numberOfDevicesPerTenant: 100000
+  adapters:
+    mqtt:
+      endpoint:
+        secretNameStrategy:
+          secretName: iot-mqtt-adapter-tls
diff --git a/templates/iot/examples/iot-config-infinispan.yaml b/templates/iot/examples/iot-config-infinispan.yaml
deleted file mode 100644
index 050e6ce2c..000000000
--- a/templates/iot/examples/iot-config-infinispan.yaml
+++ /dev/null
@@ -1,21 +0,0 @@
-kind: IoTConfig
-apiVersion: iot.enmasse.io/v1alpha1
-metadata:
-  name: default
-spec:
-  services:
-    deviceRegistry:
-      infinispan:
-        server:
-          external:
-            host: infinispan-server
-            port: 11222
-            username: app
-            password: test12
-            saslServerName: hotrod
-            saslRealm: ApplicationRealm
-  adapters:
-    mqtt:
-      endpoint:
-        secretNameStrategy:
-          secretName: iot-mqtt-adapter-tls
diff --git a/templates/iot/examples/iot-config.yaml b/templates/iot/examples/iot-config.yaml
index 4bbb7f42c..050e6ce2c 100644
--- a/templates/iot/examples/iot-config.yaml
+++ b/templates/iot/examples/iot-config.yaml
@@ -5,8 +5,15 @@ metadata:
 spec:
   services:
     deviceRegistry:
-      file:
-        numberOfDevicesPerTenant: 100000
+      infinispan:
+        server:
+          external:
+            host: infinispan-server
+            port: 11222
+            username: app
+            password: test12
+            saslServerName: hotrod
+            saslRealm: ApplicationRealm
   adapters:
     mqtt:
       endpoint:


Test patch:
diff --git a/api-model/src/test/java/io/enmasse/address/model/AddressSpaceSpecConnectorAddressRuleTest.java b/api-model/src/test/java/io/enmasse/address/model/AddressSpaceSpecConnectorAddressRuleTest.java
index e84a12758..7d8821730 100644
--- a/api-model/src/test/java/io/enmasse/address/model/AddressSpaceSpecConnectorAddressRuleTest.java
+++ b/api-model/src/test/java/io/enmasse/address/model/AddressSpaceSpecConnectorAddressRuleTest.java
@@ -6,11 +6,7 @@ package io.enmasse.address.model;
 
 import org.junit.jupiter.api.Test;
 
-import java.util.ArrayList;
-
-import static io.enmasse.address.model.validation.ValidationMatchers.isNotValid;
 import static io.enmasse.address.model.validation.ValidationMatchers.isValid;
-import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -40,6 +36,7 @@ public class AddressSpaceSpecConnectorAddressRuleTest {
         assertFalse(validRule("a", "*/"));
         assertFalse(validRule("a", "*//"));
         assertFalse(validRule("a", "//"));
+        assertFalse(validRule("a", "a*"));
     }
 
     private boolean validRule(String name, String pattern) {
diff --git a/k8s-api/src/test/java/io/enmasse/k8s/api/ConfigMapAddressApiTest.java b/k8s-api/src/test/java/io/enmasse/k8s/api/ConfigMapAddressApiTest.java
index 524b28dfd..430e17c38 100644
--- a/k8s-api/src/test/java/io/enmasse/k8s/api/ConfigMapAddressApiTest.java
+++ b/k8s-api/src/test/java/io/enmasse/k8s/api/ConfigMapAddressApiTest.java
@@ -34,7 +34,7 @@ class ConfigMapAddressApiTest extends JULInitializingTest {
     void setUp() {
         kubeServer.before();
         NamespacedKubernetesClient client = kubeServer.getClient();
-        api = new ConfigMapAddressApi(client, UUID.randomUUID().toString());
+        api = new ConfigMapAddressApi(client, UUID.randomUUID().toString(), null, null);
     }
 
     @AfterEach
diff --git a/k8s-api/src/test/java/io/enmasse/k8s/api/ConfigMapAddressSpaceApiTest.java b/k8s-api/src/test/java/io/enmasse/k8s/api/ConfigMapAddressSpaceApiTest.java
index 3b07c859f..bbb365f50 100644
--- a/k8s-api/src/test/java/io/enmasse/k8s/api/ConfigMapAddressSpaceApiTest.java
+++ b/k8s-api/src/test/java/io/enmasse/k8s/api/ConfigMapAddressSpaceApiTest.java
@@ -38,7 +38,7 @@ class ConfigMapAddressSpaceApiTest extends JULInitializingTest {
     @BeforeEach
     void setUp() {
         kubeServer.before();
-        api = new ConfigMapAddressSpaceApi(kubeServer.getClient());
+        api = new ConfigMapAddressSpaceApi(kubeServer.getClient(), null);
     }
 
     @AfterEach
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/cmdclients/KubeCMDClient.java b/systemtests/src/main/java/io/enmasse/systemtest/cmdclients/KubeCMDClient.java
index bef560d98..a4b68c058 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/cmdclients/KubeCMDClient.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/cmdclients/KubeCMDClient.java
@@ -264,11 +264,6 @@ public class KubeCMDClient extends CmdClient {
                 .replaceAll(System.getProperty("line.separator"), "");
     }
 
-    public static ExecutionResultData createOcUser(String username) {
-        List<String> deleteCmd = Arrays.asList(CMD, "create", "user", username);
-        return execute(deleteCmd, DEFAULT_SYNC_TIMEOUT, true);
-    }
-
     public static ExecutionResultData getEvents(String namespace) {
         List<String> command = Arrays.asList(CMD, "get", "events",
                 "--namespace", namespace,
@@ -348,4 +343,9 @@ public class KubeCMDClient extends CmdClient {
         List<String> ressourcesCmd = getRessourcesCmd("delete", resource, namespace, name, Optional.empty());
         return execute(ressourcesCmd, DEFAULT_SYNC_TIMEOUT, true);
     }
+
+    public static ExecutionResultData createGroupAndAddUser(String groupName, String username) {
+        return execute(Arrays.asList(CMD, "adm", "groups", "new", groupName, username), DEFAULT_SYNC_TIMEOUT, true);
+    }
+
 }
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/manager/ResourceManager.java b/systemtests/src/main/java/io/enmasse/systemtest/manager/ResourceManager.java
index df7da53aa..32a74307c 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/manager/ResourceManager.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/manager/ResourceManager.java
@@ -174,10 +174,16 @@ public abstract class ResourceManager {
     }
 
     public void createAuthService(AuthenticationService authenticationService) throws Exception {
+        createAuthService(authenticationService, true);
+    }
+
+    public void createAuthService(AuthenticationService authenticationService, boolean wait) throws Exception {
         var client = Kubernetes.getInstance().getAuthenticationServiceClient();
         LOGGER.info("AuthService {} will be created {}", authenticationService.getMetadata().getName(), authenticationService);
         client.create(authenticationService);
-        waitForAuthPods(authenticationService);
+        if (wait) {
+            waitForAuthPods(authenticationService);
+        }
     }
 
 
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/messagingclients/ClientArgument.java b/systemtests/src/main/java/io/enmasse/systemtest/messagingclients/ClientArgument.java
index a269bafa9..87da08693 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/messagingclients/ClientArgument.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/messagingclients/ClientArgument.java
@@ -67,6 +67,7 @@ public enum ClientArgument {
     CLOSE_SLEEP("--close-sleep"),
     TIMEOUT("--timeout"),
     DURATION("--duration"),
+    DEST_TYPE("--dest-type"),
 
     //sender opts
     MSG_ID("--msg-id"),
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/messagingclients/ClientType.java b/systemtests/src/main/java/io/enmasse/systemtest/messagingclients/ClientType.java
index 2bf7d3fe2..570f2ba3c 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/messagingclients/ClientType.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/messagingclients/ClientType.java
@@ -17,7 +17,9 @@ public enum ClientType {
     CLI_JAVA_OPENWIRE_JMS_SENDER,
     CLI_JAVA_OPENWIRE_JMS_RECEIVER,
     CLI_JAVA_PAHO_MQTT_SENDER,
-    CLI_JAVA_PAHO_MQTT_RECEIVER;
+    CLI_JAVA_PAHO_MQTT_RECEIVER,
+    CLI_STOMP_SENDER,
+    CLI_STOMP_RECEIVER;
 
     /**
      * Get bind client type to client executable
@@ -53,6 +55,9 @@ public enum ClientType {
                 return "./client_executable/cli-paho-java.jar";
             case CLI_JAVA_PAHO_MQTT_RECEIVER:
                 return "./client_executable/cli-paho-java.jar";
+            case CLI_STOMP_SENDER:
+            case CLI_STOMP_RECEIVER:
+                return "./client_executable/stomp-client.py";
             default:
                 return "";
         }
@@ -76,6 +81,9 @@ public enum ClientType {
             case CLI_JAVA_OPENWIRE_JMS_SENDER:
             case CLI_JAVA_OPENWIRE_JMS_RECEIVER:
                 return "cli_openwire";
+            case CLI_STOMP_SENDER:
+            case CLI_STOMP_RECEIVER:
+                return "cli_stomp";
             default:
                 return "";
         }
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/messagingclients/stomp/StompClientReceiver.java b/systemtests/src/main/java/io/enmasse/systemtest/messagingclients/stomp/StompClientReceiver.java
new file mode 100644
index 000000000..ae614ad24
--- /dev/null
+++ b/systemtests/src/main/java/io/enmasse/systemtest/messagingclients/stomp/StompClientReceiver.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2019, EnMasse authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.enmasse.systemtest.messagingclients.stomp;
+
+import io.enmasse.systemtest.messagingclients.AbstractClient;
+import io.enmasse.systemtest.messagingclients.ClientArgument;
+import io.enmasse.systemtest.messagingclients.ClientArgumentMap;
+import io.enmasse.systemtest.messagingclients.ClientType;
+
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.List;
+
+public class StompClientReceiver extends AbstractClient {
+    public StompClientReceiver() throws Exception {
+        super(ClientType.CLI_STOMP_RECEIVER);
+    }
+
+    public StompClientReceiver(Path logPath) throws Exception {
+        super(ClientType.CLI_STOMP_RECEIVER, logPath);
+    }
+
+    @Override
+    protected void fillAllowedArgs() {
+        allowedArgs.add(ClientArgument.BROKER);
+        allowedArgs.add(ClientArgument.ADDRESS);
+        allowedArgs.add(ClientArgument.COUNT);
+        allowedArgs.add(ClientArgument.TIMEOUT);
+        allowedArgs.add(ClientArgument.USERNAME);
+        allowedArgs.add(ClientArgument.PASSWORD);
+        allowedArgs.add(ClientArgument.DEST_TYPE);
+    }
+
+    @Override
+    protected ClientArgumentMap transformArguments(ClientArgumentMap args) {
+        return args;
+    }
+
+    @Override
+    protected List<String> transformExecutableCommand(String executableCommand) {
+        return Arrays.asList("python", executableCommand, "receiver");
+    }
+}
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/messagingclients/stomp/StompClientSender.java b/systemtests/src/main/java/io/enmasse/systemtest/messagingclients/stomp/StompClientSender.java
new file mode 100644
index 000000000..dd916884e
--- /dev/null
+++ b/systemtests/src/main/java/io/enmasse/systemtest/messagingclients/stomp/StompClientSender.java
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2019, EnMasse authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.enmasse.systemtest.messagingclients.stomp;
+
+import io.enmasse.systemtest.messagingclients.AbstractClient;
+import io.enmasse.systemtest.messagingclients.ClientArgument;
+import io.enmasse.systemtest.messagingclients.ClientArgumentMap;
+import io.enmasse.systemtest.messagingclients.ClientType;
+
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.List;
+
+
+public class StompClientSender extends AbstractClient {
+    public StompClientSender() throws Exception {
+        super(ClientType.CLI_STOMP_SENDER);
+    }
+
+    public StompClientSender(Path logPath) throws Exception {
+        super(ClientType.CLI_STOMP_SENDER, logPath);
+    }
+
+    @Override
+    protected void fillAllowedArgs() {
+        allowedArgs.add(ClientArgument.BROKER);
+        allowedArgs.add(ClientArgument.ADDRESS);
+        allowedArgs.add(ClientArgument.COUNT);
+        allowedArgs.add(ClientArgument.USERNAME);
+        allowedArgs.add(ClientArgument.PASSWORD);
+        allowedArgs.add(ClientArgument.MSG_CONTENT);
+        allowedArgs.add(ClientArgument.DEST_TYPE);
+    }
+
+    @Override
+    protected ClientArgumentMap transformArguments(ClientArgumentMap args) {
+        return args;
+    }
+
+    @Override
+    protected List<String> transformExecutableCommand(String executableCommand) {
+        return Arrays.asList("python", executableCommand, "sender");
+    }
+}
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/mqtt/MqttClientFactory.java b/systemtests/src/main/java/io/enmasse/systemtest/mqtt/MqttClientFactory.java
index 0d78912bb..b09ebc931 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/mqtt/MqttClientFactory.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/mqtt/MqttClientFactory.java
@@ -120,6 +120,7 @@ public class MqttClientFactory {
                 if (mqttConnectOptions == null) {
                     mqttConnectOptions = new MqttConnectOptions();
                     mqttConnectOptions.setAutomaticReconnect(true);
+                    mqttConnectOptions.setHttpsHostnameVerificationEnabled(false);
                 }
                 return MqttClientFactory.this.create(MqttClient::new, DelegatingMqttClient::new, endpoint, addressSpace, mqttConnectOptions, clientId);
             }
@@ -387,6 +388,14 @@ public class MqttClientFactory {
             return mqttClient.publish(topic, message, userContext, callback);
         }
 
+        public void reconnect() throws MqttException {
+            mqttClient.reconnect();
+        }
+
+        public boolean removeMessage(IMqttDeliveryToken token) throws MqttException {
+            return mqttClient.removeMessage(token);
+        }
+
         public IMqttToken subscribe(String topicFilter, int qos) throws MqttException {
             return mqttClient.subscribe(topicFilter, qos);
         }
@@ -504,6 +513,11 @@ public class MqttClientFactory {
             this.mqttClient.disconnectForcibly(quiesceTimeout, disconnectTimeout);
         }
 
+        @Override
+        public void reconnect() throws MqttException {
+            this.mqttClient.reconnect();
+        }
+
         @Override
         public void subscribe(String topicFilter) throws MqttException {
             this.mqttClient.subscribe(topicFilter);
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/platform/Kubernetes.java b/systemtests/src/main/java/io/enmasse/systemtest/platform/Kubernetes.java
index 6be62f34f..6d14a1b31 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/platform/Kubernetes.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/platform/Kubernetes.java
@@ -366,8 +366,8 @@ public abstract class Kubernetes {
         return logs;
     }
 
-    public void setDeploymentReplicas(String name, int numReplicas) {
-        client.apps().deployments().inNamespace(infraNamespace).withName(name).scale(numReplicas, true);
+    public void setDeploymentReplicas(String namespace, String name, int numReplicas) {
+        client.apps().deployments().inNamespace(namespace).withName(name).scale(numReplicas, true);
     }
 
     public void setStatefulSetReplicas(String name, int numReplicas) {
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/platform/apps/SystemtestsKubernetesApps.java b/systemtests/src/main/java/io/enmasse/systemtest/platform/apps/SystemtestsKubernetesApps.java
index f35408935..d07dc6069 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/platform/apps/SystemtestsKubernetesApps.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/platform/apps/SystemtestsKubernetesApps.java
@@ -268,6 +268,18 @@ public class SystemtestsKubernetesApps {
         Thread.sleep(5000);
     }
 
+    public static void scaleDownAMQBroker(String namespace) throws Exception {
+        Kubernetes kubeCli = Kubernetes.getInstance();
+        kubeCli.setDeploymentReplicas(namespace, AMQ_BROKER, 0);
+        TestUtils.waitForExpectedReadyPods(kubeCli, namespace, 0, new TimeoutBudget(30, TimeUnit.SECONDS));
+    }
+
+    public static void scaleUpAMQBroker(String namespace) throws Exception {
+        Kubernetes kubeCli = Kubernetes.getInstance();
+        kubeCli.setDeploymentReplicas(namespace, AMQ_BROKER, 1);
+        TestUtils.waitForExpectedReadyPods(kubeCli, namespace, 1, new TimeoutBudget(30, TimeUnit.SECONDS));
+    }
+
     public static void applyDirectories(final Function<InputStream, InputStream> streamManipulator, final Path... paths) throws Exception {
         loadDirectories(streamManipulator, Applicable::createOrReplace, paths);
     }
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/utils/AddressSpaceUtils.java b/systemtests/src/main/java/io/enmasse/systemtest/utils/AddressSpaceUtils.java
index c3b658b08..5e036edb3 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/utils/AddressSpaceUtils.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/utils/AddressSpaceUtils.java
@@ -230,6 +230,17 @@ public class AddressSpaceUtils {
         return addressSpace.getSpec().getType().equals(AddressSpaceType.BROKERED.toString());
     }
 
+    public static void waitForAddressSpaceConnectorsNotReady(AddressSpace addressSpace) throws Exception {
+        TestUtils.waitUntilCondition("Connectors report not ready", phase -> {
+            try {
+                AddressSpaceUtils.waitForAddressSpaceConnectorsReady(addressSpace, new TimeoutBudget(20, TimeUnit.SECONDS));
+                return false;
+            } catch (Exception ex) {
+                return ex instanceof IllegalStateException;
+            }
+        }, new TimeoutBudget(5, TimeUnit.MINUTES));
+    }
+
     public static AddressSpace waitForAddressSpaceConnectorsReady(AddressSpace addressSpace) throws Exception {
         return waitForAddressSpaceConnectorsReady(addressSpace, new TimeoutBudget(5, TimeUnit.MINUTES));
     }
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/utils/TestUtils.java b/systemtests/src/main/java/io/enmasse/systemtest/utils/TestUtils.java
index b397089f3..b0c410f7c 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/utils/TestUtils.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/utils/TestUtils.java
@@ -63,7 +63,7 @@ public class TestUtils {
      * scale up/down specific pod (type: Deployment) in address space
      */
     public static void setReplicas(Kubernetes kubernetes, String infraUuid, String deployment, int numReplicas, TimeoutBudget budget) throws InterruptedException {
-        kubernetes.setDeploymentReplicas(deployment, numReplicas);
+        kubernetes.setDeploymentReplicas(kubernetes.getInfraNamespace(), deployment, numReplicas);
         Map<String, String> labels = new HashMap<>();
         labels.put("name", deployment);
         if (infraUuid != null) {
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/bases/bridging/BridgingBase.java b/systemtests/src/test/java/io/enmasse/systemtest/bases/bridging/BridgingBase.java
index d41bf244e..97882813f 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/bases/bridging/BridgingBase.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/bases/bridging/BridgingBase.java
@@ -35,10 +35,10 @@ public abstract class BridgingBase extends TestBase implements ITestIsolatedStan
 
     protected static final String REMOTE_NAME = "remote1";
 
-    private final String remoteBrokerNamespace = "systemtests-external-broker";
-    private final String remoteBrokerUsername = "test-user";
-    private final String remoteBrokerPassword = "test-password";
-    private Endpoint remoteBrokerEndpoint;
+    protected final String remoteBrokerNamespace = "systemtests-external-broker";
+    protected final String remoteBrokerUsername = "test-user";
+    protected final String remoteBrokerPassword = "test-password";
+    protected Endpoint remoteBrokerEndpoint;
 
     protected abstract String[] remoteBrokerQueues();
 
@@ -58,6 +58,16 @@ public abstract class BridgingBase extends TestBase implements ITestIsolatedStan
         SystemtestsKubernetesApps.deleteAMQBroker(remoteBrokerNamespace);
     }
 
+    protected void scaleDownBroker() throws Exception {
+        log.info("Scaling down broker");
+        SystemtestsKubernetesApps.scaleDownAMQBroker(remoteBrokerNamespace);
+    }
+
+    protected void scaleUpBroker() throws Exception {
+        log.info("Scaling up broker");
+        SystemtestsKubernetesApps.scaleUpAMQBroker(remoteBrokerNamespace);
+    }
+
     protected AddressSpace createAddressSpace(String name, String addressRule) throws Exception {
         AddressSpace space = new AddressSpaceBuilder()
                 .withNewMetadata()
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/bases/clients/ClientTestBase.java b/systemtests/src/test/java/io/enmasse/systemtest/bases/clients/ClientTestBase.java
index ec92db580..0b82b9bea 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/bases/clients/ClientTestBase.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/bases/clients/ClientTestBase.java
@@ -8,6 +8,7 @@ import io.enmasse.address.model.Address;
 import io.enmasse.address.model.AddressBuilder;
 import io.enmasse.address.model.AddressSpace;
 import io.enmasse.systemtest.Endpoint;
+import io.enmasse.systemtest.UserCredentials;
 import io.enmasse.systemtest.bases.TestBase;
 import io.enmasse.systemtest.bases.shared.ITestBaseShared;
 import io.enmasse.systemtest.broker.ArtemisManagement;
@@ -21,6 +22,10 @@ import io.enmasse.systemtest.model.addressspace.AddressSpaceType;
 import io.enmasse.systemtest.utils.AddressSpaceUtils;
 import io.enmasse.systemtest.utils.AddressUtils;
 import io.enmasse.systemtest.utils.TestUtils;
+import io.enmasse.systemtest.utils.UserUtils;
+import io.enmasse.user.model.v1.Operation;
+import io.enmasse.user.model.v1.User;
+import io.enmasse.user.model.v1.UserAuthorizationBuilder;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.TestInfo;
@@ -30,11 +35,13 @@ import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.Future;
 
 import static org.junit.jupiter.api.Assertions.assertAll;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @ExternalClients
@@ -43,6 +50,10 @@ public abstract class ClientTestBase extends TestBase implements ITestBaseShared
     private ClientArgumentMap arguments = new ClientArgumentMap();
     private List<AbstractClient> clients;
 
+    public ClientArgumentMap getArguments() {
+        return arguments;
+    }
+
     @BeforeEach
     public void setUpClientBase(TestInfo info) {
         clients = new ArrayList<>();
@@ -459,4 +470,84 @@ public abstract class ClientTestBase extends TestBase implements ITestBaseShared
         assertEquals(expectedMsgCount, subscriber2.getMessages().size(),
                 String.format("Expected %d received messages 'number < 12.5'", expectedMsgCount));
     }
+
+    protected void doTestUserPermissions(AbstractClient sender, AbstractClient receiver) throws Exception {
+        int expectedMsgCount = 5;
+        UserCredentials publishCred = new UserCredentials("publisher", "publish");
+        UserCredentials consumCred = new UserCredentials("consumer", "consume");
+        createPublisherAndConsumer(publishCred, consumCred);
+
+        Address dest = new AddressBuilder()
+                .withNewMetadata()
+                .withNamespace(getSharedAddressSpace().getMetadata().getNamespace())
+                .withName(AddressUtils.generateAddressMetadataName(getSharedAddressSpace(), "message-basic"))
+                .endMetadata()
+                .withNewSpec()
+                .withType("queue")
+                .withAddress("message-basic" + ClientType.getAddressName(sender))
+                .withPlan(getDefaultPlan(AddressType.QUEUE))
+                .endSpec()
+                .build();
+        resourcesManager.setAddresses(dest);
+
+        arguments.put(ClientArgument.BROKER, getMessagingRoute(getSharedAddressSpace(), false).toString());
+        arguments.put(ClientArgument.ADDRESS, dest.getSpec().getAddress());
+        arguments.put(ClientArgument.COUNT, Integer.toString(expectedMsgCount));
+        arguments.put(ClientArgument.MSG_CONTENT, "msg no. %d");
+        arguments.put(ClientArgument.TIMEOUT, "30");
+        replaceCredArgs(consumCred);
+        sender.setArguments(arguments);
+        arguments.remove(ClientArgument.MSG_CONTENT);
+        replaceCredArgs(publishCred);
+        receiver.setArguments(arguments);
+
+        assertAll(
+                () -> assertFalse(sender.run(), "Sender failed. Specified user is not allowed to write"),
+                () -> assertFalse(receiver.run(), "Receiver failed. Specified user is not allowed to read"));
+        replaceCredArgs(publishCred);
+        arguments.put(ClientArgument.MSG_CONTENT, "msg no. %d");
+        sender.setArguments(arguments);
+        arguments.remove(ClientArgument.MSG_CONTENT);
+        replaceCredArgs(consumCred);
+        receiver.setArguments(arguments);
+
+        assertTrue(sender.run(), "Sender failed, expected return code 0");
+        assertTrue(receiver.run(), "Receiver failed, expected return code 0");
+
+        assertEquals(expectedMsgCount, sender.getMessages().size(),
+                String.format("Expected %d sent messages", expectedMsgCount));
+        assertEquals(expectedMsgCount, receiver.getMessages().size(),
+                String.format("Expected %d received messages", expectedMsgCount));
+    }
+
+    private void createPublisherAndConsumer(UserCredentials publishCred, UserCredentials consumCred) {
+        User publisher = (UserUtils.createUserResource(publishCred)
+                .editSpec()
+                .withAuthorization(Collections.singletonList(new UserAuthorizationBuilder()
+                        .withAddresses("*")
+                        .withOperations(Operation.send)
+                        .build()))
+                .endSpec()
+                .done());
+
+        User consumer = (UserUtils.createUserResource(consumCred)
+                .editSpec()
+                .withAuthorization(Collections.singletonList(new UserAuthorizationBuilder()
+                        .withAddresses("*")
+                        .withOperations(Operation.recv)
+                        .build()))
+                .endSpec()
+                .done());
+
+        resourcesManager.createOrUpdateUser(getSharedAddressSpace(), publisher);
+        resourcesManager.createOrUpdateUser(getSharedAddressSpace(), consumer);
+    }
+
+    private void replaceCredArgs(UserCredentials userCredentials) {
+        arguments.remove(ClientArgument.USERNAME);
+        arguments.remove(ClientArgument.PASSWORD);
+        arguments.put(ClientArgument.USERNAME, userCredentials.getUsername());
+        arguments.put(ClientArgument.PASSWORD, userCredentials.getPassword());
+    }
+
 }
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/isolated/api/ApiServerTest.java b/systemtests/src/test/java/io/enmasse/systemtest/isolated/api/ApiServerTest.java
index 68e6c6d5f..deb58d837 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/isolated/api/ApiServerTest.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/isolated/api/ApiServerTest.java
@@ -18,21 +18,28 @@ import io.enmasse.admin.model.v1.AddressPlan;
 import io.enmasse.admin.model.v1.AddressSpacePlan;
 import io.enmasse.admin.model.v1.ResourceAllowance;
 import io.enmasse.admin.model.v1.ResourceRequest;
+import io.enmasse.systemtest.Environment;
 import io.enmasse.systemtest.UserCredentials;
 import io.enmasse.systemtest.bases.TestBase;
 import io.enmasse.systemtest.bases.isolated.ITestIsolatedStandard;
 import io.enmasse.systemtest.cmdclients.KubeCMDClient;
+import io.enmasse.systemtest.executor.ExecutionResultData;
+import io.enmasse.systemtest.isolated.Credentials;
 import io.enmasse.systemtest.logs.CustomLogger;
 import io.enmasse.systemtest.model.address.AddressType;
 import io.enmasse.systemtest.model.addressplan.DestinationPlan;
 import io.enmasse.systemtest.model.addressspace.AddressSpacePlans;
 import io.enmasse.systemtest.model.addressspace.AddressSpaceType;
 import io.enmasse.systemtest.time.TimeoutBudget;
+import io.enmasse.systemtest.utils.AddressSpaceUtils;
 import io.enmasse.systemtest.utils.AddressUtils;
 import io.enmasse.systemtest.utils.PlanUtils;
 import io.enmasse.user.model.v1.UserAuthenticationType;
+import io.fabric8.kubernetes.api.model.rbac.ClusterRoleBindingBuilder;
+import io.fabric8.kubernetes.api.model.rbac.SubjectBuilder;
 import io.fabric8.kubernetes.client.KubernetesClientException;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 import org.slf4j.Logger;
 
 import java.util.Arrays;
@@ -45,6 +52,7 @@ import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 import static org.hamcrest.CoreMatchers.is;
@@ -52,10 +60,13 @@ import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import io.vertx.core.json.JsonObject;
+
 public class ApiServerTest extends TestBase implements ITestIsolatedStandard {
     private static Logger log = CustomLogger.getLogger();
 
@@ -481,4 +492,74 @@ public class ApiServerTest extends TestBase implements ITestIsolatedStandard {
         Exception exception = assertThrows(KubernetesClientException.class, () -> commonResourcesManager.replaceAddressSpace(replace));
         assertTrue(exception.getMessage().contains("Unknown address space plan no-exists"));
     }
+
+
+    @Test
+    @DisabledIfEnvironmentVariable(named = Environment.USE_MINUKUBE_ENV, matches = "true")
+    public void testCreateAddressSpaceRoleTenantEdit() throws Exception {
+
+        UserCredentials user = Credentials.userCredentials();
+        String namespace = "test-authorization";
+        String rolebindingname = "testgroupbinding";
+        String groupName = "foogroup";
+        try {
+            KubeCMDClient.loginUser(environment.getApiToken());
+
+            KubeCMDClient.createNamespace(namespace);
+
+            KubeCMDClient.createGroupAndAddUser(groupName, user.getUsername());
+
+            kubernetes.getClient().rbac().clusterRoleBindings().create(new ClusterRoleBindingBuilder()
+                    .withNewMetadata()
+                    .withName(rolebindingname)
+                    .endMetadata()
+                    .withNewRoleRef()
+                    .withKind("ClusterRole")
+                    .withApiGroup("rbac.authorization.k8s.io")
+                    .withName("enmasse.io:tenant-edit")
+                    .endRoleRef()
+                    .withSubjects(new SubjectBuilder()
+                            .withKind("Group")
+                            .withApiGroup("rbac.authorization.k8s.io")
+                            .withName(groupName)
+                            .withNamespace(namespace)
+                            .build())
+                    .build());
+
+            Supplier<AddressSpace> supplier = () -> new AddressSpaceBuilder()
+                    .withNewMetadata()
+                    .withName("test-authorization-addressspace")
+                    .withNamespace(namespace)
+                    .endMetadata()
+                    .withNewSpec()
+                    .withType(AddressSpaceType.BROKERED.toString())
+                    .withPlan(AddressSpacePlans.BROKERED)
+                    .endSpec()
+                    .build();
+
+            KubeCMDClient.loginUser(user.getUsername(), user.getPassword());
+
+            AddressSpace space1 = supplier.get();
+            JsonObject space1Json = AddressSpaceUtils.addressSpaceToJson(space1);
+            assertTrue(KubeCMDClient.createCR(namespace, space1Json.toString()).getRetCode());
+            resourcesManager.deleteAddressSpace(space1);
+
+            kubernetes.getClient().rbac().clusterRoleBindings().withName(rolebindingname).delete();
+
+            AddressSpace space2 = supplier.get();
+            JsonObject space2Json = AddressSpaceUtils.addressSpaceToJson(space2);
+            ExecutionResultData space2Result = KubeCMDClient.createCR(namespace, space2Json.toString());
+            assertFalse(space2Result.getRetCode());
+            assertTrue(space2Result.getStdErr().contains("User \""+user.getUsername()+"\" cannot get addressspaces.enmasse.io in the namespace \""+namespace+"\": no RBAC policy matched"));
+
+        } finally {
+            if (kubernetes.getClient().rbac().clusterRoleBindings().withName(rolebindingname).get() != null) {
+                kubernetes.getClient().rbac().clusterRoleBindings().withName(rolebindingname).delete();
+            }
+            KubeCMDClient.loginUser(environment.getApiToken());
+            KubeCMDClient.switchProject(environment.namespace());
+            kubernetes.deleteNamespace(namespace);
+        }
+
+    }
 }
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/isolated/api/AuthServiceTest.java b/systemtests/src/test/java/io/enmasse/systemtest/isolated/api/AuthServiceTest.java
index deeca2821..c419d13a2 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/isolated/api/AuthServiceTest.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/isolated/api/AuthServiceTest.java
@@ -126,7 +126,8 @@ class AuthServiceTest extends TestBase implements ITestIsolatedStandard {
                 "myrealm",
                 invalidCert,
                 invalidCert);
-        resourcesManager.createAuthService(externalAuth);
+        // Can't wait for it because it doesn't actually spin up any pod
+        resourcesManager.createAuthService(externalAuth, false);
         log.info(externalAuth.toString());
 
         SecretReference validCert = new SecretReference();
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/isolated/bridging/ConnectorsTest.java b/systemtests/src/test/java/io/enmasse/systemtest/isolated/bridging/ConnectorsTest.java
index 4971b069f..883c5c181 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/isolated/bridging/ConnectorsTest.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/isolated/bridging/ConnectorsTest.java
@@ -6,11 +6,12 @@ package io.enmasse.systemtest.isolated.bridging;
 
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import io.enmasse.address.model.Address;
@@ -30,6 +31,7 @@ import io.enmasse.systemtest.shared.standard.QueueTest;
 import io.enmasse.systemtest.time.TimeoutBudget;
 import io.enmasse.systemtest.utils.AddressSpaceUtils;
 import io.enmasse.systemtest.utils.TestUtils;
+import io.fabric8.kubernetes.client.KubernetesClientException;
 
 class ConnectorsTest extends BridgingBase {
 
@@ -94,7 +96,7 @@ class ConnectorsTest extends BridgingBase {
     }
 
     @Test
-    void testBadConfiguration() throws Exception {
+    void testNonExsistingHost() throws Exception {
         AddressSpace space = new AddressSpaceBuilder()
                 .withNewMetadata()
                 .withNamespace(kubernetes.getInfraNamespace())
@@ -130,6 +132,108 @@ class ConnectorsTest extends BridgingBase {
         });
     }
 
+    @Test
+    void testInvalidConnectorName() throws Exception {
+        AddressSpace space = new AddressSpaceBuilder()
+                .withNewMetadata()
+                .withNamespace(kubernetes.getInfraNamespace())
+                .withName("invalid-connector-name")
+                .endMetadata()
+                .withNewSpec()
+                .withType(AddressSpaceType.STANDARD.toString())
+                .withPlan(AddressSpacePlans.STANDARD_SMALL)
+                .withConnectors(new AddressSpaceSpecConnectorBuilder()
+                        .withName("/ect/dhcp")
+                        .addToEndpointHosts(new AddressSpaceSpecConnectorEndpointBuilder()
+                                .withHost(remoteBrokerEndpoint.getHost())
+                                .withPort(remoteBrokerEndpoint.getPort())
+                                .build())
+                        .withCredentials(new AddressSpaceSpecConnectorCredentialsBuilder()
+                                .withNewUsername()
+                                    .withValue(remoteBrokerUsername)
+                                    .endUsername()
+                                .withNewPassword()
+                                    .withValue(remoteBrokerPassword)
+                                    .endPassword()
+                                .build())
+                        .addToAddresses(new AddressSpaceSpecConnectorAddressRuleBuilder()
+                                .withName("queuesrule")
+                                .withPattern("*")
+                                .build())
+                        .build())
+                .endSpec()
+                .build();
+        try {
+            resourcesManager.createAddressSpace(space);
+            Assertions.fail();
+        } catch (KubernetesClientException e) {
+            assertEquals(400, e.getCode());
+            assertTrue(e.getMessage().contains("Invalid address space connector name"));
+        }
+    }
+
+    @Test
+    void testInvalidAddressRulePattern() throws Exception {
+        AddressSpace space = new AddressSpaceBuilder()
+                .withNewMetadata()
+                .withNamespace(kubernetes.getInfraNamespace())
+                .withName("invalid-connector-name")
+                .endMetadata()
+                .withNewSpec()
+                .withType(AddressSpaceType.STANDARD.toString())
+                .withPlan(AddressSpacePlans.STANDARD_SMALL)
+                .withConnectors(new AddressSpaceSpecConnectorBuilder()
+                        .withName(REMOTE_NAME)
+                        .addToEndpointHosts(new AddressSpaceSpecConnectorEndpointBuilder()
+                                .withHost(remoteBrokerEndpoint.getHost())
+                                .withPort(remoteBrokerEndpoint.getPort())
+                                .build())
+                        .withCredentials(new AddressSpaceSpecConnectorCredentialsBuilder()
+                                .withNewUsername()
+                                    .withValue(remoteBrokerUsername)
+                                    .endUsername()
+                                .withNewPassword()
+                                    .withValue(remoteBrokerPassword)
+                                    .endPassword()
+                                .build())
+                        .addToAddresses(new AddressSpaceSpecConnectorAddressRuleBuilder()
+                                .withName("queuesrule")
+                                .withPattern("queue*")
+                                .build())
+                        .build())
+                .endSpec()
+                .build();
+        try {
+            resourcesManager.createAddressSpace(space);
+            Assertions.fail();
+        } catch (KubernetesClientException e) {
+            assertEquals(400, e.getCode());
+            assertTrue(e.getMessage().contains("Invalid address space connector address rule pattern"));
+        }
+
+    }
+
+    @Test
+    void testRestartBroker() throws Exception {
+        AddressSpace space = createAddressSpace("restart-broker", BASIC_QUEUES_PATTERN);
+
+        UserCredentials localUser = new UserCredentials("test", "test");
+        resourcesManager.createOrUpdateUser(space, localUser);
+
+        int messagesBatch = 50;
+        String[] remoteQueues = new String [] {BASIC_QUEUE1};
+        sendToConnectorReceiveInBroker(space, localUser, remoteQueues, messagesBatch);
+
+        scaleDownBroker();
+        AddressSpaceUtils.waitForAddressSpaceConnectorsNotReady(space);
+
+        scaleUpBroker();
+        AddressSpaceUtils.waitForAddressSpaceConnectorsReady(space);
+
+        sendToConnectorReceiveInBroker(space, localUser, remoteQueues, messagesBatch);
+        sendToBrokerReceiveInConnector(space, localUser, remoteQueues, messagesBatch);
+    }
+
     private void doTestSendThroughConnector(String addressRule, String[] remoteQueues) throws Exception, InterruptedException, ExecutionException, TimeoutException {
         AddressSpace space = createAddressSpace("send-to-connector", addressRule);
 
@@ -138,8 +242,22 @@ class ConnectorsTest extends BridgingBase {
 
         int messagesBatch = 50;
 
-        //send through connector
+        sendToConnectorReceiveInBroker(space, localUser, remoteQueues, messagesBatch);
+    }
 
+    private void doTestReceiveThroughConnector(String addressRule, String[] remoteQueues) throws Exception {
+        AddressSpace space = createAddressSpace("receive-from-connector", addressRule);
+
+        UserCredentials localUser = new UserCredentials("test", "test");
+        resourcesManager.createOrUpdateUser(space, localUser);
+
+        int messagesBatch = 50;
+
+        sendToBrokerReceiveInConnector(space, localUser, remoteQueues, messagesBatch);
+    }
+
+    private void sendToConnectorReceiveInBroker(AddressSpace space, UserCredentials localUser, String[] remoteQueues, int messagesBatch) throws Exception {
+        //send through connector
         AmqpClient localClient = getAmqpClientFactory().createQueueClient(space);
         localClient.getConnectOptions().setCredentials(localUser);
 
@@ -149,26 +267,16 @@ class ConnectorsTest extends BridgingBase {
         }
 
         //receive in remote broker
-
         AmqpClient clientToRemote = createClientToRemoteBroker();
 
         for(String remoteQueue : remoteQueues) {
             var receivedFromQueue = clientToRemote.recvMessages(remoteQueue, messagesBatch);
             assertThat("Wrong count of messages received from queue: "+remoteQueue, receivedFromQueue.get(1, TimeUnit.MINUTES).size(), is(messagesBatch));
         }
-
     }
 
-    private void doTestReceiveThroughConnector(String addressRule, String[] remoteQueues) throws Exception, InterruptedException, ExecutionException, TimeoutException {
-        AddressSpace space = createAddressSpace("receive-from-connector", addressRule);
-
-        UserCredentials localUser = new UserCredentials("test", "test");
-        resourcesManager.createOrUpdateUser(space, localUser);
-
-        int messagesBatch = 50;
-
+    private void sendToBrokerReceiveInConnector(AddressSpace space, UserCredentials localUser, String[] remoteQueues, int messagesBatch) throws Exception {
         //send to remote broker
-
         AmqpClient clientToRemote = createClientToRemoteBroker();
 
         for(String remoteQueue : remoteQueues) {
@@ -176,7 +284,6 @@ class ConnectorsTest extends BridgingBase {
         }
 
         //receive through connector
-
         AmqpClient localClient = getAmqpClientFactory().createQueueClient(space);
         localClient.getConnectOptions().setCredentials(localUser);
 
@@ -185,7 +292,6 @@ class ConnectorsTest extends BridgingBase {
             var receivedFromQueue = localClient.recvMessages(connectorQueue, messagesBatch);
             assertThat("Wrong count of messages received from connector queue: "+connectorQueue, receivedFromQueue.get(1, TimeUnit.MINUTES).size(), is(messagesBatch));
         }
-
     }
 
     private String getRemoteName(String remoteQueue) {
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/isolated/bridging/ForwardersTest.java b/systemtests/src/test/java/io/enmasse/systemtest/isolated/bridging/ForwardersTest.java
index 29a38579a..460841965 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/isolated/bridging/ForwardersTest.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/isolated/bridging/ForwardersTest.java
@@ -21,6 +21,7 @@ import io.enmasse.systemtest.bases.bridging.BridgingBase;
 import io.enmasse.systemtest.model.address.AddressType;
 import io.enmasse.systemtest.model.addressplan.DestinationPlan;
 import io.enmasse.systemtest.time.TimeoutBudget;
+import io.enmasse.systemtest.utils.AddressSpaceUtils;
 import io.enmasse.systemtest.utils.AddressUtils;
 import io.enmasse.systemtest.utils.TestUtils;
 
@@ -68,20 +69,7 @@ public class ForwardersTest extends BridgingBase {
 
         int messagesBatch = 50;
 
-        //send to address with forwarder
-
-        AmqpClient localClient = getAmqpClientFactory().createQueueClient(space);
-        localClient.getConnectOptions().setCredentials(localUser);
-
-        localClient.sendMessages(forwarder.getSpec().getAddress(), TestUtils.generateMessages(messagesBatch));
-
-        //receive in remote broker
-
-        AmqpClient clientToRemote = createClientToRemoteBroker();
-
-        var receivedInRemote = clientToRemote.recvMessages(REMOTE_QUEUE1, messagesBatch);
-
-        assertThat("Wrong count of messages received from remote queue: "+REMOTE_QUEUE1, receivedInRemote.get(1, TimeUnit.MINUTES).size(), is(messagesBatch));
+        doTestSendToForwarder(space, forwarder, localUser, REMOTE_QUEUE1, messagesBatch);
 
     }
 
@@ -129,4 +117,88 @@ public class ForwardersTest extends BridgingBase {
 
     }
 
+    @Test
+    public void testForwardToUnavailableBroker() throws Exception {
+
+        AddressSpace space = createAddressSpace("forward-to-remote", "*");
+        Address forwarder = new AddressBuilder()
+                .withNewMetadata()
+                .withName(AddressUtils.generateAddressMetadataName(space, "forwarder-queue1"))
+                .withNamespace(kubernetes.getInfraNamespace())
+                .endMetadata()
+                .withNewSpec()
+                .withAddress("forwarder-queue1")
+                .withType(AddressType.QUEUE.toString())
+                .withPlan(DestinationPlan.STANDARD_SMALL_QUEUE)
+                .addToForwarders(new AddressSpecForwarderBuilder()
+                        .withName("forwarder1")
+                        .withRemoteAddress(REMOTE_NAME + "/" + REMOTE_QUEUE1)
+                        .withDirection(AddressSpecForwarderDirection.out)
+                        .build())
+                .endSpec()
+                .build();
+        resourcesManager.setAddresses(forwarder);
+        AddressUtils.waitForForwardersReady(new TimeoutBudget(1, TimeUnit.MINUTES), forwarder);
+
+        UserCredentials localUser = new UserCredentials("test", "test");
+        resourcesManager.createOrUpdateUser(space, localUser);
+
+        doTestSendToForwarder(space, forwarder, localUser, REMOTE_QUEUE1, 5);
+
+        //make broker unavailable
+        scaleDownBroker();
+
+        //check connector and address forwarder is not ready
+        AddressSpaceUtils.waitForAddressSpaceConnectorsNotReady(space);
+        TestUtils.waitUntilCondition("Forwarders not ready", phase -> {
+            try {
+                AddressUtils.waitForForwardersReady(new TimeoutBudget(20, TimeUnit.SECONDS), forwarder);
+                return false;
+            } catch (Exception ex) {
+                return ex instanceof IllegalStateException;
+            }
+        }, new TimeoutBudget(3, TimeUnit.MINUTES));
+        //however address should be still ready
+        AddressUtils.waitForDestinationsReady(new TimeoutBudget(30, TimeUnit.SECONDS), forwarder);
+
+        //send to forwarder
+        int messagesBatch = 50;
+        AmqpClient localClient = getAmqpClientFactory().createQueueClient(space);
+        localClient.getConnectOptions().setCredentials(localUser);
+
+        localClient.sendMessages(forwarder.getSpec().getAddress(), TestUtils.generateMessages(messagesBatch));
+
+        //wake up the broker
+        scaleUpBroker();
+
+        //wait until forwarder is ready again
+        AddressUtils.waitForDestinationsReady(new TimeoutBudget(30, TimeUnit.SECONDS), forwarder);
+        AddressSpaceUtils.waitForAddressSpaceConnectorsReady(space);
+        AddressUtils.waitForForwardersReady(new TimeoutBudget(1, TimeUnit.MINUTES), forwarder);
+
+        //check messages where automatically forwarded once broker is back up again
+        AmqpClient clientToRemote = createClientToRemoteBroker();
+
+        var receivedInRemote = clientToRemote.recvMessages(REMOTE_QUEUE1, messagesBatch);
+
+        assertThat("Wrong count of messages received from remote queue: "+REMOTE_QUEUE1, receivedInRemote.get(1, TimeUnit.MINUTES).size(), is(messagesBatch));
+    }
+
+    private void doTestSendToForwarder(AddressSpace space, Address forwarder, UserCredentials localUser, String rometeAddress, int messagesBatch) throws Exception {
+        //send to address with forwarder
+
+        AmqpClient localClient = getAmqpClientFactory().createQueueClient(space);
+        localClient.getConnectOptions().setCredentials(localUser);
+
+        localClient.sendMessages(forwarder.getSpec().getAddress(), TestUtils.generateMessages(messagesBatch));
+
+        //receive in remote broker
+
+        AmqpClient clientToRemote = createClientToRemoteBroker();
+
+        var receivedInRemote = clientToRemote.recvMessages(rometeAddress, messagesBatch);
+
+        assertThat("Wrong count of messages received from remote queue: "+rometeAddress, receivedInRemote.get(1, TimeUnit.MINUTES).size(), is(messagesBatch));
+    }
+
 }
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/isolated/certs/CertProviderTest.java b/systemtests/src/test/java/io/enmasse/systemtest/isolated/certs/CertProviderTest.java
index c43a89183..6bdd99ec5 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/isolated/certs/CertProviderTest.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/isolated/certs/CertProviderTest.java
@@ -251,12 +251,18 @@ class CertProviderTest extends TestBase implements ITestIsolatedStandard {
 
                 TimeoutBudget timeout = new TimeoutBudget(5, TimeUnit.MINUTES);
                 Exception lastException = null;
+                int maxRetries = 10;
+                int retry = 0;
                 while (!timeout.timeoutExpired()) {
                     try {
                         log.info("Making request to openshift-cert-validator {}", request);
                         JsonObject response = client.test(request);
                         if (response.containsKey("error")) {
-                            fail("Error testing openshift provider " + response.getString("error"));
+                            if (retry < maxRetries) {
+                                retry++;
+                            } else {
+                                fail("Error testing openshift provider " + response.getString("error"));
+                            }
                         } else {
                             testSucceeded = true;
                             return;
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/shared/brokered/clients/stomp/MsgPatternsTest.java b/systemtests/src/test/java/io/enmasse/systemtest/shared/brokered/clients/stomp/MsgPatternsTest.java
new file mode 100644
index 000000000..9edd43a8d
--- /dev/null
+++ b/systemtests/src/test/java/io/enmasse/systemtest/shared/brokered/clients/stomp/MsgPatternsTest.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2019, EnMasse authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.enmasse.systemtest.shared.brokered.clients.stomp;
+
+import io.enmasse.systemtest.bases.clients.ClientTestBase;
+import io.enmasse.systemtest.bases.shared.ITestSharedBrokered;
+import io.enmasse.systemtest.broker.ArtemisManagement;
+import io.enmasse.systemtest.messagingclients.ClientArgument;
+import io.enmasse.systemtest.messagingclients.stomp.StompClientReceiver;
+import io.enmasse.systemtest.messagingclients.stomp.StompClientSender;
+import io.enmasse.systemtest.resolvers.ArtemisManagementParameterResolver;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(ArtemisManagementParameterResolver.class)
+class MsgPatternsTest extends ClientTestBase implements ITestSharedBrokered {
+
+    @Test
+    void testBasicMessage() throws Exception {
+        getArguments().put(ClientArgument.DEST_TYPE, "ANYCAST");
+        doBasicMessageTest(new StompClientSender(logPath), new StompClientReceiver(logPath));
+    }
+
+    @Test
+    void testTopicSubscribe(ArtemisManagement artemisManagement) throws Exception {
+        getArguments().put(ClientArgument.DEST_TYPE, "MULTICAST");
+        doTopicSubscribeTest(artemisManagement, new StompClientSender(logPath), new StompClientReceiver(logPath), new StompClientReceiver(logPath), false);
+    }
+
+    @Test
+    void testStompUserPermissions() throws Exception {
+        getArguments().put(ClientArgument.DEST_TYPE, "ANYCAST");
+        doTestUserPermissions(new StompClientSender(logPath), new StompClientReceiver(logPath));
+    }
+}