Repository: EnMasseProject/enmasse
Pull request #3178 (2019-09-11 12:04:02)

Add support for address forwarding to standard address space
* Doc: Update design doc
* Model: Add classes for forwarder spec and status
* Broker AMQP Connector: Add support for creating links in both directions
* Agent: Add support for creating broker -> router connectors + links based on forwarder definitions
* Standard Controller: Collect forwarder status from routers and report metrics on the number of forwarders in ready vs not ready state.
* Other: Rename Status to AddressStatus

Issue #3150

@famartinrh With this it should be possible to test forwarding of queues and subscriptions as well

enmasse-ci: :heavy_exclamation_mark: **Build Failed** :heavy_exclamation_mark:
lulf: @enmasse-ci run tests testcase=standard.**

Patch:
diff --git a/agent/lib/artemis.js b/agent/lib/artemis.js
index 399bbf58b..7a6c6a39d 100644
--- a/agent/lib/artemis.js
+++ b/agent/lib/artemis.js
@@ -560,16 +560,29 @@ Artemis.prototype.findDivert = function (name) {
     );
 };
 
-Artemis.prototype.createConnectorService = function (name, source, target) {
+Artemis.prototype.createConnectorService = function (connector) {
     var parameters = {
         "host": process.env.MESSAGING_SERVICE_HOST,
         "port": process.env.MESSAGING_SERVICE_PORT_AMQPS_BROKER,
-        "containerId": name,
-        "clusterId": name,
-        "clientAddress": target,
-        "sourceAddress": source
+        "clusterId": connector.clusterId
     };
-    return this._request('broker', 'createConnectorService', [name, "org.apache.activemq.artemis.integration.amqp.AMQPConnectorServiceFactory", parameters]);
+
+    if (connector.containerId !== undefined) {
+	parameters.containerId = connector.containerId;
+    }
+    if (connector.linkName !== undefined) {
+	parameters.linkName = connector.linkName;
+    }
+    if (connector.targetAddress !== undefined) {
+	parameters.targetAddress = connector.targetAddress;
+    }
+    if (connector.sourceAddress !== undefined) {
+	parameters.sourceAddress = connector.sourceAddress;
+    }
+    if (connector.direction !== undefined) {
+	parameters.direction = connector.direction;
+    }
+    return this._request('broker', 'createConnectorService', [connector.name, "org.apache.activemq.artemis.integration.amqp.AMQPConnectorServiceFactory", parameters]);
 }
 
 
@@ -625,12 +638,12 @@ Artemis.prototype.getGlobalMaxSize = function ()
 /**
  * Create connector service if one does not already exist.
  */
-Artemis.prototype.ensureConnectorService = function (name, source, target) {
+Artemis.prototype.ensureConnectorService = function (connector) {
     var broker = this;
-    return broker.findConnectorService(name).then(
+    return broker.findConnectorService(connector.name).then(
         function (found) {
             if (!found) {
-                return broker.createConnectorService(name, source, target);
+                return broker.createConnectorService(connector);
             }
         }
     );
diff --git a/agent/lib/broker_controller.js b/agent/lib/broker_controller.js
index 0e0643005..70c36028d 100644
--- a/agent/lib/broker_controller.js
+++ b/agent/lib/broker_controller.js
@@ -28,7 +28,7 @@ function BrokerController(event_sink) {
     events.EventEmitter.call(this);
     this.check_in_progress = false;
     this.post_event = event_sink || function (event) { log.debug('event: %j', event); };
-    this.serial_sync = require('./utils.js').serialize(this._sync_broker_addresses.bind(this));
+    this.serial_sync = require('./utils.js').serialize(this._sync_addresses_and_forwarders.bind(this));
     this.addresses_synchronized = false;
     this.busy_count = 0;
     this.retrieve_count = 0;
@@ -155,7 +155,7 @@ function transform_address_stats(address) {
 function transform_connection_stats(raw) {
     return {
         id: raw.connectionID,
-        host: raw.clientAddress,
+        host: raw.targetAddress,
         container: 'not available',
         user: raw.sessions.length ? raw.sessions[0].principal : '',
         senders: [],
@@ -478,6 +478,7 @@ BrokerController.prototype.check_broker_addresses = function () {
     }
 };
 
+// Only used in tests
 BrokerController.prototype._sync_addresses = function (addresses) {
     this.addresses = addresses.reduce(function (map, a) { map[a.address] = a; return map; }, {});
     return this._sync_broker_addresses();
@@ -518,45 +519,91 @@ BrokerController.prototype._sync_broker_addresses = function (retry) {
     });
 };
 
-BrokerController.prototype.destroy_connector = function (address) {
+BrokerController.prototype.destroy_connector = function (connector) {
     var self = this;
-    log.info('[%s] Deleting connector for "%s"...', self.id, address);
-    return self.broker.destroyConnectorService(address).then(function() {
-        log.info('[%s] Deleted connector for "%s"', self.id, address);
+    log.info('[%s] Deleting connector for "%s"...', self.id, connector.name);
+    return self.broker.destroyConnectorService(connector.name).then(function() {
+        log.info('[%s] Deleted connector for "%s"', self.id, connector.name);
     }).catch(function (error) {
-        log.error('[%s] Failed to delete connector for "%s": %s', self.id, address, error);
+        log.error('[%s] Failed to delete connector for "%s": %s', self.id, connector.name, error);
     });
 };
 
-BrokerController.prototype.create_connector = function (address) {
+BrokerController.prototype.create_connector = function (connector) {
     var self = this;
-    log.info('[%s] Creating connector for "%s"...', self.id, address);
-    return self.broker.createConnectorService(address).then(function() {
-        log.info('[%s] Created connector for "%s"', self.id, address);
+    log.info('[%s] Creating connector "%j"...', self.id, connector);
+    return self.broker.createConnectorService(connector).then(function() {
+        log.info('[%s] Created connector for "%s"', self.id, connector.name);
     }).catch(function (error) {
-        log.error('[%s] Failed to create connector for "%s": %s', self.id, address, error);
+        log.error('[%s] Failed to create connector for "%s": %s', self.id, connector.name, error);
     });
 };
 
-BrokerController.prototype.destroy_connectors = function (addresses) {
-    return Promise.all(addresses.map(this.destroy_connector.bind(this)));
+BrokerController.prototype.destroy_connectors = function (connectors) {
+    return Promise.all(connectors.map(this.destroy_connector.bind(this)));
 };
 
-BrokerController.prototype.create_connectors = function (addresses) {
-    return Promise.all(addresses.map(this.create_connector.bind(this)));
+BrokerController.prototype.create_connectors = function (connectors) {
+    return Promise.all(connectors.map(this.create_connector.bind(this)));
 };
 
 function connectors_of_interest(name) {
-    return name !== 'amqp-connector' && name !== 'router-connector';
+    return name !== 'amqp-connector' &&
+	name !== 'router-connector-in' &&
+	name !== 'router-connector-out' &&
+	name !== 'amqp-connector-in' &&
+	name !== 'amqp-connector-out' &&
+	name !== 'address-connector-in' &&
+	name !== 'address-connector-out' &&
+	name !== 'ragent-connector' &&
+	!name.startsWith("$override");
 }
 
-BrokerController.prototype._ensure_connectors = function (desired) {
+function connector_compare(a, b) {
+    return myutils.string_compare(a.name, b.name);
+}
+
+BrokerController.prototype._sync_broker_forwarders = function () {
     var self = this;
+    var desired = [];
+    for (var address in this.addresses) {
+	var address_type = this.addresses[address].type;
+	// We will only check queues and subscriptions for forwarders
+	if (address_type == "queue" || address_type == "subscription") {
+	    var address_name = this.addresses[address].name;
+	    var forwarders = this.addresses[address].forwarders;
+	    if (forwarders !== undefined) {
+		for (var fwdidx in forwarders) {
+		    var forwarder = forwarders[fwdidx];
+		    var forwarder_name = address_name + "." + forwarder.name + "." + forwarder.direction;
+		    // Only create forwarder for subscriptions if the direction is outward
+		    if (address_type === "queue" || (address_type === "subscription" && forwarder.direction === "out")) {
+			var sourceAddress = forwarder.direction === "out" ? address : forwarder.remoteAddress;
+			var targetAddress = forwarder.direction === "out" ? forwarder.remoteAddress : address;
+			var forwarder_entry = {
+			    name: forwarder_name,
+			    clusterId: address_name,
+			    containerId: address_name,
+			    linkName: forwarder_name,
+			    targetAddress: targetAddress,
+			    sourceAddress: sourceAddress,
+			    direction: forwarder.direction
+			};
+			desired.push(forwarder_entry);
+		    }
+		}
+	    }
+	}
+    }
+    log.debug("desired connectors: %j", desired);
     return this.broker.getConnectorServices().then(function (results) {
         var actual = results.filter(connectors_of_interest);
         actual.sort();
+	actual = actual.map(function (name) {
+	    return {name: name}
+	});
 
-        var difference = myutils.changes(actual, desired, myutils.string_compare);
+        var difference = myutils.changes(actual, desired, connector_compare);
         if (difference) {
             log.info('[%s] %d connectors missing, %d to be removed', self.id, difference.added.length, difference.removed.length);
             return self.destroy_connectors(difference.removed).then(function () {
@@ -570,10 +617,8 @@ BrokerController.prototype._ensure_connectors = function (desired) {
     });
 };
 
-BrokerController.prototype._sync_addresses_and_connectors = function () {
-    var desired = Object.keys(this.addresses);
-    desired.sort();
-    return this._sync_broker_addresses().then(this._ensure_connectors.bind(this, desired));
+BrokerController.prototype._sync_addresses_and_forwarders = function () {
+    return this._sync_broker_addresses().then(this._sync_broker_forwarders());
 }
 
 module.exports.create_controller = function (connection, event_sink) {
diff --git a/agent/lib/subctrl.js b/agent/lib/subctrl.js
index c98ee75b6..f5a55f1ac 100644
--- a/agent/lib/subctrl.js
+++ b/agent/lib/subctrl.js
@@ -117,7 +117,16 @@ SubscriptionControl.prototype.subscribe = function (subscription_id, topics) {
     return ensure_queue(subscription_id, pods).then(
         function (pod) {
             log.info('after ensure looking at pod ' + JSON.stringify(pod.name) + " with broker undefined " + (pod.broker === undefined));
-            return pod.broker.ensureConnectorService(subscription_id, subscription_id, subscription_id).then(
+	    var connector = {
+		name: "$override." + subscription_id,
+		clusterId: subscription_id,
+		linkName: subscription_id,
+		sourceAddress: subscription_id,
+		targetAddress: subscription_id,
+		containerId: subscription_id,
+		direction: 'out'
+	    };
+            return pod.broker.ensureConnectorService(connector).then(
                 function () {
                     return Promise.all(Object.keys(topics).map(
                         function (topic) {
diff --git a/api-model/src/main/java/io/enmasse/address/model/Address.java b/api-model/src/main/java/io/enmasse/address/model/Address.java
index 1058ba0d8..34a7cd791 100644
--- a/api-model/src/main/java/io/enmasse/address/model/Address.java
+++ b/api-model/src/main/java/io/enmasse/address/model/Address.java
@@ -10,6 +10,7 @@ import java.util.*;
 import javax.validation.Valid;
 import javax.validation.constraints.NotNull;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import io.enmasse.common.model.AbstractHasMetadata;
 import io.enmasse.common.model.DefaultCustomResource;
 import io.enmasse.model.validation.AddressName;
@@ -43,7 +44,7 @@ public class Address extends AbstractHasMetadata<Address> implements AddressOrAd
     @NotNull @Valid
     private AddressSpec spec = new AddressSpec();
     @NotNull @Valid
-    private Status status = new Status();
+    private AddressStatus status = new AddressStatus();
 
     public Address() {
         super(KIND, CoreCrd.API_VERSION);
@@ -59,11 +60,11 @@ public class Address extends AbstractHasMetadata<Address> implements AddressOrAd
         return spec;
     }
 
-    public void setStatus(Status status) {
+    public void setStatus(AddressStatus status) {
         this.status = status;
     }
 
-    public Status getStatus() {
+    public AddressStatus getStatus() {
         return status;
     }
 
@@ -131,4 +132,9 @@ public class Address extends AbstractHasMetadata<Address> implements AddressOrAd
         }
         return extractAddressFromName(address);
     }
+
+    @JsonIgnore
+    public String getForwarderLinkName(AddressSpecForwarder forwarder) {
+        return String.format("%s.%s.%s", getMetadata().getName(), forwarder.getName(), forwarder.getDirection().name());
+    }
 }
diff --git a/api-model/src/main/java/io/enmasse/address/model/AddressSpec.java b/api-model/src/main/java/io/enmasse/address/model/AddressSpec.java
index b9e0788e1..b1805d2ba 100644
--- a/api-model/src/main/java/io/enmasse/address/model/AddressSpec.java
+++ b/api-model/src/main/java/io/enmasse/address/model/AddressSpec.java
@@ -4,8 +4,10 @@
  */
 package io.enmasse.address.model;
 
+import java.util.List;
 import java.util.Objects;
 
+import javax.validation.Valid;
 import javax.validation.constraints.NotNull;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
@@ -39,6 +41,8 @@ public class AddressSpec extends AbstractWithAdditionalProperties {
     private String plan;
     private String topic;
 
+    private List<@Valid AddressSpecForwarder> forwarders;
+
     public void setAddress(String address) {
         this.address = address;
     }
@@ -83,16 +87,27 @@ public class AddressSpec extends AbstractWithAdditionalProperties {
         return this.topic;
     }
 
+    public List<AddressSpecForwarder> getForwarders() {
+        return forwarders;
+    }
+
+    public void setForwarders(List<AddressSpecForwarder> forwarders) {
+        this.forwarders = forwarders;
+    }
+
     @Override
     public String toString() {
 
          final StringBuilder sb = new StringBuilder("{");
 
-         sb.append("address=").append(address).append(",");
-         sb.append("type=").append(type).append(",");
-         sb.append("plan=").append(plan).append(",");
+         sb.append("address=").append(address);
+         sb.append(",type=").append(type);
+         sb.append(",plan=").append(plan);
          if (topic != null) {
-             sb.append("topic=").append(topic);
+             sb.append(",topic=").append(topic);
+         }
+         if (forwarders != null) {
+             sb.append(",forwarders=").append(forwarders);
          }
          sb.append("}");
 
@@ -116,5 +131,4 @@ public class AddressSpec extends AbstractWithAdditionalProperties {
     public int hashCode() {
         return Objects.hash(address, addressSpace);
     }
-
 }
diff --git a/api-model/src/main/java/io/enmasse/address/model/AddressSpecForwarder.java b/api-model/src/main/java/io/enmasse/address/model/AddressSpecForwarder.java
new file mode 100644
index 000000000..2ebcfd01e
--- /dev/null
+++ b/api-model/src/main/java/io/enmasse/address/model/AddressSpecForwarder.java
@@ -0,0 +1,68 @@
+/*
+ * 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.address.model;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import io.enmasse.admin.model.v1.AbstractWithAdditionalProperties;
+import io.fabric8.kubernetes.api.model.Doneable;
+import io.sundr.builder.annotations.Buildable;
+import io.sundr.builder.annotations.BuildableReference;
+import io.sundr.builder.annotations.Inline;
+
+import javax.validation.constraints.NotNull;
+
+@Buildable(
+        editableEnabled = false,
+        generateBuilderPackage = false,
+        builderPackage = "io.fabric8.kubernetes.api.builder",
+        refs= {@BuildableReference(AbstractWithAdditionalProperties.class)},
+        inline = @Inline(
+                type = Doneable.class,
+                prefix = "Doneable",
+                value = "done"
+        )
+)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class AddressSpecForwarder extends AbstractWithAdditionalProperties {
+    @NotNull
+    private String name;
+    @NotNull
+    private String remoteAddress;
+    @NotNull
+    private AddressSpecForwarderDirection direction;
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public String getRemoteAddress() {
+        return remoteAddress;
+    }
+
+    public void setRemoteAddress(String remoteAddress) {
+        this.remoteAddress = remoteAddress;
+    }
+
+    public AddressSpecForwarderDirection getDirection() {
+        return direction;
+    }
+
+    public void setDirection(AddressSpecForwarderDirection direction) {
+        this.direction = direction;
+    }
+
+    @Override
+    public String toString() {
+        return "AddressSpecForwarder{" +
+                "name='" + name + '\'' +
+                ", remoteAddress='" + remoteAddress + '\'' +
+                ", direction=" + direction +
+                '}';
+    }
+}
diff --git a/api-model/src/main/java/io/enmasse/address/model/AddressSpecForwarderDirection.java b/api-model/src/main/java/io/enmasse/address/model/AddressSpecForwarderDirection.java
new file mode 100644
index 000000000..b91b2db7e
--- /dev/null
+++ b/api-model/src/main/java/io/enmasse/address/model/AddressSpecForwarderDirection.java
@@ -0,0 +1,10 @@
+/*
+ * 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.address.model;
+
+public enum AddressSpecForwarderDirection {
+    in,
+    out
+}
diff --git a/api-model/src/main/java/io/enmasse/address/model/Status.java b/api-model/src/main/java/io/enmasse/address/model/AddressStatus.java
similarity index 68%
rename from api-model/src/main/java/io/enmasse/address/model/Status.java
rename to api-model/src/main/java/io/enmasse/address/model/AddressStatus.java
index 18ad6f62f..f4cc9fbd7 100644
--- a/api-model/src/main/java/io/enmasse/address/model/Status.java
+++ b/api-model/src/main/java/io/enmasse/address/model/AddressStatus.java
@@ -10,6 +10,7 @@ import java.util.List;
 import java.util.Objects;
 
 import javax.validation.Valid;
+import javax.validation.constraints.NotNull;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -35,7 +36,7 @@ import io.sundr.builder.annotations.Inline;
                 )
         )
 @JsonInclude(JsonInclude.Include.NON_NULL)
-public class Status extends AbstractWithAdditionalProperties {
+public class AddressStatus extends AbstractWithAdditionalProperties {
 
     @JsonProperty("isReady")
     private boolean ready = false;
@@ -44,14 +45,16 @@ public class Status extends AbstractWithAdditionalProperties {
     private List<@Valid BrokerStatus> brokerStatuses = new ArrayList<>();
     private AddressPlanStatus planStatus;
 
-    public Status() {
+    private List<@Valid AddressStatusForwarder> forwarderStatuses;
+
+    public AddressStatus() {
     }
 
-    public Status(boolean ready) {
+    public AddressStatus(boolean ready) {
         this.ready = ready;
     }
 
-    public Status(io.enmasse.address.model.Status other) {
+    public AddressStatus(AddressStatus other) {
         this.ready = other.isReady();
         this.phase = other.getPhase();
         this.messages = new ArrayList<>(other.getMessages());
@@ -59,6 +62,17 @@ public class Status extends AbstractWithAdditionalProperties {
         for (BrokerStatus brokerStatus : other.getBrokerStatuses()) {
             brokerStatuses.add(new BrokerStatus(brokerStatus.getClusterId(), brokerStatus.getContainerId(), brokerStatus.getState()));
         }
+        if (other.getForwarderStatuses() != null) {
+            this.forwarderStatuses = new ArrayList<>();
+            for (AddressStatusForwarder forwarderStatus : other.getForwarderStatuses()) {
+                forwarderStatuses.add(new AddressStatusForwarderBuilder()
+                        .withName(forwarderStatus.getName())
+                        .withReady(forwarderStatus.isReady())
+                        .withMessages(new ArrayList<>(forwarderStatus.getMessages()))
+                        .build());
+            }
+        }
+
     }
 
     public boolean isReady() {
@@ -73,12 +87,12 @@ public class Status extends AbstractWithAdditionalProperties {
         return Collections.unmodifiableList(brokerStatuses);
     }
 
-    public Status setReady(boolean ready) {
+    public AddressStatus setReady(boolean ready) {
         this.ready = ready;
         return this;
     }
 
-    public Status setPhase(Phase phase) {
+    public AddressStatus setPhase(Phase phase) {
         this.phase = phase;
         return this;
     }
@@ -87,27 +101,27 @@ public class Status extends AbstractWithAdditionalProperties {
         return messages;
     }
 
-    public Status appendMessage(String message) {
+    public AddressStatus appendMessage(String message) {
         this.messages.add(message);
         return this;
     }
 
-    public Status clearMessages() {
+    public AddressStatus clearMessages() {
         this.messages.clear();
         return this;
     }
 
-    public Status setMessages(List<String> messages) {
+    public AddressStatus setMessages(List<String> messages) {
         this.messages = new ArrayList<>(messages);
         return this;
     }
 
-    public Status appendBrokerStatus(BrokerStatus brokerStatus) {
+    public AddressStatus appendBrokerStatus(BrokerStatus brokerStatus) {
         this.brokerStatuses.add(brokerStatus);
         return this;
     }
 
-    public Status setBrokerStatuses(List<BrokerStatus> brokerStatuses) {
+    public AddressStatus setBrokerStatuses(List<BrokerStatus> brokerStatuses) {
         this.brokerStatuses = new ArrayList<>(brokerStatuses);
         return this;
     }
@@ -117,17 +131,18 @@ public class Status extends AbstractWithAdditionalProperties {
     public boolean equals(Object o) {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
-        Status status = (Status) o;
+        AddressStatus status = (AddressStatus) o;
         return ready == status.ready &&
                 phase == status.phase &&
                 Objects.equals(messages, status.messages) &&
                 Objects.equals(brokerStatuses, status.brokerStatuses) &&
-                Objects.equals(planStatus, status.planStatus);
+                Objects.equals(planStatus, status.planStatus) &&
+                Objects.equals(forwarderStatuses, status.forwarderStatuses);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(ready, phase, messages, brokerStatuses, planStatus);
+        return Objects.hash(ready, phase, messages, brokerStatuses, planStatus, forwarderStatuses);
     }
 
 
@@ -139,6 +154,7 @@ public class Status extends AbstractWithAdditionalProperties {
                 .append(",").append("messages=").append(messages)
                 .append(",").append("brokerStatuses=").append(brokerStatuses)
                 .append(",").append("planStatus=").append(planStatus)
+                .append(",").append("forwarderStatuses=").append(forwarderStatuses)
                 .append("}")
                 .toString();
     }
@@ -154,4 +170,12 @@ public class Status extends AbstractWithAdditionalProperties {
     public void setPlanStatus(AddressPlanStatus planStatus) {
         this.planStatus = planStatus;
     }
+
+    public List<AddressStatusForwarder> getForwarderStatuses() {
+        return forwarderStatuses;
+    }
+
+    public void setForwarderStatuses(List<AddressStatusForwarder> forwarderStatuses) {
+        this.forwarderStatuses = new ArrayList<>(forwarderStatuses);
+    }
 }
diff --git a/api-model/src/main/java/io/enmasse/address/model/AddressStatusForwarder.java b/api-model/src/main/java/io/enmasse/address/model/AddressStatusForwarder.java
new file mode 100644
index 000000000..b9ab4f54b
--- /dev/null
+++ b/api-model/src/main/java/io/enmasse/address/model/AddressStatusForwarder.java
@@ -0,0 +1,93 @@
+/*
+ * 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.address.model;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.enmasse.admin.model.v1.AbstractWithAdditionalProperties;
+import io.fabric8.kubernetes.api.model.Doneable;
+import io.sundr.builder.annotations.Buildable;
+import io.sundr.builder.annotations.BuildableReference;
+import io.sundr.builder.annotations.Inline;
+
+import javax.validation.constraints.NotNull;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+@Buildable(
+        editableEnabled = false,
+        generateBuilderPackage = false,
+        builderPackage = "io.fabric8.kubernetes.api.builder",
+        refs= {@BuildableReference(AbstractWithAdditionalProperties.class)},
+        inline = @Inline(
+                type = Doneable.class,
+                prefix = "Doneable",
+                value = "done"
+        )
+)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class AddressStatusForwarder extends AbstractWithAdditionalProperties {
+
+    @NotNull
+    private String name;
+
+    @JsonProperty("isReady")
+    private boolean ready = false;
+
+    private List<String> messages = new ArrayList<>();
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public List<String> getMessages() {
+        return messages;
+    }
+
+    public void setMessages(List<String> messages) {
+        this.messages = new ArrayList<>(messages);
+    }
+
+    @Override
+    public String toString() {
+        return "AddressStatusForwarder{" +
+                "name='" + name + '\'' +
+                ", ready=" + ready +
+                ", messages=" + messages +
+                '}';
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        AddressStatusForwarder that = (AddressStatusForwarder) o;
+        return ready == that.ready &&
+                Objects.equals(name, that.name) &&
+                Objects.equals(messages, that.messages);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(name, ready, messages);
+    }
+
+    public boolean isReady() {
+        return ready;
+    }
+
+    public void setReady(boolean ready) {
+        this.ready = ready;
+    }
+
+    public void appendMessage(String message) {
+        this.messages.add(message);
+    }
+}
diff --git a/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/AMQPConnectorService.java b/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/AMQPConnectorService.java
index af868607f..a2592bd85 100644
--- a/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/AMQPConnectorService.java
+++ b/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/AMQPConnectorService.java
@@ -59,7 +59,7 @@ public class AMQPConnectorService implements ConnectorService, BaseConnectionLif
    private volatile boolean started = false;
    private final Map<String, Object> connectorConfig;
 
-   public AMQPConnectorService(String connectorName, Map<String, Object> connectorConfig, String containerId, String groupId, Optional<SubscriberInfo> subscriberInfo, ActiveMQServer server, ScheduledExecutorService scheduledExecutorService, ExecutorService nettyThreadPool, int idleTimeout) {
+   public AMQPConnectorService(String connectorName, Map<String, Object> connectorConfig, String containerId, String groupId, Optional<LinkInfo> linkInfo, ActiveMQServer server, ScheduledExecutorService scheduledExecutorService, ExecutorService nettyThreadPool, int idleTimeout) {
       this.name = connectorName;
       this.connectorConfig = connectorConfig;
       this.server = server;
@@ -96,7 +96,7 @@ public class AMQPConnectorService implements ConnectorService, BaseConnectionLif
       } else  {
          ActiveMQAMQPLogger.LOGGER.infov("Disabling SSL for AMQP Connector {0}", name);
       }
-      this.lifecycleHandler = new ProtonClientConnectionManager(factory, subscriberInfo.map(LinkInitiator::new), saslClientFactory);
+      this.lifecycleHandler = new ProtonClientConnectionManager(factory, linkInfo.map(LinkInitiator::new), saslClientFactory);
    }
 
    @Override
diff --git a/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/AMQPConnectorServiceFactory.java b/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/AMQPConnectorServiceFactory.java
index 9a45bc7ad..ac84b29fc 100644
--- a/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/AMQPConnectorServiceFactory.java
+++ b/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/AMQPConnectorServiceFactory.java
@@ -37,10 +37,12 @@ import java.util.concurrent.ScheduledExecutorService;
  * Connector service factory for AMQP Connector Services that can be used to establish outgoing AMQP connections.
  */
 public class AMQPConnectorServiceFactory implements ConnectorServiceFactory {
-   private static final String CONTAINER = "containerId";
+   private static final String CONTAINER_ID = "containerId";
    private static final String CLUSTER = "clusterId";
    private static final String SOURCE_ADDRESS = "sourceAddress";
-   private static final String CLIENT_ADDRESS = "clientAddress";
+   private static final String TARGET_ADDRESS = "targetAddress";
+   private static final String LINK_NAME = "linkName";
+   private static final String DIRECTION = "direction";
    private static final String NETTY_THREADS = "nettyThreads";
    private static final String IDLE_TIMEOUT = "idleTimeoutMs";
 
@@ -49,9 +51,11 @@ public class AMQPConnectorServiceFactory implements ConnectorServiceFactory {
 
    private static Set<String> initializeAllowedProperties() {
       Set<String> properties = initializeRequiredProperties();
-      properties.add(CLIENT_ADDRESS);
+      properties.add(TARGET_ADDRESS);
       properties.add(SOURCE_ADDRESS);
-      properties.add(CONTAINER);
+      properties.add(DIRECTION);
+      properties.add(LINK_NAME);
+      properties.add(CONTAINER_ID);
       properties.add(TransportConstants.SSL_ENABLED_PROP_NAME);
       properties.add(TransportConstants.SSL_PROVIDER);
       properties.add(TransportConstants.VERIFY_HOST_PROP_NAME);
@@ -99,20 +103,22 @@ public class AMQPConnectorServiceFactory implements ConnectorServiceFactory {
       setOrDefault(configuration, connectorConfig, TransportConstants.SSL_PROVIDER, "OPENSSL");
 
       Optional<String> sourceAddress = Optional.ofNullable((String)configuration.get(SOURCE_ADDRESS));
-      Optional<String> clientAddress = Optional.ofNullable((String)configuration.get(CLIENT_ADDRESS));
-      Optional<String> container = Optional.ofNullable((String)configuration.get(CONTAINER));
+      Optional<String> targetAddress = Optional.ofNullable((String)configuration.get(TARGET_ADDRESS));
+      Optional<Direction> direction = Optional.ofNullable((String)configuration.get(DIRECTION)).map(Direction::valueOf);
+      Optional<String> linkName = Optional.ofNullable((String)configuration.get(LINK_NAME));
+      String containerId = Optional.ofNullable((String)configuration.get(CONTAINER_ID)).orElse(clusterId);
 
       int nettyThreads = Optional.ofNullable((String)configuration.get(NETTY_THREADS)).map(Integer::parseInt).orElse(4);
       int idleTimeout = Optional.ofNullable((String)configuration.get(IDLE_TIMEOUT)).map(Integer::parseInt).orElse(16_000);
 
-      Optional<SubscriberInfo> info = sourceAddress.flatMap(s ->
-              clientAddress.flatMap(c ->
-                      container.map(o -> new SubscriberInfo(o, s, c))));
+      Optional<LinkInfo> linkInfo = sourceAddress.flatMap(s ->
+              targetAddress.flatMap(t ->
+                      direction.map(d -> new LinkInfo(linkName.orElse(null), s, t, d))));
 
-      ActiveMQAMQPLogger.LOGGER.infof("Creating connector host %s port %s", configuration.get(TransportConstants.HOST_PROP_NAME), configuration.get(TransportConstants.PORT_PROP_NAME));
-      String containerId = clusterId;
-      if (container.isPresent()) {
-         containerId = container.get();
+      if (linkInfo.isPresent()) {
+         ActiveMQAMQPLogger.LOGGER.infof("Creating connector host %s port %s with link %s", configuration.get(TransportConstants.HOST_PROP_NAME), configuration.get(TransportConstants.PORT_PROP_NAME), linkInfo.get());
+      } else {
+         ActiveMQAMQPLogger.LOGGER.infof("Creating connector host %s port %s", configuration.get(TransportConstants.HOST_PROP_NAME), configuration.get(TransportConstants.PORT_PROP_NAME));
       }
 
       ExecutorService nettyThreadPool = Executors.newFixedThreadPool(nettyThreads, new DefaultThreadFactory("connector-" + connectorName));
@@ -120,7 +126,7 @@ public class AMQPConnectorServiceFactory implements ConnectorServiceFactory {
               connectorConfig,
               containerId,
               clusterId,
-              info,
+              linkInfo,
               ((PostOfficeImpl)postOffice).getServer(),
               scheduledExecutorService,
               nettyThreadPool,
diff --git a/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/Direction.java b/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/Direction.java
new file mode 100644
index 000000000..70f184148
--- /dev/null
+++ b/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/Direction.java
@@ -0,0 +1,10 @@
+/*
+ * Copyright 2019, EnMasse authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package org.apache.activemq.artemis.integration.amqp;
+
+public enum Direction {
+    in,
+    out
+}
diff --git a/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/SubscriberInfo.java b/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/LinkInfo.java
similarity index 58%
rename from broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/SubscriberInfo.java
rename to broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/LinkInfo.java
index 9de4690e0..5dfa49dab 100644
--- a/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/SubscriberInfo.java
+++ b/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/LinkInfo.java
@@ -20,26 +20,42 @@
  */
 package org.apache.activemq.artemis.integration.amqp;
 
-public class SubscriberInfo {
-   private final String clientId;
+public class LinkInfo {
+   private final String linkName;
    private final String sourceAddress;
-   private final String clientAddress;
+   private final String targetAddress;
+   private final Direction direction;
 
-   public SubscriberInfo(String clientId, String sourceAddress, String clientAddress) {
-      this.clientId = clientId;
+   public LinkInfo(String linkName, String sourceAddress, String targetAddress, Direction direction) {
+      this.linkName = linkName;
       this.sourceAddress = sourceAddress;
-      this.clientAddress = clientAddress;
+      this.targetAddress = targetAddress;
+      this.direction = direction;
    }
 
-   public String getClientAddress() {
-      return clientAddress;
+   public String getTargetAddress() {
+      return targetAddress;
    }
 
-   public String getClientId() {
-      return clientId;
+   public String getLinkName() {
+      return linkName;
    }
 
    public String getSourceAddress() {
       return sourceAddress;
    }
+
+   public Direction getDirection() {
+      return direction;
+   }
+
+   @Override
+   public String toString() {
+      return "LinkInfo{" +
+              "linkName='" + linkName + '\'' +
+              ", sourceAddress='" + sourceAddress + '\'' +
+              ", targetAddress='" + targetAddress + '\'' +
+              ", direction=" + direction +
+              '}';
+   }
 }
diff --git a/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/LinkInitiator.java b/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/LinkInitiator.java
index 880e317ae..9b3b6eac5 100644
--- a/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/LinkInitiator.java
+++ b/broker-plugin/amqp-connector/src/main/java/org/apache/activemq/artemis/integration/amqp/LinkInitiator.java
@@ -10,7 +10,7 @@
  * (the "License"); you may not use this file except in compliance with
  * the License. You may obtain a copy of the License at
  *
- *     http://www.apache.org/licenses/LICENSE-2.0
+ *    http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -24,24 +24,20 @@ import io.netty.buffer.ByteBuf;
 import org.apache.activemq.artemis.protocol.amqp.proton.handler.EventHandler;
 import org.apache.activemq.artemis.protocol.amqp.proton.handler.ProtonHandler;
 import org.apache.activemq.artemis.spi.core.remoting.ReadyListener;
+import org.apache.qpid.proton.amqp.messaging.Received;
 import org.apache.qpid.proton.amqp.messaging.Source;
 import org.apache.qpid.proton.amqp.messaging.Target;
 import org.apache.qpid.proton.amqp.messaging.TerminusDurability;
-import org.apache.qpid.proton.engine.Connection;
-import org.apache.qpid.proton.engine.Delivery;
-import org.apache.qpid.proton.engine.Link;
-import org.apache.qpid.proton.engine.Sender;
-import org.apache.qpid.proton.engine.Session;
-import org.apache.qpid.proton.engine.Transport;
+import org.apache.qpid.proton.engine.*;
 
 /**
  * Event handler for establishing outgoing links
  */
 public class LinkInitiator implements EventHandler {
-   private final SubscriberInfo subscriberInfo;
+   private final LinkInfo linkInfo;
 
-   public LinkInitiator(SubscriberInfo subscriberInfo) {
-      this.subscriberInfo = subscriberInfo;
+   public LinkInitiator(LinkInfo linkInfo) {
+      this.linkInfo = linkInfo;
    }
 
    @Override
@@ -51,12 +47,12 @@ public class LinkInitiator implements EventHandler {
 
    @Override
    public void onSaslRemoteMechanismChosen(ProtonHandler handler, String mech) {
-      
+
    }
 
    @Override
    public void onAuthFailed(final ProtonHandler protonHandler, final Connection connection) {
-      
+
    }
 
    @Override
@@ -111,7 +107,7 @@ public class LinkInitiator implements EventHandler {
 
    @Override
    public void onRemoteOpen(Session session) throws Exception {
-      createSender(session);
+      createLink(session);
    }
 
    @Override
@@ -141,17 +137,18 @@ public class LinkInitiator implements EventHandler {
 
    @Override
    public void onRemoteOpen(Link link) throws Exception {
-
+      ActiveMQAMQPLogger.LOGGER.infov("{0} onRemoteOpen", link.getName());
    }
 
    @Override
    public void onLocalClose(Link link) throws Exception {
-
+      ActiveMQAMQPLogger.LOGGER.infov("{0} onLocalClose", link.getName());
    }
 
    @Override
    public void onRemoteClose(Link link) throws Exception {
-
+      ActiveMQAMQPLogger.LOGGER.infov("Link {0} closed. Closing connection", link.getName());
+      link.getSession().getConnection().close();
    }
 
    @Override
@@ -166,12 +163,13 @@ public class LinkInitiator implements EventHandler {
 
    @Override
    public void onRemoteDetach(Link link) throws Exception {
-
+      ActiveMQAMQPLogger.LOGGER.infov("Link {0} detached. Closing connection", link.getName());
+      link.getSession().getConnection().close();
    }
 
    @Override
    public void onLocalDetach(Link link) throws Exception {
-
+      ActiveMQAMQPLogger.LOGGER.infov("{0} onLocalDetach", link.getName());
    }
 
    @Override
@@ -181,7 +179,10 @@ public class LinkInitiator implements EventHandler {
 
    @Override
    public boolean flowControl(ReadyListener readyListener) {
-       return true;
+      if (this.linkInfo.getDirection().equals(Direction.in)) {
+         readyListener.readyForWriting();
+      }
+      return true;
    }
 
    @Override
@@ -194,14 +195,48 @@ public class LinkInitiator implements EventHandler {
 
    }
 
-   private void createSender(org.apache.qpid.proton.engine.Session session) throws Exception {
-      Sender sender = session.sender(subscriberInfo.getClientId());
+   private void createLink(org.apache.qpid.proton.engine.Session session) throws Exception {
+      switch (linkInfo.getDirection()) {
+         case out:
+            createSender(session);
+            break;
+         case in:
+            createReceiver(session);
+            break;
+      }
+   }
+
+   private void createReceiver(Session session) {
+      Receiver receiver;
+      if (linkInfo.getLinkName() != null) {
+         receiver = session.receiver(linkInfo.getLinkName());
+      } else {
+         receiver = session.receiver(linkInfo.getSourceAddress());
+      }
+      Target target = new Target();
+      target.setAddress(linkInfo.getTargetAddress());
+      receiver.setTarget(target);
+
+      Source source = new Source();
+      source.setAddress(linkInfo.getSourceAddress());
+      source.setDurable(TerminusDurability.UNSETTLED_STATE);
+      receiver.setSource(source);
+      receiver.open();
+   }
+
+   private void createSender(Session session) {
+      Sender sender;
+      if (linkInfo.getLinkName() != null) {
+         sender = session.sender(linkInfo.getLinkName());
+      } else {
+         sender = session.sender(linkInfo.getTargetAddress());
+      }
       Target target = new Target();
-      target.setAddress(subscriberInfo.getClientAddress());
+      target.setAddress(linkInfo.getTargetAddress());
       sender.setTarget(target);
 
       Source source = new Source();
-      source.setAddress(subscriberInfo.getClientAddress());
+      source.setAddress(linkInfo.getSourceAddress());
       source.setDurable(TerminusDurability.UNSETTLED_STATE);
       sender.setSource(source);
 
diff --git a/documentation/design/proposals/bridging_external.adoc b/documentation/design/proposals/bridging_external.adoc
index 124633763..f7cf82d9a 100644
--- a/documentation/design/proposals/bridging_external.adoc
+++ b/documentation/design/proposals/bridging_external.adoc
@@ -123,7 +123,7 @@ The standard controller will collect statistics about forwarders and attach to a
 
 ==== Agent
 
-For each forwarder on an address, the agent will create an autoLink pointing to a connector matching the address pattern.
+For each forwarder on an address, the agent will create a connector from the broker to the remote address. 
 
 == Testing
 
diff --git a/documentation/modules/proc-iot-installing-services.adoc b/documentation/modules/proc-iot-installing-services.adoc
index 4a8dc7c8f..cca1e4ad1 100644
--- a/documentation/modules/proc-iot-installing-services.adoc
+++ b/documentation/modules/proc-iot-installing-services.adoc
@@ -10,6 +10,12 @@ To get started using the IoT feature on {ProductName}, you must first install th
 
 .Procedure
 
+. (Optional) If you want to deploy to a project other than `{ProductNamespace}` you must run the following command and substitute `{ProductNamespace}` in subsequent steps:
++
+[options="nowrap",subs="+quotes,attributes"]
+----
+sed -i 's/{ProductNamespace}/_my-project_/' install/preview-bundles/iot/*.yaml
+----
 . Deploy the IoT bundles:
 +
 [options="nowrap",subs="attributes"]
diff --git a/iot/iot-application-base/pom.xml b/iot/iot-application-base/pom.xml
index 557427a4c..c60ae6591 100644
--- a/iot/iot-application-base/pom.xml
+++ b/iot/iot-application-base/pom.xml
@@ -28,18 +28,18 @@
       <artifactId>netty-tcnative</artifactId>
     </dependency>
 
-    <dependency>
-      <groupId>io.micrometer</groupId>
-      <artifactId>micrometer-registry-prometheus</artifactId>
-    </dependency>
-
     <dependency>
       <groupId>io.netty</groupId>
       <artifactId>netty-tcnative</artifactId>
-      <classifier>linux-x86_64-fedora</classifier>
+      <classifier>${netty-tcnative.classifier}</classifier>
       <scope>runtime</scope>
     </dependency>
 
+    <dependency>
+      <groupId>io.micrometer</groupId>
+      <artifactId>micrometer-registry-prometheus</artifactId>
+    </dependency>
+
   </dependencies>
 
 </project>
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 8379e22e5..a52c40a85 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
@@ -29,6 +29,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.*;
+import java.util.function.BinaryOperator;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
@@ -123,6 +124,17 @@ public class AddressController implements Watcher<Address> {
                 address.getStatus().setReady(true);
             }
 
+            if (address.getSpec().getForwarders() != null) {
+                List<AddressStatusForwarder> forwarderStatuses = new ArrayList<>();
+                for (AddressSpecForwarder forwarder : address.getSpec().getForwarders()) {
+                    forwarderStatuses.add(new AddressStatusForwarderBuilder()
+                            .withName(forwarder.getName())
+                            .withReady(true)
+                            .build());
+                }
+                address.getStatus().setForwarderStatuses(forwarderStatuses);
+            }
+
             Address existing = validAddresses.get(address.getSpec().getAddress());
             if (existing != null) {
                 if (!address.getStatus().getPhase().equals(Pending) && existing.getStatus().getPhase().equals(Pending)) {
@@ -171,7 +183,11 @@ public class AddressController implements Watcher<Address> {
         long provisionResources = System.nanoTime();
 
         Set<Address> liveAddresses = filterByPhases(addressSet, EnumSet.of(Configuring, Active));
-        List<RouterStatus> routerStatusList = checkRouterStatuses();
+        boolean checkRouterLinks = liveAddresses.stream()
+                .anyMatch(a -> Arrays.asList("queue", "subscription").contains(a.getSpec().getType()) &&
+                        a.getSpec().getForwarders() != null && !a.getSpec().getForwarders().isEmpty());
+
+        List<RouterStatus> routerStatusList = checkRouterStatuses(checkRouterLinks);
 
         Set<String> subserveTopics = Collections.emptySet();
         if (withMqtt) {
@@ -243,6 +259,11 @@ public class AddressController implements Watcher<Address> {
             notReady = Float.NaN;
 
         }
+
+        Set<Address> addressesWithForwarders = filterBy(addressSet, address -> (address.getStatus().getForwarderStatuses() != null && !address.getStatus().getForwarderStatuses().isEmpty()));
+        long readyForwarders = countForwardersReady(addressesWithForwarders, true);
+        long notReadyForwarders = countForwardersReady(addressesWithForwarders, false);
+
         metrics.reportMetric(new Metric("addresses_ready_total", "Total number of addresses in ready state", MetricType.gauge, new MetricValue(ready, now, metricLabels)));
         metrics.reportMetric(new Metric("addresses_not_ready_total", "Total number of address in a not ready state", MetricType.gauge, new MetricValue(notReady, now, metricLabels)));
         metrics.reportMetric(new Metric("addresses_total", "Total number of addresses", MetricType.gauge, new MetricValue(addressList.size(), now, metricLabels)));
@@ -253,6 +274,10 @@ public class AddressController implements Watcher<Address> {
         metrics.reportMetric(new Metric("addresses_configuring_total", "Total number of addresses in Configuring state", MetricType.gauge, new MetricValue(countByPhase.get(Configuring), now, metricLabels)));
         metrics.reportMetric(new Metric("addresses_active_total", "Total number of addresses in Active state", MetricType.gauge, new MetricValue(countByPhase.get(Active), now, metricLabels)));
 
+        metrics.reportMetric(new Metric("addresses_forwarders_total", "Total number of forwarders", MetricType.gauge, new MetricValue(readyForwarders + notReadyForwarders, now, metricLabels)));
+        metrics.reportMetric(new Metric("addresses_forwarders_ready_total", "Total number of forwarders in ready state", MetricType.gauge, new MetricValue(readyForwarders, now, metricLabels)));
+        metrics.reportMetric(new Metric("addresses_forwarders_not_ready_total", "Total number of forwarders in not ready state", MetricType.gauge, new MetricValue(notReadyForwarders, now, metricLabels)));
+
         long totalTime = gcTerminating - start;
         metrics.reportMetric(new Metric("standard_controller_loop_duration_seconds", "Time spent in controller loop", MetricType.gauge, new MetricValue((double) totalTime / 1_000_000_000.0, now, metricLabels)));
 
@@ -260,6 +285,18 @@ public class AddressController implements Watcher<Address> {
 
     }
 
+    private long countForwardersReady(Set<Address> addressesWithForwarders, boolean desired) {
+        long total = 0;
+        for (Address address : addressesWithForwarders) {
+            for (AddressStatusForwarder forwarder : address.getStatus().getForwarderStatuses()) {
+                if (forwarder.isReady() == desired) {
+                    total++;
+                }
+            }
+        }
+        return total;
+    }
+
     private Set<String> checkRegisteredSubserveTopics() {
         SubserveStatusCollector statusCollector = new SubserveStatusCollector(vertx, options.getCertDir());
 
@@ -489,9 +526,9 @@ public class AddressController implements Watcher<Address> {
         }
     }
 
-    private List<RouterStatus> checkRouterStatuses() throws Exception {
+    private List<RouterStatus> checkRouterStatuses(boolean checkRouterLinks) throws Exception {
 
-        RouterStatusCollector routerStatusCollector = new RouterStatusCollector(routerManagement);
+        RouterStatusCollector routerStatusCollector = new RouterStatusCollector(routerManagement, checkRouterLinks);
         List<RouterStatus> routerStatusList = new ArrayList<>();
         for (Pod router : kubernetes.listRouters()) {
             if (Readiness.isPodReady(router)) {
@@ -530,6 +567,10 @@ public class AddressController implements Watcher<Address> {
                         ok += routerStatus.checkAutoLinks(address);
                     }
                     ok += RouterStatus.checkActiveAutoLink(address, routerStatusList);
+                    ok += RouterStatus.checkForwarderLinks(address, routerStatusList);
+                    break;
+                case "subscription":
+                    ok += RouterStatus.checkForwarderLinks(address, routerStatusList);
                     break;
                 case "topic":
                     ok += checkBrokerStatus(address, clusterOk);
@@ -587,11 +628,11 @@ public class AddressController implements Watcher<Address> {
     }
 
     private class ProvisionState {
-        private final Status status;
+        private final AddressStatus status;
         private final String plan;
 
-        public ProvisionState(Status status, String plan) {
-            this.status = new Status(status);
+        public ProvisionState(AddressStatus status, String plan) {
+            this.status = new AddressStatus(status);
             this.plan = plan;
         }
 
diff --git a/standard-controller/src/main/java/io/enmasse/controller/standard/AddressProvisioner.java b/standard-controller/src/main/java/io/enmasse/controller/standard/AddressProvisioner.java
index 273143947..befc5fbca 100644
--- a/standard-controller/src/main/java/io/enmasse/controller/standard/AddressProvisioner.java
+++ b/standard-controller/src/main/java/io/enmasse/controller/standard/AddressProvisioner.java
@@ -143,7 +143,7 @@ public class AddressProvisioner {
             Map<String, Map<String, UsageInfo>> newUsageMap, Map<String, Double> limits) {
         for (Address address : pending) {
             if (!Phase.Configuring.equals(address.getStatus().getPhase())) {
-                Status previousStatus = new Status(address.getStatus());
+                AddressStatus previousStatus = new AddressStatus(address.getStatus());
 
                 Map<String, Map<String, UsageInfo>> neededMap = checkQuotaForAddress(limits, newUsageMap, address, all);
                 if (neededMap != null) {
@@ -254,7 +254,7 @@ public class AddressProvisioner {
 
     private Map<String, Map<String, UsageInfo>> checkQuotaForAddress(Map<String, Double> limits, Map<String, Map<String, UsageInfo>> usage, Address address, Set<Address> addressSet) {
         AddressPlan desiredPlan = addressResolver.getDesiredPlan(address);
-        AddressPlanStatus appliedPlan = Optional.ofNullable(address.getStatus()).map(Status::getPlanStatus).orElse(null);
+        AddressPlanStatus appliedPlan = Optional.ofNullable(address.getStatus()).map(AddressStatus::getPlanStatus).orElse(null);
 
         Map<String, Map<String, UsageInfo>> needed = copyUsageMap(usage);
 
diff --git a/standard-controller/src/main/java/io/enmasse/controller/standard/RouterStatus.java b/standard-controller/src/main/java/io/enmasse/controller/standard/RouterStatus.java
index 907bb7545..b82ab4f46 100644
--- a/standard-controller/src/main/java/io/enmasse/controller/standard/RouterStatus.java
+++ b/standard-controller/src/main/java/io/enmasse/controller/standard/RouterStatus.java
@@ -8,8 +8,7 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 
-import io.enmasse.address.model.Address;
-import io.enmasse.address.model.BrokerStatus;
+import io.enmasse.address.model.*;
 
 class RouterStatus {
     private final String routerId;
@@ -17,13 +16,15 @@ class RouterStatus {
     private final List<List<String>> autoLinks;
     private final List<List<String>> linkRoutes;
     private final List<String> connections;
+    private final List<List<String>> links;
 
-    RouterStatus(String routerId, List<String> addresses, List<List<String>> autoLinks, List<List<String>> linkRoutes, List<String> connections) {
+    RouterStatus(String routerId, List<String> addresses, List<List<String>> autoLinks, List<List<String>> linkRoutes, List<String> connections, List<List<String>> links) {
         this.routerId = routerId;
         this.addresses = addresses;
         this.autoLinks = autoLinks;
         this.linkRoutes = linkRoutes;
         this.connections = connections;
+        this.links = links;
     }
 
     public String getRouterId() {
@@ -160,6 +161,50 @@ class RouterStatus {
         return ok;
     }
 
+    public static int checkForwarderLinks(Address address, List<RouterStatus> routerStatusList) {
+        if (address.getSpec().getForwarders() == null || address.getSpec().getForwarders().isEmpty()) {
+            return 0;
+        }
+
+        int ok = 0;
+        for (AddressSpecForwarder forwarder : address.getSpec().getForwarders()) {
+            boolean found = false;
+            boolean isUp = false;
+            for (RouterStatus routerStatus : routerStatusList) {
+                for (List<String> entry : routerStatus.links) {
+                    String linkName = entry.get(0);
+                    if (linkName.equals(address.getForwarderLinkName(forwarder))) {
+                        found = true;
+                        String operStatus = entry.get(1);
+
+                        if ("up".equals(operStatus)) {
+                            isUp = true;
+                        }
+                        break;
+                    }
+                }
+            }
+
+            if (!found) {
+                updateForwarderStatus(forwarder.getName(), false, "Unable to find link for forwarder '" + forwarder.getName() + "'", address.getStatus().getForwarderStatuses());
+            } else if (!isUp) {
+                updateForwarderStatus(forwarder.getName(), false, "Unable to find link in the up state for forwarder '" + forwarder.getName() + "'", address.getStatus().getForwarderStatuses());
+            } else {
+                ok++;
+            }
+        }
+        return ok;
+    }
+
+    private static void updateForwarderStatus(String name, boolean isReady, String message, List<AddressStatusForwarder> forwarderStatuses) {
+        forwarderStatuses.stream()
+                .filter(c -> c.getName().equals(name))
+                .findFirst().ifPresent(s -> {
+            s.setReady(isReady);
+            s.appendMessage(message);
+        });
+    }
+
     @Override
     public String toString() {
         return new StringBuilder()
@@ -167,7 +212,8 @@ class RouterStatus {
                 .append("addresses=").append(addresses).append(",")
                 .append("autoLinks=").append(autoLinks).append(",")
                 .append("linkRoutes=").append(linkRoutes).append(",")
-                .append("connections=").append(connections).append("}")
+                .append("connections=").append(connections).append(",")
+                .append("links=").append(links).append("}")
                 .toString();
     }
 }
diff --git a/standard-controller/src/main/java/io/enmasse/controller/standard/RouterStatusCollector.java b/standard-controller/src/main/java/io/enmasse/controller/standard/RouterStatusCollector.java
index db3f3400e..aa57f2967 100644
--- a/standard-controller/src/main/java/io/enmasse/controller/standard/RouterStatusCollector.java
+++ b/standard-controller/src/main/java/io/enmasse/controller/standard/RouterStatusCollector.java
@@ -13,15 +13,18 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
 class RouterStatusCollector {
     private static final Logger log = LoggerFactory.getLogger(RouterStatusCollector.class);
     private final RouterManagement routerManagement;
+    private final boolean checkRouterLinks;
 
-    public RouterStatusCollector(RouterManagement routerManagement) {
+    public RouterStatusCollector(RouterManagement routerManagement, boolean checkRouterLinks) {
         this.routerManagement = routerManagement;
+        this.checkRouterLinks = checkRouterLinks;
     }
 
     public RouterStatus collect(Pod router) throws Exception {
@@ -50,25 +53,26 @@ class RouterStatusCollector {
     private static final RouterEntity linkRoute = new RouterEntity("org.apache.qpid.dispatch.router.config.linkRoute", "prefix", "containerId", "direction", "operStatus");
     private static final RouterEntity connection = new RouterEntity("org.apache.qpid.dispatch.connection", "container");
 
-    private static final RouterEntity[] entities = new RouterEntity[]{
-            address,
-            autoLink,
-            linkRoute,
-            connection
-    };
+    private static final RouterEntity link = new RouterEntity("org.apache.qpid.dispatch.router.link", "linkName", "operStatus", "linkDir");
 
     private RouterStatus doCollectStatus(Pod router, int port) throws Exception {
         String host = router.getStatus().getPodIP();
         log.debug("Checking router status of router : {}", router.getMetadata().getName());
 
-        Map<RouterEntity, List<List>> results = routerManagement.query(host, port, entities);
+        Map<RouterEntity, List<List>> results;
+        if (checkRouterLinks) {
+            results = routerManagement.query(host, port, address, autoLink, linkRoute, connection, link);
+        } else {
+            results = routerManagement.query(host, port, address, autoLink, linkRoute, connection);
+        }
 
         String routerId = router.getMetadata().getName();
         return new RouterStatus(routerId,
                 filterOnAttribute(String.class, 0, results.get(address)),
                 toTyped(String.class, results.get(autoLink)),
                 toTyped(String.class, results.get(linkRoute)),
-                filterOnAttribute(String.class, 0, results.get(connection)));
+                filterOnAttribute(String.class, 0, results.get(connection)),
+                toTyped(String.class, results.getOrDefault(link, Collections.emptyList())));
     }
 
     private static <T> List<List<T>> toTyped(Class<T> type, List<List> list) {


Test patch:
diff --git a/agent/testlib/mock_broker.js b/agent/testlib/mock_broker.js
index 01e82ab7d..3c90f0f53 100644
--- a/agent/testlib/mock_broker.js
+++ b/agent/testlib/mock_broker.js
@@ -193,7 +193,7 @@ function MockBroker (name) {
             if (self.objects.some(function (o) { return o.type === 'connector' && o.name === name; })) {
                 throw new Error('connector for ' + name + ' already exists!');
             } else {
-                self.add_connector(name, {name: name, source: params.sourceAddress, target: params.clientAddress});
+                self.add_connector(name, {name: name, source: params.sourceAddress, target: params.targetAddress});
             }
         },
         destroyConnectorService : function (name) {
diff --git a/api-model/src/test/java/io/enmasse/address/model/v1/address/AddressTest.java b/api-model/src/test/java/io/enmasse/address/model/v1/address/AddressTest.java
index a1f9989f1..b2f4cd75b 100644
--- a/api-model/src/test/java/io/enmasse/address/model/v1/address/AddressTest.java
+++ b/api-model/src/test/java/io/enmasse/address/model/v1/address/AddressTest.java
@@ -8,7 +8,7 @@ import io.enmasse.address.model.Address;
 import io.enmasse.address.model.AddressBuilder;
 import io.enmasse.address.model.AddressList;
 import io.enmasse.address.model.Phase;
-import io.enmasse.address.model.Status;
+import io.enmasse.address.model.AddressStatus;
 import org.junit.jupiter.api.Test;
 
 import static io.enmasse.address.model.validation.ValidationMatchers.isValid;
@@ -116,7 +116,7 @@ public class AddressTest {
                 .withAddressSpace("myspace")
                 .endSpec()
 
-                .withStatus(new Status(true).setPhase(Phase.Active).appendMessage("foo"))
+                .withStatus(new AddressStatus(true).setPhase(Phase.Active).appendMessage("foo"))
                 .build();
 
         Address b = new AddressBuilder(a).build();
@@ -144,7 +144,7 @@ public class AddressTest {
                 .withType("t1")
                 .endSpec()
 
-                .withStatus(new Status(true).setPhase(Phase.Active).appendMessage("foo"))
+                .withStatus(new AddressStatus(true).setPhase(Phase.Active).appendMessage("foo"))
                 .build();
 
         Address b = new AddressBuilder(a).build();
diff --git a/standard-controller/src/test/java/io/enmasse/controller/standard/AddressControllerTest.java b/standard-controller/src/test/java/io/enmasse/controller/standard/AddressControllerTest.java
index e3140cf25..055dea7ce 100644
--- a/standard-controller/src/test/java/io/enmasse/controller/standard/AddressControllerTest.java
+++ b/standard-controller/src/test/java/io/enmasse/controller/standard/AddressControllerTest.java
@@ -391,7 +391,7 @@ public class AddressControllerTest {
                 .withPlan("small-queue")
                 .endSpec()
 
-                .withStatus(new Status(true).setPhase(Phase.Active)
+                .withStatus(new AddressStatus(true).setPhase(Phase.Active)
                         .appendBrokerStatus(new BrokerStatus("broker-infra-0", "broker-infra-0-0", BrokerState.Migrating))
                         .appendBrokerStatus(new BrokerStatus("broker-infra-1", "broker-infra-1-0", BrokerState.Active)))
 
@@ -445,4 +445,37 @@ public class AddressControllerTest {
         assertEquals("broker-infra-1", alive.getStatus().getBrokerStatuses().get(0).getClusterId());
         assertEquals(BrokerState.Active, alive.getStatus().getBrokerStatuses().get(0).getState());
     }
+
+    @Test
+    public void testForwarderStatus() throws Exception {
+        Address a = new AddressBuilder()
+                .withNewMetadata()
+                .withName("myspace.a1")
+                .withNamespace("ns")
+                .endMetadata()
+
+                .withNewSpec()
+                .withAddress("q1")
+                .withType("queue")
+                .withPlan("small-queue")
+                .addNewForwarder()
+                .withName("fwd1")
+                .withRemoteAddress("remote1/queue1")
+                .withDirection(AddressSpecForwarderDirection.in)
+                .endForwarder()
+                .endSpec()
+                .withNewStatus()
+                .withReady(true)
+                .withPhase(Phase.Configuring)
+                .endStatus()
+                .build();
+
+
+        controller.onUpdate(Arrays.asList(a));
+        ArgumentCaptor<Address> captor = ArgumentCaptor.forClass(Address.class);
+        verify(mockApi).replaceAddress(captor.capture());
+        Address captured = captor.getValue();
+        assertEquals(captured.getStatus().getForwarderStatuses().size(), a.getSpec().getForwarders().size());
+        assertFalse(captured.getStatus().getForwarderStatuses().get(0).isReady());
+    }
 }
diff --git a/standard-controller/src/test/java/io/enmasse/controller/standard/AddressProvisionerTest.java b/standard-controller/src/test/java/io/enmasse/controller/standard/AddressProvisionerTest.java
index 1e2feff5c..ae0d82db1 100644
--- a/standard-controller/src/test/java/io/enmasse/controller/standard/AddressProvisionerTest.java
+++ b/standard-controller/src/test/java/io/enmasse/controller/standard/AddressProvisionerTest.java
@@ -35,22 +35,13 @@ import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import io.enmasse.address.model.*;
 import io.enmasse.admin.model.v1.StandardInfraConfig;
 import io.enmasse.admin.model.v1.StandardInfraConfigSpec;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.internal.util.collections.Sets;
 
-import io.enmasse.address.model.Address;
-import io.enmasse.address.model.AddressBuilder;
-import io.enmasse.address.model.AddressPlanStatus;
-import io.enmasse.address.model.AddressResolver;
-import io.enmasse.address.model.AddressSpaceResolver;
-import io.enmasse.address.model.BrokerState;
-import io.enmasse.address.model.BrokerStatus;
-import io.enmasse.address.model.BrokerStatusBuilder;
-import io.enmasse.address.model.Phase;
-import io.enmasse.address.model.StatusBuilder;
 import io.enmasse.admin.model.v1.ResourceAllowance;
 import io.enmasse.config.AnnotationKeys;
 import io.enmasse.k8s.api.EventLogger;
@@ -133,7 +124,7 @@ public class AddressProvisionerTest {
                 .withType("queue")
                 .endSpec()
 
-                .withStatus(new StatusBuilder()
+                .withStatus(new AddressStatusBuilder()
                         .withReady(true)
                         .addToBrokerStatuses(new BrokerStatus("broker-0", "broker-0-0").setState(BrokerState.Active))
                         .build())
diff --git a/systemtests/pom.xml b/systemtests/pom.xml
index fdbf87403..647344fd7 100644
--- a/systemtests/pom.xml
+++ b/systemtests/pom.xml
@@ -174,7 +174,7 @@
             <id>systemtests</id>
             <properties>
                 <skip.tests>false</skip.tests>
-                <groups>shared-standard | shared-brokered | shared-mqtt | isolated | marathon | shared-iot</groups>
+                <groups>(shared-standard | shared-brokered | shared-mqtt | isolated | marathon | shared-iot) &amp; !upgrade</groups>
             </properties>
         </profile>
         <profile>
@@ -182,14 +182,14 @@
             <properties>
                 <skip.tests>false</skip.tests>
                 <groups>(shared-standard | shared-brokered | shared-mqtt) &amp; !isolated &amp; !smoke &amp; !marathon
-                    &amp; !shared-iot &amp; !olm
+                    &amp; !shared-iot &amp; !olm &amp; !upgrade
                 </groups>
             </properties>
         </profile>
         <profile>
             <id>isolated</id>
             <properties>
-                <groups>(isolated | isolated-broker | isolated-standard) &amp; !marathon &amp; !shared-iot &amp; !olm</groups>
+                <groups>(isolated | isolated-broker | isolated-standard) &amp; !marathon &amp; !shared-iot &amp; !olm &amp; !upgrade</groups>
                 <skip.tests>false</skip.tests>
             </properties>
         </profile>
@@ -198,7 +198,7 @@
             <properties>
                 <skip.tests>false</skip.tests>
                 <groups>(shared-brokered | shared-standard | shared-mqtt) &amp; !nonPR &amp; !smoke &amp; !marathon
-                    &amp; !shared-iot &amp; !isolated &amp; !olm
+                    &amp; !shared-iot &amp; !isolated &amp; !olm &amp; !upgrade
                 </groups>
             </properties>
         </profile>
@@ -206,20 +206,20 @@
             <id>isolated-pr</id>
             <properties>
                 <skip.tests>false</skip.tests>
-                <groups>isolated &amp; !nonPR &amp; !smoke &amp; !marathon &amp; !shared-iot &amp; !olm</groups>
+                <groups>isolated &amp; !nonPR &amp; !smoke &amp; !marathon &amp; !shared-iot &amp; !olm &amp; !upgrade</groups>
             </properties>
         </profile>
         <profile>
             <id>shared-iot</id>
             <properties>
-                <groups>shared-iot &amp; !marathon &amp; !olm</groups>
+                <groups>shared-iot &amp; !marathon &amp; !olm &amp; !upgrade</groups>
                 <skip.tests>false</skip.tests>
             </properties>
         </profile>
         <profile>
             <id>smoke-iot</id>
             <properties>
-                <groups>shared-iot &amp; smoke &amp; !marathon &amp; !olm</groups>
+                <groups>shared-iot &amp; smoke &amp; !marathon &amp; !olm &amp; !upgrade</groups>
                 <skip.tests>false</skip.tests>
             </properties>
         </profile>
@@ -233,7 +233,7 @@
         <profile>
             <id>smoke</id>
             <properties>
-                <groups>smoke &amp; !shared-iot &amp; !marathon &amp; !olm</groups>
+                <groups>smoke &amp; !shared-iot &amp; !marathon &amp; !olm &amp; !upgrade</groups>
                 <skip.tests>false</skip.tests>
             </properties>
         </profile>
@@ -248,7 +248,7 @@
             <id>shared-release</id>
             <properties>
                 <groups>(shared-brokered | shared-standard | shared-mqtt) &amp; !noneAuth &amp; !smoke &amp; !marathon
-                    &amp; !shared-iot &amp; !isolated &amp; !olm
+                    &amp; !shared-iot &amp; !isolated &amp; !olm &amp; !upgrade
                 </groups>
                 <skip.tests>false</skip.tests>
             </properties>
@@ -256,14 +256,14 @@
         <profile>
             <id>isolated-release</id>
             <properties>
-                <groups>isolated &amp; !noneAuth &amp; !smoke &amp; !marathon &amp; !olm</groups>
+                <groups>isolated &amp; !noneAuth &amp; !smoke &amp; !marathon &amp; !olm &amp; !upgrade</groups>
                 <skip.tests>false</skip.tests>
             </properties>
         </profile>
         <profile>
             <id>iot-release</id>
             <properties>
-                <groups>shared-iot &amp; !noneAuth &amp; !smoke &amp; !marathon &amp; !olm</groups>
+                <groups>shared-iot &amp; !noneAuth &amp; !smoke &amp; !marathon &amp; !olm &amp; !upgrade</groups>
                 <skip.tests>false</skip.tests>
             </properties>
         </profile>
@@ -271,7 +271,7 @@
             <id>olm</id>
             <properties>
                 <groups>olm &amp; !shared-standard &amp; !shared-brokered &amp; !shared-mqtt &amp; !noneAuth &amp;
-                    !smoke &amp; !marathon &amp; !shared-iot &amp; !smoke-iot &amp; !isolated
+                    !smoke &amp; !marathon &amp; !shared-iot &amp; !smoke-iot &amp; !isolated &amp; !upgrade
                 </groups>
                 <skip.tests>false</skip.tests>
             </properties>
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/amqp/AmqpClientFactory.java b/systemtests/src/main/java/io/enmasse/systemtest/amqp/AmqpClientFactory.java
index 94d833bba..2618a931f 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/amqp/AmqpClientFactory.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/amqp/AmqpClientFactory.java
@@ -125,7 +125,7 @@ public class AmqpClientFactory {
         return createClient(connectOptions);
     }
 
-    protected AmqpClient createClient(AmqpConnectOptions connectOptions) {
+    public AmqpClient createClient(AmqpConnectOptions connectOptions) {
         AmqpClient client = new AmqpClient(connectOptions);
         clients.add(client);
         return client;
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/logs/GlobalLogCollector.java b/systemtests/src/main/java/io/enmasse/systemtest/logs/GlobalLogCollector.java
index 263b8866b..927420f30 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/logs/GlobalLogCollector.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/logs/GlobalLogCollector.java
@@ -205,13 +205,8 @@ public class GlobalLogCollector {
     private String[] enmasseRouterCmd(String cmd, String... extraArgs) {
         List<String> allArgs = new ArrayList<>();
         allArgs.add(cmd);
-        allArgs.add("--sasl-mechanisms=EXTERNAL");
-        allArgs.add("--ssl-certificate=/etc/enmasse-certs/tls.crt");
-        allArgs.add("--ssl-key=/etc/enmasse-certs/tls.key");
-        allArgs.add("--ssl-trustfile=/etc/enmasse-certs/ca.crt");
-        allArgs.add("--ssl-disable-peer-name-verify");
         allArgs.add("-b");
-        allArgs.add("127.0.0.1:55671");
+        allArgs.add("127.0.0.1:7777");
         allArgs.addAll(Arrays.asList(extraArgs));
         return allArgs.toArray(String[]::new);
     }
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/manager/CommonResourcesManager.java b/systemtests/src/main/java/io/enmasse/systemtest/manager/CommonResourcesManager.java
index 6fc958390..10f9e1d28 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/manager/CommonResourcesManager.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/manager/CommonResourcesManager.java
@@ -12,6 +12,7 @@ import io.enmasse.admin.model.v1.BrokeredInfraConfig;
 import io.enmasse.admin.model.v1.InfraConfig;
 import io.enmasse.admin.model.v1.StandardInfraConfig;
 import io.enmasse.systemtest.Environment;
+import io.enmasse.systemtest.UserCredentials;
 import io.enmasse.systemtest.amqp.AmqpClientFactory;
 import io.enmasse.systemtest.logs.CustomLogger;
 import io.enmasse.systemtest.mqtt.MqttClientFactory;
@@ -76,12 +77,20 @@ public class CommonResourcesManager extends ResourceManager {
         mqttClientFactory = new MqttClientFactory(addressSpace, defaultCredentials);
     }
 
+    public void initFactories(AddressSpace addressSpace, UserCredentials userCredentials) throws Exception {
+        closeClientFactories(amqpClientFactory, mqttClientFactory);
+        amqpClientFactory = new AmqpClientFactory(addressSpace, userCredentials);
+        mqttClientFactory = new MqttClientFactory(addressSpace, userCredentials);
+    }
+
     @Override
     public void setup() {
-        if (!reuseAddressSpace) {
-            currentAddressSpaces = new ArrayList<>();
+        if (currentAddressSpaces.size() != 0) {
+            initFactories(currentAddressSpaces.get(0));
+        } else {
             initFactories(null);
         }
+
     }
 
     @Override
@@ -94,30 +103,34 @@ public class CommonResourcesManager extends ResourceManager {
                 Kubernetes.getInstance().getAddressSpacePlanClient().withName(addressSpacePlan.getMetadata().getName()).cascading(true).delete();
                 LOGGER.info("AddressSpace plan {} deleted", addressSpacePlan.getMetadata().getName());
             }
+            addressSpacePlans.clear();
 
             for (AddressPlan addressPlan : addressPlans) {
                 Kubernetes.getInstance().getAddressPlanClient().withName(addressPlan.getMetadata().getName()).cascading(true).delete();
                 LOGGER.info("Address plan {} deleted", addressPlan.getMetadata().getName());
             }
+            addressPlans.clear();
 
             for (StandardInfraConfig infraConfigDefinition : standardInfraConfigs) {
                 Kubernetes.getInstance().getStandardInfraConfigClient().withName(infraConfigDefinition.getMetadata().getName()).cascading(true).delete();
                 LOGGER.info("Standardinfraconfig {} deleted", infraConfigDefinition.getMetadata().getName());
             }
+            standardInfraConfigs.clear();
 
             for (BrokeredInfraConfig infraConfigDefinition : brokeredInfraConfigs) {
                 Kubernetes.getInstance().getBrokeredInfraConfigClient().withName(infraConfigDefinition.getMetadata().getName()).cascading(true).delete();
                 LOGGER.info("Brokeredinfraconfig {} deleted", infraConfigDefinition.getMetadata().getName());
             }
+            brokeredInfraConfigs.clear();
 
             for (AuthenticationService authService : authServices) {
                 Kubernetes.getInstance().getAuthenticationServiceClient().withName(authService.getMetadata().getName()).cascading(true).delete();
                 TestUtils.waitForNReplicas(0, false, Map.of("name", authService.getMetadata().getName()), Collections.emptyMap(), new TimeoutBudget(1, TimeUnit.MINUTES), 5000);
                 LOGGER.info("AuthService {} deleted", authService.getMetadata().getName());
             }
+            authServices.clear();
+
 
-            addressPlans.clear();
-            addressSpacePlans.clear();
             closeClientFactories(amqpClientFactory, mqttClientFactory);
             amqpClientFactory = null;
             mqttClientFactory = null;
@@ -131,8 +144,8 @@ public class CommonResourcesManager extends ResourceManager {
     //------------------------------------------------------------------------------------------------
 
     public void createAddressPlan(AddressPlan addressPlan) throws Exception {
-        createAddressPlan(addressPlan, false);
         addressPlans.add(addressPlan);
+        super.createAddressPlan(addressPlan);
     }
 
     public void removeAddressPlan(AddressPlan addressPlan) throws Exception {
@@ -140,8 +153,8 @@ public class CommonResourcesManager extends ResourceManager {
         addressPlans.removeIf(addressPlanIter -> addressPlanIter.getMetadata().getName().equals(addressPlan.getMetadata().getName()));
     }
 
-    public void replaceAddressPlan(AddressPlan plan) throws Exception {
-        super.createAddressPlan(plan, true);
+    public void replaceAddressPlan(AddressPlan plan) throws InterruptedException {
+        super.replaceAddressPlan(plan);
     }
 
     public AddressPlan getAddressPlan(String name) throws Exception {
@@ -153,8 +166,8 @@ public class CommonResourcesManager extends ResourceManager {
     //------------------------------------------------------------------------------------------------
 
     public void createAddressSpacePlan(AddressSpacePlan addressSpacePlan) throws Exception {
-        super.createAddressSpacePlan(addressSpacePlan);
         addressSpacePlans.add(addressSpacePlan);
+        super.createAddressSpacePlan(addressSpacePlan);
     }
 
     public void removeAddressSpacePlan(AddressSpacePlan addressSpacePlan) throws Exception {
@@ -180,11 +193,11 @@ public class CommonResourcesManager extends ResourceManager {
 
     public void createInfraConfig(InfraConfig infraConfigDefinition) throws Exception {
         if (infraConfigDefinition instanceof StandardInfraConfig) {
-            super.createInfraConfig((StandardInfraConfig) infraConfigDefinition);
             standardInfraConfigs.add((StandardInfraConfig) infraConfigDefinition);
+            super.createInfraConfig((StandardInfraConfig) infraConfigDefinition);
         } else {
-            super.createInfraConfig((BrokeredInfraConfig) infraConfigDefinition);
             brokeredInfraConfigs.add((BrokeredInfraConfig) infraConfigDefinition);
+            super.createInfraConfig((BrokeredInfraConfig) infraConfigDefinition);
         }
     }
 
@@ -219,6 +232,12 @@ public class CommonResourcesManager extends ResourceManager {
         }
     }
 
+    public void createAuthService(AuthenticationService authenticationService) throws Exception {
+        authServices.add(authenticationService);
+        super.createAuthService(authenticationService);
+    }
+
+
     public void removeAuthService(AuthenticationService authService) throws Exception {
         super.removeAuthService(authService);
         authServices.removeIf(authserviceId -> authserviceId.getMetadata().getName().equals(authService.getMetadata().getName()));
@@ -226,8 +245,8 @@ public class CommonResourcesManager extends ResourceManager {
 
     public void createAddressSpace(AddressSpace addressSpace) throws Exception {
         if (!AddressSpaceUtils.existAddressSpace(addressSpace.getMetadata().getNamespace(), addressSpace.getMetadata().getName())) {
-            super.createAddressSpace(addressSpace);
             currentAddressSpaces.add(addressSpace);
+            super.createAddressSpace(addressSpace);
         } else {
             super.waitForAddressSpaceReady(addressSpace);
         }
@@ -252,7 +271,6 @@ public class CommonResourcesManager extends ResourceManager {
             if (!AddressSpaceUtils.existAddressSpace(addressSpace.getMetadata().getNamespace(), addressSpace.getMetadata().getName())) {
                 LOGGER.info("Address space '" + addressSpace + "' doesn't exist and will be created.");
                 spaces.add(addressSpace);
-                currentAddressSpaces.add(addressSpace);
             } else {
                 LOGGER.warn("Address space '" + addressSpace + "' already exists.");
             }
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 15e09f67f..df7da53aa 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/manager/ResourceManager.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/manager/ResourceManager.java
@@ -90,23 +90,22 @@ public abstract class ResourceManager {
     // Address plans
     //------------------------------------------------------------------------------------------------
 
-    public void createAddressPlan(AddressPlan addressPlan, boolean replaceExisting) throws Exception {
+    public void createAddressPlan(AddressPlan addressPlan) throws Exception {
         LOGGER.info("Address plan {} will be created {}", addressPlan.getMetadata().getName(), addressPlan);
         var client = Kubernetes.getInstance().getAddressPlanClient();
-        if (replaceExisting) {
-            client.createOrReplace(addressPlan);
-        } else {
-            client.create(addressPlan);
-        }
+        client.create(addressPlan);
         Thread.sleep(1000);
     }
 
-    public void removeAddressPlan(AddressPlan addressPlan) throws Exception {
-        Kubernetes.getInstance().getAddressPlanClient().withName(addressPlan.getMetadata().getName()).cascading(true).delete();
+    public void replaceAddressPlan(AddressPlan addressPlan) throws InterruptedException {
+        LOGGER.info("Address plan {} will be replaced {}", addressPlan.getMetadata().getName(), addressPlan);
+        var client = Kubernetes.getInstance().getAddressPlanClient();
+        client.createOrReplace(addressPlan);
+        Thread.sleep(1000);
     }
 
-    public void replaceAddressPlan(AddressPlan plan) throws Exception {
-        Kubernetes.getInstance().getAddressPlanClient().createOrReplace(plan);
+    public void removeAddressPlan(AddressPlan addressPlan) throws Exception {
+        Kubernetes.getInstance().getAddressPlanClient().withName(addressPlan.getMetadata().getName()).cascading(true).delete();
     }
 
     public AddressPlan getAddressPlan(String name) throws Exception {
@@ -118,17 +117,9 @@ public abstract class ResourceManager {
     //------------------------------------------------------------------------------------------------
 
     public void createAddressSpacePlan(AddressSpacePlan addressSpacePlan) throws Exception {
-        createAddressSpacePlan(addressSpacePlan, false);
-    }
-
-    public void createAddressSpacePlan(AddressSpacePlan addressSpacePlan, boolean replaceExisting) throws Exception {
         LOGGER.info("AddressSpace plan {} will be created {}", addressSpacePlan.getMetadata().getName(), addressSpacePlan);
         var client = Kubernetes.getInstance().getAddressSpacePlanClient();
-        if (replaceExisting) {
-            client.createOrReplace(addressSpacePlan);
-        } else {
-            client.create(addressSpacePlan);
-        }
+        client.create(addressSpacePlan);
         Thread.sleep(1000);
     }
 
@@ -183,8 +174,8 @@ public abstract class ResourceManager {
     }
 
     public void createAuthService(AuthenticationService authenticationService) throws Exception {
-        LOGGER.info("AuthService {} will be created {}", authenticationService.getMetadata().getName(), authenticationService);
         var client = Kubernetes.getInstance().getAuthenticationServiceClient();
+        LOGGER.info("AuthService {} will be created {}", authenticationService.getMetadata().getName(), authenticationService);
         client.create(authenticationService);
         waitForAuthPods(authenticationService);
     }
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 a1ed69e65..6be62f34f 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/platform/Kubernetes.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/platform/Kubernetes.java
@@ -578,14 +578,13 @@ public abstract class Kubernetes {
      * @param resources service resource
      * @return endpoint of new service
      */
-    public Endpoint createServiceFromResource(String namespace, Service resources) {
+    public void createServiceFromResource(String namespace, Service resources) {
         if (!serviceExists(namespace, resources.getMetadata().getName())) {
             Service serRes = client.services().inNamespace(namespace).create(resources);
             log.info("Service {} created", serRes.getMetadata().getName());
         } else {
             log.info("Service {} already exists", resources.getMetadata().getName());
         }
-        return getEndpoint(resources.getMetadata().getName(), namespace, "http");
     }
 
     /**
@@ -914,4 +913,5 @@ public abstract class Kubernetes {
     public CustomResourceDefinition getCRD(String name) {
         return client.customResourceDefinitions().withName(name).get();
     }
+
 }
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/platform/OpenShift.java b/systemtests/src/main/java/io/enmasse/systemtest/platform/OpenShift.java
index 2f80906bb..2aef89b14 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/platform/OpenShift.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/platform/OpenShift.java
@@ -109,4 +109,5 @@ public class OpenShift extends Kubernetes {
             return getEndpoint(endpointName, namespace, port);
         }
     }
+
 }
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 02086634e..f35408935 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
@@ -14,6 +14,7 @@ import io.enmasse.systemtest.time.TimeoutBudget;
 import io.enmasse.systemtest.utils.TestUtils;
 import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
+import io.fabric8.kubernetes.api.model.ContainerPortBuilder;
 import io.fabric8.kubernetes.api.model.EnvFromSourceBuilder;
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.IntOrString;
@@ -70,6 +71,7 @@ public class SystemtestsKubernetesApps {
     public static final String POSTGRES_APP = "postgres-app";
     public static final String INFINISPAN_SERVER = "infinispan-server";
     private static final Path INFINISPAN_EXAMPLE_BASE = Paths.get("../templates/iot/examples/infinispan");
+    private static final String AMQ_BROKER = "amq-broker";
 
     public static String getMessagingAppPodName() throws Exception {
         TestUtils.waitUntilCondition("Pod is reachable", waitPhase -> Kubernetes.getInstance().listPods(MESSAGING_PROJECT).stream().filter(pod -> pod.getMetadata().getName().contains(MESSAGING_CLIENTS) &&
@@ -237,6 +239,34 @@ public class SystemtestsKubernetesApps {
                 INFINISPAN_EXAMPLE_BASE.resolve("manual"));
     }
 
+    public static Endpoint getAMQBrokerEndpoint(String namespace) {
+        return Kubernetes.getInstance().getEndpoint("amqp", namespace, "amqp");
+    }
+
+    public static void deployAMQBroker(String namespace, String user, String password, String... queues) throws Exception {
+        Kubernetes kubeCli = Kubernetes.getInstance();
+        if (!kubeCli.namespaceExists(namespace)) {
+            kubeCli.createNamespace(namespace);
+        }
+
+        kubeCli.createDeploymentFromResource(namespace, getBrokerDeployment(user, password, queues));
+        kubeCli.createServiceFromResource(namespace, getSystemtestsServiceResource("amqp", AMQ_BROKER, 5672, "amqp"));
+
+        kubeCli.getClient()
+            .apps().deployments()
+            .inNamespace(namespace)
+            .withName(AMQ_BROKER)
+            .waitUntilReady(5, TimeUnit.MINUTES);
+
+        Thread.sleep(5000);
+    }
+
+    public static void deleteAMQBroker(String namespace) throws Exception {
+        Kubernetes kubeCli = Kubernetes.getInstance();
+        kubeCli.deleteService(namespace, "amqp");
+        kubeCli.deleteDeployment(namespace, AMQ_BROKER);
+        Thread.sleep(5000);
+    }
 
     public static void applyDirectories(final Function<InputStream, InputStream> streamManipulator, final Path... paths) throws Exception {
         loadDirectories(streamManipulator, Applicable::createOrReplace, paths);
@@ -389,6 +419,23 @@ public class SystemtestsKubernetesApps {
                 .build();
     }
 
+    private static Service getSystemtestsServiceResource(String serviceName, String appName, int port, String portName) {
+        return new ServiceBuilder()
+                .withNewMetadata()
+                .withName(serviceName)
+                .addToLabels("run", appName)
+                .endMetadata()
+                .withNewSpec()
+                .addToSelector("app", appName)
+                .addNewPort()
+                .withName(portName)
+                .withPort(port)
+                .withTargetPort(new IntOrString(port))
+                .endPort()
+                .endSpec()
+                .build();
+    }
+
     private static Ingress getSystemtestsIngressResource(String appName, int port) throws Exception {
         Environment env = Environment.getInstance();
         IngressBackend backend = new IngressBackend();
@@ -547,4 +594,57 @@ public class SystemtestsKubernetesApps {
                 .build();
     }
 
-}
+    private static Deployment getBrokerDeployment(String user, String password, String... queues) {
+        return new DeploymentBuilder()
+                .withNewMetadata()
+                .withName(AMQ_BROKER)
+                .addToLabels("app", AMQ_BROKER)
+                .addToLabels("template", AMQ_BROKER)
+                .endMetadata()
+                .withNewSpec()
+                .withNewSelector()
+                .addToMatchLabels("app", AMQ_BROKER)
+                .endSelector()
+                .withReplicas(1)
+                .withNewTemplate()
+                .withNewMetadata()
+                .addToLabels("app", AMQ_BROKER)
+                .endMetadata()
+                .withNewSpec()
+                .addNewInitContainer()
+                    .withName("artemis-init")
+                    .withImage("quay.io/enmasse/artemis-base:2.10.0")
+                    .withCommand("/bin/sh")
+                    .withArgs("-c", "/opt/apache-artemis/bin/artemis create /var/run/artemis --allow-anonymous --force --user "+user+" --password "+password+" --role admin --queues "+String.join(",", queues))
+                    .withVolumeMounts(new VolumeMountBuilder()
+                            .withName("data")
+                            .withMountPath("/var/run/artemis")
+                            .build())
+                .endInitContainer()
+                .addNewContainer()
+                    .withName(AMQ_BROKER)
+                    .withImage("quay.io/enmasse/artemis-base:2.10.0")
+                    .withImagePullPolicy("IfNotPresent")
+                    .withCommand("/bin/sh")
+                    .withArgs("-c", "/var/run/artemis/bin/artemis run")
+                    .addToPorts(new ContainerPortBuilder()
+                            .withContainerPort(5672)
+                            .withName("amqp")
+                            .build())
+                    .withVolumeMounts(new VolumeMountBuilder()
+                            .withName("data")
+                            .withMountPath("/var/run/artemis")
+                            .build())
+                    .endContainer()
+                .addNewVolume()
+                .withName("data")
+                .withNewEmptyDir()
+                .endEmptyDir()
+                .endVolume()
+                .endSpec()
+                .endTemplate()
+                .endSpec()
+                .build();
+    }
+
+}
\ No newline at end of file
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 c61c8334d..b72dd7ff9 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/utils/AddressSpaceUtils.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/utils/AddressSpaceUtils.java
@@ -5,7 +5,10 @@
 package io.enmasse.systemtest.utils;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+
 import io.enmasse.address.model.AddressSpace;
+import io.enmasse.address.model.AddressSpaceStatus;
+import io.enmasse.address.model.AddressSpaceStatusConnector;
 import io.enmasse.address.model.EndpointSpec;
 import io.enmasse.address.model.EndpointStatus;
 import io.enmasse.systemtest.Endpoint;
@@ -19,12 +22,15 @@ import io.enmasse.systemtest.time.TimeoutBudget;
 import io.vertx.core.json.JsonObject;
 import org.slf4j.Logger;
 
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 public class AddressSpaceUtils {
     private static Logger log = CustomLogger.getLogger();
@@ -223,4 +229,60 @@ public class AddressSpaceUtils {
     public boolean isBrokered(AddressSpace addressSpace) {
         return addressSpace.getSpec().getType().equals(AddressSpaceType.BROKERED.toString());
     }
+
+    public static AddressSpace waitForAddressSpaceConnectorsReady(AddressSpace addressSpace) throws Exception {
+        return waitForAddressSpaceConnectorsReady(addressSpace, new TimeoutBudget(5, TimeUnit.MINUTES));
+    }
+
+    public static AddressSpace waitForAddressSpaceConnectorsReady(AddressSpace addressSpace, TimeoutBudget budget) throws Exception {
+        waitForAddressSpaceReady(addressSpace);
+        boolean isReady = false;
+        var client = Kubernetes.getInstance().getAddressSpaceClient(addressSpace.getMetadata().getNamespace());
+
+        String name = addressSpace.getMetadata().getName();
+        AddressSpace clientAddressSpace = addressSpace;
+        while (budget.timeLeft() >= 0 && !isReady) {
+            clientAddressSpace = client.withName(name).get();
+            isReady = areAddressSpaceConnectorsReady(clientAddressSpace);
+            if (!isReady) {
+                Thread.sleep(10000);
+            }
+            log.info("Waiting until connectors of address space: '{}' messages {} will be in ready state", name, getConnectorStatuses(clientAddressSpace));
+        }
+        clientAddressSpace = client.withName(name).get();
+        isReady = areAddressSpaceConnectorsReady(clientAddressSpace);
+        if (!isReady) {
+            throw new IllegalStateException(String.format("Connectors of Address Space %s are not in Ready state within timeout: %s", name, getConnectorStatuses(clientAddressSpace)));
+        }
+        log.info("Connectors of address space {} are ready for use", name);
+        return clientAddressSpace;
+    }
+
+    /**
+     * Returns true only if all connectorStatuses report isReady=true
+     * @param addressSpace
+     * @return
+     */
+    public static boolean areAddressSpaceConnectorsReady(AddressSpace addressSpace) {
+        return Optional.ofNullable(addressSpace)
+            .map(AddressSpace::getStatus)
+            .map(AddressSpaceStatus::getConnectorStatuses)
+            .map(Stream::of)
+            .orElseGet(Stream::empty)
+            .flatMap(Collection::stream)
+            .map(AddressSpaceStatusConnector::isReady)
+            .allMatch(ready -> ready == true);
+    }
+
+    public static String getConnectorStatuses(AddressSpace addressSpace) {
+        return Optional.ofNullable(addressSpace)
+            .map(AddressSpace::getStatus)
+            .map(AddressSpaceStatus::getConnectorStatuses)
+            .map(Stream::of)
+            .orElseGet(Stream::empty)
+            .flatMap(Collection::stream)
+            .collect(Collectors.toList())
+            .toString();
+    }
+
 }
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/watcher/TestWatcher.java b/systemtests/src/main/java/io/enmasse/systemtest/watcher/TestWatcher.java
index 261692afb..a285426c2 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/watcher/TestWatcher.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/watcher/TestWatcher.java
@@ -37,13 +37,9 @@ public class TestWatcher implements TestExecutionExceptionHandler, LifecycleMeth
 
     @Override
     public void afterTestExecution(ExtensionContext extensionContext) throws Exception {
-      LOGGER.info("Teardown section: ");
-        if ((testInfo.getTests().size() != 1) && testInfo.getCurrentTestIndex() != testInfo.getTests().size() - 1) {
-            if (testInfo.isTestShared() && sharedResourcesManager.getSharedAddressSpace() != null) {
-                tearDownSharedResources();
-            } else {
-                tearDownCommonResources();
-            }
+        LOGGER.info("Teardown section: ");
+        if (testInfo.isTestShared() && sharedResourcesManager.getSharedAddressSpace() != null) {
+            tearDownSharedResources();
         } else {
             tearDownCommonResources();
         }
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/bases/common/PersistentMessagesTestBase.java b/systemtests/src/test/java/io/enmasse/systemtest/bases/common/PersistentMessagesTestBase.java
deleted file mode 100644
index 6d0f72226..000000000
--- a/systemtests/src/test/java/io/enmasse/systemtest/bases/common/PersistentMessagesTestBase.java
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * 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.bases.common;
-
-import io.enmasse.address.model.Address;
-import io.enmasse.address.model.AddressSpace;
-import io.enmasse.systemtest.UserCredentials;
-import io.enmasse.systemtest.amqp.AmqpClient;
-import io.enmasse.systemtest.bases.TestBase;
-import io.enmasse.systemtest.bases.shared.ITestBaseShared;
-import io.enmasse.systemtest.clients.ClientUtils;
-import io.enmasse.systemtest.logs.CustomLogger;
-import io.enmasse.systemtest.time.TimeoutBudget;
-import io.enmasse.systemtest.utils.AddressUtils;
-import io.enmasse.systemtest.utils.TestUtils;
-import org.apache.qpid.proton.message.Message;
-import org.slf4j.Logger;
-
-import java.io.IOException;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.junit.jupiter.api.Assertions.fail;
-
-public class PersistentMessagesTestBase extends TestBase implements ITestBaseShared {
-
-    private static Logger log = CustomLogger.getLogger();
-    private ClientUtils clientUtils = new ClientUtils();
-
-    protected void doTestQueuePersistentMessages(Address address, int messagesBatch) throws Exception {
-        resourcesManager.setAddresses(address);
-
-        // Wait for the first console pod to be terminated
-        TestUtils.waitForConsoleRollingUpdate(kubernetes.getInfraNamespace());
-        TestUtils.waitUntilDeployed(kubernetes.getInfraNamespace());
-
-        int podCount = kubernetes.listPods().size();
-        clientUtils.sendDurableMessages(resourcesManager, getSharedAddressSpace(), address, defaultCredentials, messagesBatch);
-        restartBrokers(podCount);
-
-        // Seems that the service/route can sometimes not be immediately available despite the pod being Ready.
-        assertConnectable(getSharedAddressSpace(), defaultCredentials);
-        clientUtils.receiveDurableMessages(resourcesManager, getSharedAddressSpace(), address, defaultCredentials, messagesBatch);
-
-    }
-
-    protected void doTestTopicPersistentMessages(Address topic, Address subscription) throws Exception {
-        resourcesManager.setAddresses(topic, subscription);
-
-        int podCount = kubernetes.listPods().size();
-
-        AmqpClient client = resourcesManager.getAmqpClientFactory().createTopicClient(getSharedAddressSpace());
-        client.getConnectOptions().setCredentials(defaultCredentials);
-
-        log.info("Subscribe first receiver");
-        Future<List<Message>> recvResults = client.recvMessages(AddressUtils.getQualifiedSubscriptionAddress(subscription), 30);
-        clientUtils.sendDurableMessages(resourcesManager, getSharedAddressSpace(), topic, defaultCredentials, 30);
-        assertThat("Wrong messages received: ", recvResults.get(1, TimeUnit.MINUTES).size(), is(30));
-        clientUtils.sendDurableMessages(resourcesManager, getSharedAddressSpace(), topic, defaultCredentials, 30);
-
-        restartBrokers(podCount);
-
-        // Seems that the service/route can sometimes not be immediately available despite the pod being Ready.
-        assertConnectable(getSharedAddressSpace(), defaultCredentials);
-
-        log.info("Subscribe receiver again");
-        recvResults = client.recvMessages(AddressUtils.getQualifiedSubscriptionAddress(subscription), 30);
-        assertThat("Wrong messages received: batch2", recvResults.get(1, TimeUnit.MINUTES).size(), is(30));
-    }
-
-    private void restartBrokers(int podCount) throws Exception {
-        Map<String, String> labelSelector = Collections.singletonMap("role", "broker");
-        List<String> uids = kubernetes.listPods(kubernetes.getInfraNamespace(), labelSelector).stream().map(pod -> pod.getMetadata().getUid()).collect(Collectors.toList());
-
-        kubernetes.deletePod(kubernetes.getInfraNamespace(), labelSelector);
-
-        waitForPodsToTerminate(uids);
-        TestUtils.waitForExpectedReadyPods(kubernetes, kubernetes.getInfraNamespace(), podCount, new TimeoutBudget(10, TimeUnit.MINUTES));
-        log.info("Broker pods restarted");
-    }
-
-    private void assertConnectable(AddressSpace space, UserCredentials user) throws Exception {
-        TimeoutBudget budget = new TimeoutBudget(1, TimeUnit.MINUTES);
-        String name = space.getMetadata().getName();
-        do {
-            try {
-                clientUtils.connectAddressSpace(resourcesManager, space, user);
-                log.info("Successfully connected to address space : {}", name);
-                return;
-            } catch (IOException e) {
-                log.info("Failed to connect to address space: {} - {}", name, e.getMessage());
-            }
-            Thread.sleep(1000);
-        } while (!budget.timeoutExpired());
-
-        fail(String.format("Failed to assert address space %s connectable within timeout", name));
-    }
-
-}
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/isolated/PersistentMessagesTest.java b/systemtests/src/test/java/io/enmasse/systemtest/isolated/PersistentMessagesTest.java
new file mode 100644
index 000000000..b289602ff
--- /dev/null
+++ b/systemtests/src/test/java/io/enmasse/systemtest/isolated/PersistentMessagesTest.java
@@ -0,0 +1,229 @@
+/*
+ * 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.isolated;
+
+import io.enmasse.address.model.Address;
+import io.enmasse.address.model.AddressBuilder;
+import io.enmasse.address.model.AddressSpace;
+import io.enmasse.address.model.AddressSpaceBuilder;
+import io.enmasse.systemtest.UserCredentials;
+import io.enmasse.systemtest.amqp.AmqpClient;
+import io.enmasse.systemtest.bases.TestBase;
+import io.enmasse.systemtest.bases.isolated.ITestBaseIsolated;
+import io.enmasse.systemtest.clients.ClientUtils;
+import io.enmasse.systemtest.logs.CustomLogger;
+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.AddressUtils;
+import io.enmasse.systemtest.utils.TestUtils;
+import org.apache.qpid.proton.message.Message;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.fail;
+
+class PersistentMessagesTest extends TestBase implements ITestBaseIsolated {
+    private static Logger log = CustomLogger.getLogger();
+    private ClientUtils clientUtils = new ClientUtils();
+    private UserCredentials credentials = new UserCredentials("test", "test");
+
+    @Test
+    void testBrokeredPersistentMessages() throws Exception {
+        AddressSpace brokered = new AddressSpaceBuilder()
+                .withNewMetadata()
+                .withName("brokered-persistent")
+                .withNamespace(kubernetes.getInfraNamespace())
+                .endMetadata()
+                .withNewSpec()
+                .withType(AddressSpaceType.BROKERED.toString())
+                .withPlan(AddressSpacePlans.BROKERED)
+                .withNewAuthenticationService()
+                .withName("standard-authservice")
+                .endAuthenticationService()
+                .endSpec()
+                .build();
+        resourcesManager.createAddressSpace(brokered);
+        resourcesManager.createOrUpdateUser(brokered, credentials);
+
+
+        Address brokeredQueue = new AddressBuilder()
+                .withNewMetadata()
+                .withNamespace(brokered.getMetadata().getNamespace())
+                .withName(AddressUtils.generateAddressMetadataName(brokered, "test-queue-brokered"))
+                .endMetadata()
+                .withNewSpec()
+                .withType("queue")
+                .withAddress("test-queue-brokered")
+                .withPlan(DestinationPlan.BROKERED_QUEUE)
+                .endSpec()
+                .build();
+        doTestQueuePersistentMessages(brokered, brokeredQueue, 100);
+    }
+
+    @Test
+    void testStandardPersistentMessages() throws Exception {
+        AddressSpace standard = new AddressSpaceBuilder()
+                .withNewMetadata()
+                .withName("standard-persistent")
+                .withNamespace(kubernetes.getInfraNamespace())
+                .endMetadata()
+                .withNewSpec()
+                .withType(AddressSpaceType.STANDARD.toString())
+                .withPlan(AddressSpacePlans.STANDARD_UNLIMITED)
+                .withNewAuthenticationService()
+                .withName("standard-authservice")
+                .endAuthenticationService()
+                .endSpec()
+                .build();
+        resourcesManager.createAddressSpace(standard);
+        resourcesManager.createOrUpdateUser(standard, credentials);
+
+        Address standardQueue = new AddressBuilder()
+                .withNewMetadata()
+                .withNamespace(standard.getMetadata().getNamespace())
+                .withName(AddressUtils.generateAddressMetadataName(standard, "test-queue-standard"))
+                .endMetadata()
+                .withNewSpec()
+                .withType("queue")
+                .withAddress("test-queue-standard")
+                .withPlan(DestinationPlan.STANDARD_SMALL_QUEUE)
+                .endSpec()
+                .build();
+        doTestQueuePersistentMessages(standard, standardQueue, 30);
+
+        Address standardLargeQueue = new AddressBuilder()
+                .withNewMetadata()
+                .withNamespace(standard.getMetadata().getNamespace())
+                .withName(AddressUtils.generateAddressMetadataName(standard, "test-large-queue-standard"))
+                .endMetadata()
+                .withNewSpec()
+                .withType("queue")
+                .withAddress("test-large-queue-standard")
+                .withPlan(DestinationPlan.STANDARD_LARGE_QUEUE)
+                .endSpec()
+                .build();
+        doTestQueuePersistentMessages(standard, standardLargeQueue, 30);
+
+        Address standardXLargeQueue = new AddressBuilder()
+                .withNewMetadata()
+                .withNamespace(standard.getMetadata().getNamespace())
+                .withName(AddressUtils.generateAddressMetadataName(standard, "test-xlarge-queue-standard"))
+                .endMetadata()
+                .withNewSpec()
+                .withType("queue")
+                .withAddress("test-xlarge-queue-standard")
+                .withPlan(DestinationPlan.STANDARD_XLARGE_QUEUE)
+                .endSpec()
+                .build();
+        doTestQueuePersistentMessages(standard, standardXLargeQueue, 30);
+
+        Address standardTopic = new AddressBuilder()
+                .withNewMetadata()
+                .withNamespace(standard.getMetadata().getNamespace())
+                .withName(AddressUtils.generateAddressMetadataName(standard, "test-topic-standard"))
+                .endMetadata()
+                .withNewSpec()
+                .withType("topic")
+                .withAddress("test-topic-standard")
+                .withPlan(DestinationPlan.STANDARD_LARGE_TOPIC)
+                .endSpec()
+                .build();
+        Address standardSub = new AddressBuilder()
+                .withNewMetadata()
+                .withNamespace(standard.getMetadata().getNamespace())
+                .withName(AddressUtils.generateAddressMetadataName(standard, "test-sub-standard"))
+                .endMetadata()
+                .withNewSpec()
+                .withType("subscription")
+                .withAddress("test-sub-standard")
+                .withTopic(standardTopic.getSpec().getAddress())
+                .withPlan(DestinationPlan.STANDARD_SMALL_SUBSCRIPTION)
+                .endSpec()
+                .build();
+        doTestTopicPersistentMessages(standard, standardTopic, standardSub);
+    }
+
+    private void doTestQueuePersistentMessages(AddressSpace addSpace, Address address, int messagesBatch) throws Exception {
+        resourcesManager.setAddresses(address);
+
+        // Wait for the first console pod to be terminated
+        TestUtils.waitForConsoleRollingUpdate(kubernetes.getInfraNamespace());
+        TestUtils.waitUntilDeployed(kubernetes.getInfraNamespace());
+
+        int podCount = kubernetes.listPods().size();
+        clientUtils.sendDurableMessages(resourcesManager, addSpace, address, credentials, messagesBatch);
+        restartBrokers(podCount);
+
+        // Seems that the service/route can sometimes not be immediately available despite the pod being Ready.
+        assertConnectable(addSpace, credentials);
+        clientUtils.receiveDurableMessages(resourcesManager, addSpace, address, credentials, messagesBatch);
+
+    }
+
+    private void doTestTopicPersistentMessages(AddressSpace addSpace, Address topic, Address subscription) throws Exception {
+        resourcesManager.setAddresses(topic, subscription);
+
+        int podCount = kubernetes.listPods().size();
+
+        AmqpClient client = resourcesManager.getAmqpClientFactory().createTopicClient(addSpace);
+        client.getConnectOptions().setCredentials(credentials);
+
+        log.info("Subscribe first receiver");
+        Future<List<Message>> recvResults = client.recvMessages(AddressUtils.getQualifiedSubscriptionAddress(subscription), 30);
+        clientUtils.sendDurableMessages(resourcesManager, addSpace, topic, credentials, 30);
+        assertThat("Wrong messages received: ", recvResults.get(1, TimeUnit.MINUTES).size(), is(30));
+        clientUtils.sendDurableMessages(resourcesManager, addSpace, topic, credentials, 30);
+
+        restartBrokers(podCount);
+
+        // Seems that the service/route can sometimes not be immediately available despite the pod being Ready.
+        assertConnectable(addSpace, credentials);
+
+        log.info("Subscribe receiver again");
+        recvResults = client.recvMessages(AddressUtils.getQualifiedSubscriptionAddress(subscription), 30);
+        assertThat("Wrong messages received: batch2", recvResults.get(1, TimeUnit.MINUTES).size(), is(30));
+    }
+
+    private void restartBrokers(int podCount) throws Exception {
+        Map<String, String> labelSelector = Collections.singletonMap("role", "broker");
+        List<String> uids = kubernetes.listPods(kubernetes.getInfraNamespace(), labelSelector).stream().map(pod -> pod.getMetadata().getUid()).collect(Collectors.toList());
+
+        kubernetes.deletePod(kubernetes.getInfraNamespace(), labelSelector);
+
+        waitForPodsToTerminate(uids);
+        TestUtils.waitForExpectedReadyPods(kubernetes, kubernetes.getInfraNamespace(), podCount, new TimeoutBudget(10, TimeUnit.MINUTES));
+        log.info("Broker pods restarted");
+    }
+
+    private void assertConnectable(AddressSpace space, UserCredentials user) throws Exception {
+        TimeoutBudget budget = new TimeoutBudget(1, TimeUnit.MINUTES);
+        String name = space.getMetadata().getName();
+        do {
+            try {
+                clientUtils.connectAddressSpace(resourcesManager, space, user);
+                log.info("Successfully connected to address space : {}", name);
+                return;
+            } catch (IOException e) {
+                log.info("Failed to connect to address space: {} - {}", name, e.getMessage());
+            }
+            Thread.sleep(1000);
+        } while (!budget.timeoutExpired());
+
+        fail(String.format("Failed to assert address space %s connectable within timeout", name));
+    }
+
+}
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 0d16100b0..68e6c6d5f 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
@@ -73,7 +73,7 @@ public class ApiServerTest extends TestBase implements ITestIsolatedStandard {
     void testRestApiGetSchema() throws Exception {
         AddressPlan queuePlan = PlanUtils.createAddressPlanObject("test-schema-rest-api-addr-plan", AddressType.QUEUE,
                 Arrays.asList(new ResourceRequest("broker", 0.6), new ResourceRequest("router", 0.0)));
-        resourcesManager.createAddressPlan(queuePlan, false);
+        resourcesManager.createAddressPlan(queuePlan);
 
         //define and create address space plan
         List<ResourceAllowance> resources = Arrays.asList(
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/isolated/api/CustomResourceDefinitionAddressesTest.java b/systemtests/src/test/java/io/enmasse/systemtest/isolated/api/CustomResourceDefinitionAddressesTest.java
index 386de9f3e..a5fe9ca77 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/isolated/api/CustomResourceDefinitionAddressesTest.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/isolated/api/CustomResourceDefinitionAddressesTest.java
@@ -45,7 +45,7 @@ public class CustomResourceDefinitionAddressesTest extends TestBase implements I
     void setUpSelenium() throws Exception {
         brokered = new AddressSpaceBuilder()
                 .withNewMetadata()
-                .withName("ccreateAddressSpacePlanrd-address-space")
+                .withName("crd-address-space")
                 .withNamespace(kubernetes.getInfraNamespace())
                 .endMetadata()
                 .withNewSpec()
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
new file mode 100644
index 000000000..9c47203f3
--- /dev/null
+++ b/systemtests/src/test/java/io/enmasse/systemtest/isolated/bridging/ConnectorsTest.java
@@ -0,0 +1,267 @@
+/*
+ * 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.isolated.bridging;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.slf4j.Logger;
+
+import io.enmasse.address.model.Address;
+import io.enmasse.address.model.AddressBuilder;
+import io.enmasse.address.model.AddressSpace;
+import io.enmasse.address.model.AddressSpaceBuilder;
+import io.enmasse.address.model.AddressSpaceSpecConnectorAddressRuleBuilder;
+import io.enmasse.address.model.AddressSpaceSpecConnectorBuilder;
+import io.enmasse.address.model.AddressSpaceSpecConnectorCredentialsBuilder;
+import io.enmasse.address.model.AddressSpaceSpecConnectorEndpointBuilder;
+import io.enmasse.systemtest.Endpoint;
+import io.enmasse.systemtest.UserCredentials;
+import io.enmasse.systemtest.amqp.AmqpClient;
+import io.enmasse.systemtest.amqp.AmqpConnectOptions;
+import io.enmasse.systemtest.amqp.QueueTerminusFactory;
+import io.enmasse.systemtest.bases.TestBase;
+import io.enmasse.systemtest.bases.isolated.ITestIsolatedStandard;
+import io.enmasse.systemtest.logs.CustomLogger;
+import io.enmasse.systemtest.model.addressspace.AddressSpacePlans;
+import io.enmasse.systemtest.model.addressspace.AddressSpaceType;
+import io.enmasse.systemtest.platform.apps.SystemtestsKubernetesApps;
+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.vertx.proton.ProtonClientOptions;
+import io.vertx.proton.ProtonQoS;
+
+class ConnectorsTest extends TestBase implements ITestIsolatedStandard{
+
+    //tested usecases
+    //Sending messages to a remote AMQP endpoint via a local address space - by creating a connector and using prefixing
+    //Receiving messages from a remote AMQP endpoint via a local address space - by creating a connector and using prefixing
+    //If I config a connector to refer to a host that does not exist, I'd expect the addressspace overall to report ready true, whereas the connector's status should report the failure.
+
+    private static final String REMOTE_NAME = "remote1";
+    private static final String BASIC_QUEUE1 = "basic1";
+    private static final String BASIC_QUEUE2 = "basic2";
+    private static final String SLASHED_QUEUE1 = "dummy/foo";
+    private static final String SLASHED_QUEUE2 = "dummy/baz";
+    private static final String BASIC_QUEUES_PATTERN = "*";
+    private static final String SLASHED_QUEUES_PATTERN = "dummy/*";
+
+    private static Logger log = CustomLogger.getLogger();
+
+    private final String remoteBrokerNamespace = "systemtests-external-broker";
+    private final String remoteBrokerUsername = "test-user";
+    private final String remoteBrokerPassword = "test-password";
+    private Endpoint remoteBrokerEndpoint;
+
+    @BeforeEach
+    void deployBroker() throws Exception {
+        SystemtestsKubernetesApps.deployAMQBroker(remoteBrokerNamespace, remoteBrokerUsername, remoteBrokerPassword, SLASHED_QUEUE1, SLASHED_QUEUE2, BASIC_QUEUE1, BASIC_QUEUE2);
+        remoteBrokerEndpoint = SystemtestsKubernetesApps.getAMQBrokerEndpoint(remoteBrokerNamespace);
+        log.info("Broker endpoint: {}", remoteBrokerEndpoint);
+    }
+
+    @AfterEach
+    void undeployBroker(ExtensionContext context) throws Exception {
+        if (context.getExecutionException().isPresent()) { //test failed
+            logCollector.collectLogsOfPodsInNamespace(remoteBrokerNamespace);
+            logCollector.collectEvents(remoteBrokerNamespace);
+        }
+        SystemtestsKubernetesApps.deleteAMQBroker(remoteBrokerNamespace);
+    }
+
+    @Test
+    void testBrokerDeployment() throws Exception {
+        AmqpClient client = createClientToRemoteBroker();
+
+        Address testQueue = new AddressBuilder()
+                .withNewMetadata()
+                .withName("testtt")
+                .endMetadata()
+                .withNewSpec()
+                .withAddress(SLASHED_QUEUE1)
+                .endSpec()
+                .build();
+        QueueTest.runQueueTest(client, testQueue);
+        client.close();
+    }
+
+    @Test
+    void testSendThroughConnector1() throws Exception {
+        doTestSendThroughConnector(BASIC_QUEUES_PATTERN, new String[] {BASIC_QUEUE1, BASIC_QUEUE2});
+    }
+
+    @Test
+    void testSendThroughConnector2() throws Exception {
+        doTestSendThroughConnector(SLASHED_QUEUES_PATTERN, new String [] {SLASHED_QUEUE1, SLASHED_QUEUE2});
+    }
+
+    @Test
+    void testReceiveThroughConnector1() throws Exception {
+        doTestReceiveThroughConnector(BASIC_QUEUES_PATTERN, new String[] {BASIC_QUEUE1, BASIC_QUEUE2});
+    }
+
+    @Test
+    void testReceiveThroughConnector2() throws Exception {
+        doTestReceiveThroughConnector(SLASHED_QUEUES_PATTERN, new String [] {SLASHED_QUEUE1, SLASHED_QUEUE2});
+    }
+
+    @Test
+    void testBadConfiguration() throws Exception {
+        AddressSpace space = new AddressSpaceBuilder()
+                .withNewMetadata()
+                .withNamespace(kubernetes.getInfraNamespace())
+                .withName("send-to-connector")
+                .endMetadata()
+                .withNewSpec()
+                .withType(AddressSpaceType.STANDARD.toString())
+                .withPlan(AddressSpacePlans.STANDARD_SMALL)
+                .withConnectors(new AddressSpaceSpecConnectorBuilder()
+                        .withName(REMOTE_NAME)
+                        .addToEndpointHosts(new AddressSpaceSpecConnectorEndpointBuilder()
+                                .withHost("nonexistinghost.jeje.hola")
+                                .withPort(8080)
+                                .build())
+                        .withCredentials(new AddressSpaceSpecConnectorCredentialsBuilder()
+                                .withNewUsername()
+                                    .withValue("dummy")
+                                    .endUsername()
+                                .withNewPassword()
+                                    .withValue("dummy")
+                                    .endPassword()
+                                .build())
+                        .addToAddresses(new AddressSpaceSpecConnectorAddressRuleBuilder()
+                                .withName("queuesrule")
+                                .withPattern(BASIC_QUEUES_PATTERN)
+                                .build())
+                        .build())
+                .endSpec()
+                .build();
+        resourcesManager.createAddressSpace(space);
+        Assertions.assertThrows(IllegalStateException.class, () -> {
+            AddressSpaceUtils.waitForAddressSpaceConnectorsReady(space, new TimeoutBudget(1, TimeUnit.MINUTES));
+        });
+    }
+
+    private void doTestSendThroughConnector(String addressRule, String[] remoteQueues) throws Exception, InterruptedException, ExecutionException, TimeoutException {
+        AddressSpace space = createAddressSpace("send-to-connector", addressRule);
+
+        UserCredentials localUser = new UserCredentials("test", "test");
+        resourcesManager.createOrUpdateUser(space, localUser);
+
+        int messagesBatch = 50;
+
+        //send through connector
+
+        AmqpClient localClient = getAmqpClientFactory().createQueueClient(space);
+        localClient.getConnectOptions().setCredentials(localUser);
+
+        for(String remoteQueue : remoteQueues) {
+            String connectorQueue = REMOTE_NAME + "/" + remoteQueue;
+            localClient.sendMessages(connectorQueue, TestUtils.generateMessages(messagesBatch));
+        }
+
+        //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;
+
+        //send to remote broker
+
+        AmqpClient clientToRemote = createClientToRemoteBroker();
+
+        for(String remoteQueue : remoteQueues) {
+            clientToRemote.sendMessages(remoteQueue, TestUtils.generateMessages(messagesBatch));
+        }
+
+        //receive through connector
+
+        AmqpClient localClient = getAmqpClientFactory().createQueueClient(space);
+        localClient.getConnectOptions().setCredentials(localUser);
+
+        for(String remoteQueue : remoteQueues) {
+            String connectorQueue = REMOTE_NAME + "/" + remoteQueue;
+            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 AddressSpace createAddressSpace(String name, String addressRule) throws Exception {
+        AddressSpace space = new AddressSpaceBuilder()
+                .withNewMetadata()
+                .withNamespace(kubernetes.getInfraNamespace())
+                .withName(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(addressRule)
+                                .build())
+                        .build())
+                .endSpec()
+                .build();
+        resourcesManager.createAddressSpace(space);
+        AddressSpaceUtils.waitForAddressSpaceConnectorsReady(space);
+        return space;
+    }
+
+    private AmqpClient createClientToRemoteBroker() {
+
+        ProtonClientOptions clientOptions = new ProtonClientOptions();
+        clientOptions.setSsl(false);
+
+        AmqpConnectOptions connectOptions = new AmqpConnectOptions()
+                .setTerminusFactory(new QueueTerminusFactory())
+                .setEndpoint(remoteBrokerEndpoint)
+                .setProtonClientOptions(clientOptions)
+                .setQos(ProtonQoS.AT_LEAST_ONCE)
+                .setUsername(remoteBrokerUsername)
+                .setPassword(remoteBrokerPassword);
+
+        return getAmqpClientFactory().createClient(connectOptions);
+    }
+
+}
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/isolated/infra/brokered/InfraTestBrokered.java b/systemtests/src/test/java/io/enmasse/systemtest/isolated/infra/brokered/InfraTestBrokered.java
index a4bad113d..edffc38e1 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/isolated/infra/brokered/InfraTestBrokered.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/isolated/infra/brokered/InfraTestBrokered.java
@@ -67,7 +67,7 @@ class InfraTestBrokered extends InfraTestBase implements ITestIsolatedBrokered {
         exampleAddressPlan = PlanUtils.createAddressPlanObject("example-queue-plan-brokered", AddressType.QUEUE,
                 Collections.singletonList(new ResourceRequest("broker", 1.0)));
 
-        resourcesManager.createAddressPlan(exampleAddressPlan, false);
+        resourcesManager.createAddressPlan(exampleAddressPlan);
 
         AddressSpacePlan exampleSpacePlan = new AddressSpacePlanBuilder()
                 .withNewMetadata()
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/isolated/infra/standard/InfraTestStandard.java b/systemtests/src/test/java/io/enmasse/systemtest/isolated/infra/standard/InfraTestStandard.java
index 96b3e8886..478d5d1a8 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/isolated/infra/standard/InfraTestStandard.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/isolated/infra/standard/InfraTestStandard.java
@@ -85,7 +85,7 @@ class InfraTestStandard extends InfraTestBase implements ITestIsolatedStandard {
         exampleAddressPlan = PlanUtils.createAddressPlanObject("example-queue-plan-standard", AddressType.QUEUE,
                 Arrays.asList(new ResourceRequest("broker", 1.0), new ResourceRequest("router", 1.0)));
 
-        resourcesManager.createAddressPlan(exampleAddressPlan, false);
+        resourcesManager.createAddressPlan(exampleAddressPlan);
 
         AddressSpacePlan exampleSpacePlan = new AddressSpacePlanBuilder()
                 .withNewMetadata()
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/isolated/standard/AnycastTest.java b/systemtests/src/test/java/io/enmasse/systemtest/isolated/standard/AnycastTest.java
index bfc80ceda..4c4dccb91 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/isolated/standard/AnycastTest.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/isolated/standard/AnycastTest.java
@@ -9,6 +9,8 @@ import io.enmasse.address.model.Address;
 import io.enmasse.address.model.AddressBuilder;
 import io.enmasse.address.model.AddressSpace;
 import io.enmasse.address.model.AddressSpaceBuilder;
+import io.enmasse.systemtest.Environment;
+import io.enmasse.systemtest.UserCredentials;
 import io.enmasse.systemtest.amqp.AmqpClient;
 import io.enmasse.systemtest.bases.TestBase;
 import io.enmasse.systemtest.bases.isolated.ITestIsolatedStandard;
@@ -34,6 +36,8 @@ import static org.hamcrest.MatcherAssert.assertThat;
 public class AnycastTest extends TestBase implements ITestIsolatedStandard {
 
     private AddressSpace addressSpace;
+    private UserCredentials userCredentials = new UserCredentials("test", "test");
+
 
     public static void runAnycastTest(Address dest, AmqpClient... clients) throws InterruptedException, TimeoutException, IOException, ExecutionException {
         if (clients.length == 0) {
@@ -75,6 +79,8 @@ public class AnycastTest extends TestBase implements ITestIsolatedStandard {
 
     @Test
     void testMultipleReceivers() throws Exception {
+
+
         Address dest = new AddressBuilder()
                 .withNewMetadata()
                 .withNamespace(addressSpace.getMetadata().getNamespace())
@@ -87,6 +93,10 @@ public class AnycastTest extends TestBase implements ITestIsolatedStandard {
                 .endSpec()
                 .build();
         resourcesManager.setAddresses(dest);
+
+        resourcesManager.createOrUpdateUser(addressSpace, userCredentials);
+        commonResourcesManager.initFactories(addressSpace, userCredentials);
+
         AmqpClient client1 = getAmqpClientFactory().createQueueClient();
         AmqpClient client2 = getAmqpClientFactory().createQueueClient();
         AmqpClient client3 = getAmqpClientFactory().createQueueClient();
@@ -144,6 +154,9 @@ public class AnycastTest extends TestBase implements ITestIsolatedStandard {
 
         waitForRouterReplicas(addressSpace, 3);
 
+        resourcesManager.createOrUpdateUser(addressSpace, userCredentials);
+        commonResourcesManager.initFactories(addressSpace, userCredentials);
+
         //simple send/receive
         AmqpClient client1 = getAmqpClientFactory().createQueueClient();
         AmqpClient client2 = getAmqpClientFactory().createQueueClient();
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/shared/brokered/BrokeredPersistentMessagesTest.java b/systemtests/src/test/java/io/enmasse/systemtest/shared/brokered/BrokeredPersistentMessagesTest.java
deleted file mode 100644
index fb68d8875..000000000
--- a/systemtests/src/test/java/io/enmasse/systemtest/shared/brokered/BrokeredPersistentMessagesTest.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * 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;
-
-import io.enmasse.address.model.Address;
-import io.enmasse.address.model.AddressBuilder;
-import io.enmasse.systemtest.bases.common.PersistentMessagesTestBase;
-import io.enmasse.systemtest.bases.shared.ITestSharedBrokered;
-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.utils.AddressUtils;
-import org.junit.jupiter.api.Test;
-
-public class BrokeredPersistentMessagesTest extends PersistentMessagesTestBase implements ITestSharedBrokered {
-
-    @Override
-    public String getDefaultAddrSpaceIdentifier() {
-        return "brokered-persistent";
-    }
-
-    @Override
-    public AddressSpaceType getAddressSpaceType() {
-        return AddressSpaceType.BROKERED;
-    }
-
-    @Override
-    public String getDefaultAddressSpacePlan() {
-        return AddressSpacePlans.BROKERED;
-    }
-
-    @Test
-    void testBrokeredQueue() throws Exception {
-        Address brokeredQueue = new AddressBuilder()
-                .withNewMetadata()
-                .withNamespace(getSharedAddressSpace().getMetadata().getNamespace())
-                .withName(AddressUtils.generateAddressMetadataName(getSharedAddressSpace(), "test-queue-brokered"))
-                .endMetadata()
-                .withNewSpec()
-                .withType("queue")
-                .withAddress("test-queue-brokered")
-                .withPlan(DestinationPlan.BROKERED_QUEUE)
-                .endSpec()
-                .build();
-        doTestQueuePersistentMessages(brokeredQueue, 100);
-    }
-
-}
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/isolated/api/UserApiTest.java b/systemtests/src/test/java/io/enmasse/systemtest/shared/brokered/api/UserApiTest.java
similarity index 53%
rename from systemtests/src/test/java/io/enmasse/systemtest/isolated/api/UserApiTest.java
rename to systemtests/src/test/java/io/enmasse/systemtest/shared/brokered/api/UserApiTest.java
index 7eda8ea1c..08a297aef 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/isolated/api/UserApiTest.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/shared/brokered/api/UserApiTest.java
@@ -1,26 +1,16 @@
 /*
- * Copyright 2018, EnMasse authors.
+ * 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.isolated.api;
+package io.enmasse.systemtest.shared.brokered.api;
 
-import io.enmasse.address.model.Address;
-import io.enmasse.address.model.AddressBuilder;
 import io.enmasse.address.model.AddressSpace;
-import io.enmasse.address.model.AddressSpaceBuilder;
 import io.enmasse.systemtest.UserCredentials;
-import io.enmasse.systemtest.amqp.AmqpClient;
-import io.enmasse.systemtest.amqp.UnauthorizedAccessException;
 import io.enmasse.systemtest.bases.TestBase;
-import io.enmasse.systemtest.bases.isolated.ITestIsolatedStandard;
+import io.enmasse.systemtest.bases.shared.ITestSharedBrokered;
 import io.enmasse.systemtest.cmdclients.KubeCMDClient;
 import io.enmasse.systemtest.executor.ExecutionResultData;
 import io.enmasse.systemtest.logs.CustomLogger;
-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.utils.AddressSpaceUtils;
-import io.enmasse.systemtest.utils.AddressUtils;
 import io.enmasse.systemtest.utils.UserUtils;
 import io.enmasse.user.model.v1.DoneableUser;
 import io.enmasse.user.model.v1.Operation;
@@ -28,65 +18,23 @@ import io.enmasse.user.model.v1.User;
 import io.enmasse.user.model.v1.UserAuthorizationBuilder;
 import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.vertx.core.json.JsonObject;
-import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
-class UserApiTest extends TestBase implements ITestIsolatedStandard {
-    private static Logger log = CustomLogger.getLogger();
-    private AddressSpace brokered = new AddressSpaceBuilder()
-            .withNewMetadata()
-            .withName("user-api-brokered")
-            .withNamespace(kubernetes.getInfraNamespace())
-            .endMetadata()
-            .withNewSpec()
-            .withType(AddressSpaceType.BROKERED.toString())
-            .withPlan(AddressSpacePlans.BROKERED)
-            .withNewAuthenticationService()
-            .withName("standard-authservice")
-            .endAuthenticationService()
-            .endSpec()
-            .build();
-    private AddressSpace standard = new AddressSpaceBuilder()
-            .withNewMetadata()
-            .withName("user-api-standard")
-            .withNamespace(kubernetes.getInfraNamespace())
-            .endMetadata()
-            .withNewSpec()
-            .withType(AddressSpaceType.STANDARD.toString())
-            .withPlan(AddressSpacePlans.STANDARD_SMALL)
-            .withNewAuthenticationService()
-            .withName("standard-authservice")
-            .endAuthenticationService()
-            .endSpec()
-            .build();
-    private Map<AddressSpace, User> users = new HashMap<>();
+public class UserApiTest extends TestBase implements ITestSharedBrokered {
 
-    @BeforeEach
-    void deployAddressSpaces() throws Exception {
-        if (!AddressSpaceUtils.existAddressSpace(brokered.getMetadata().getNamespace(), brokered.getMetadata().getName())) {
-            resourcesManager.createAddressSpace(brokered);
-        }
-        if (!AddressSpaceUtils.existAddressSpace(standard.getMetadata().getNamespace(), standard.getMetadata().getName())) {
-            resourcesManager.createAddressSpace(standard);
-        }
-        users.clear();
-    }
+    private Map<AddressSpace, User> users = new HashMap<>();
+    private Logger LOGGER = CustomLogger.getLogger();
 
     @AfterEach
     void cleanUsers() {
@@ -94,23 +42,11 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
             try {
                 resourcesManager.removeUser(addressSpace, user);
             } catch (Exception e) {
-                log.info("Clean: User not exists {}", user.getSpec().getUsername());
+                LOGGER.info("Clean: User not exists {}", user.getSpec().getUsername());
             }
         });
     }
 
-    @BeforeAll
-    void beforeAll() {
-        commonResourcesManager.setReuseAddressSpace();
-    }
-
-    @AfterAll
-    void removeAddressSpaces() throws Exception {
-        commonResourcesManager.unsetReuseAddressSpace();
-        commonResourcesManager.deleteAddressspacesFromList();
-    }
-
-
     @Test
     void testCreateDeleteUserUsingCRD() throws Exception {
         UserCredentials cred = new UserCredentials("pepanatestovani", "pepaNaTestovani");
@@ -123,19 +59,19 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
                 .endSpec()
                 .done();
 
-        JsonObject userDefinitionPayload = UserUtils.userToJson(brokered.getMetadata().getName(), testUser);
-        users.put(brokered, testUser);
+        JsonObject userDefinitionPayload = UserUtils.userToJson(getSharedAddressSpace().getMetadata().getName(), testUser);
+        users.put(getSharedAddressSpace(), testUser);
 
         //create user
         assertThat(KubeCMDClient.createCR(kubernetes.getInfraNamespace(), userDefinitionPayload.toString()).getRetCode(), is(true));
-        assertThat(KubeCMDClient.getUser(kubernetes.getInfraNamespace(), brokered.getMetadata().getName(), cred.getUsername()).getRetCode(), is(true));
+        assertThat(KubeCMDClient.getUser(kubernetes.getInfraNamespace(), getSharedAddressSpace().getMetadata().getName(), cred.getUsername()).getRetCode(), is(true));
 
         //patch user
-        assertThat(KubeCMDClient.patchCR(User.KIND.toLowerCase(), brokered.getMetadata().getName() + "." + cred.getUsername(), "{\"spec\":{\"authentication\":{\"password\":\"aGVp\"}}}").getRetCode(), is(true));
+        assertThat(KubeCMDClient.patchCR(User.KIND.toLowerCase(), getSharedAddressSpace().getMetadata().getName() + "." + cred.getUsername(), "{\"spec\":{\"authentication\":{\"password\":\"aGVp\"}}}").getRetCode(), is(true));
 
         //delete user
-        assertThat(KubeCMDClient.deleteUser(kubernetes.getInfraNamespace(), brokered.getMetadata().getName(), cred.getUsername()).getRetCode(), is(true));
-        assertThat(KubeCMDClient.getUser(kubernetes.getInfraNamespace(), brokered.getMetadata().getName(), cred.getUsername()).getRetCode(), is(false));
+        assertThat(KubeCMDClient.deleteUser(kubernetes.getInfraNamespace(), getSharedAddressSpace().getMetadata().getName(), cred.getUsername()).getRetCode(), is(true));
+        assertThat(KubeCMDClient.getUser(kubernetes.getInfraNamespace(), getSharedAddressSpace().getMetadata().getName(), cred.getUsername()).getRetCode(), is(false));
     }
 
     @Test
@@ -150,15 +86,15 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
                 .endSpec()
                 .done();
 
-        JsonObject userDefinitionPayload = UserUtils.userToJson(brokered.getMetadata().getName(), testUser);
+        JsonObject userDefinitionPayload = UserUtils.userToJson(getSharedAddressSpace().getMetadata().getName(), testUser);
         userDefinitionPayload.getJsonObject("spec").getJsonArray("authorization").getJsonObject(0).getJsonArray("operations").add("unknown");
-        users.put(brokered, testUser);
+        users.put(getSharedAddressSpace(), testUser);
 
         //create user
         ExecutionResultData createUserResponse = KubeCMDClient.createCR(kubernetes.getInfraNamespace(), userDefinitionPayload.toString());
         assertThat(createUserResponse.getRetCode(), is(false));
         assertTrue(createUserResponse.getStdErr().contains("value not one of declared Enum instance names: [send, view, recv, manage]"));
-        assertThat(KubeCMDClient.getUser(kubernetes.getInfraNamespace(), brokered.getMetadata().getName(), cred.getUsername()).getRetCode(), is(false));
+        assertThat(KubeCMDClient.getUser(kubernetes.getInfraNamespace(), getSharedAddressSpace().getMetadata().getName(), cred.getUsername()).getRetCode(), is(false));
 
         User testUser2 = UserUtils.createUserResource(cred)
                 .editSpec()
@@ -175,7 +111,7 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
         ExecutionResultData createUserResponse2 = KubeCMDClient.createCR(kubernetes.getInfraNamespace(), userDefinitionPayload2.toString());
         assertThat(createUserResponse2.getRetCode(), is(false));
         assertTrue(createUserResponse2.getStdErr().contains(String.format("The name of the object (.%s) is not valid", cred.getUsername())));
-        assertThat(KubeCMDClient.getUser(kubernetes.getInfraNamespace(), brokered.getMetadata().getName(), cred.getUsername()).getRetCode(), is(false));
+        assertThat(KubeCMDClient.getUser(kubernetes.getInfraNamespace(), getSharedAddressSpace().getMetadata().getName(), cred.getUsername()).getRetCode(), is(false));
     }
 
     @Test
@@ -190,12 +126,12 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
                 .endSpec()
                 .done();
 
-        JsonObject userDefinitionPayload = UserUtils.userToJson(brokered.getMetadata().getName(), testUser);
-        users.put(brokered, testUser);
+        JsonObject userDefinitionPayload = UserUtils.userToJson(getSharedAddressSpace().getMetadata().getName(), testUser);
+        users.put(getSharedAddressSpace(), testUser);
 
         //create user
         assertThat(KubeCMDClient.createCR(kubernetes.getInfraNamespace(), userDefinitionPayload.toString()).getRetCode(), is(true));
-        assertThat(KubeCMDClient.getUser(kubernetes.getInfraNamespace(), brokered.getMetadata().getName(), cred.getUsername()).getRetCode(), is(true));
+        assertThat(KubeCMDClient.getUser(kubernetes.getInfraNamespace(), getSharedAddressSpace().getMetadata().getName(), cred.getUsername()).getRetCode(), is(true));
 
         testUser = new DoneableUser(testUser)
                 .editSpec()
@@ -205,80 +141,32 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
                 .endAuthorization()
                 .endSpec().done();
 
-        userDefinitionPayload = UserUtils.userToJson(brokered.getMetadata().getName(), testUser);
+        userDefinitionPayload = UserUtils.userToJson(getSharedAddressSpace().getMetadata().getName(), testUser);
 
         //update user
         assertThat(KubeCMDClient.updateCR(kubernetes.getInfraNamespace(), userDefinitionPayload.toString()).getRetCode(), is(true));
-        assertThat(KubeCMDClient.getUser(kubernetes.getInfraNamespace(), brokered.getMetadata().getName(), cred.getUsername()).getRetCode(), is(true));
-        assertTrue(resourcesManager.getUser(brokered, testUser.getSpec().getUsername()).toString().contains(Operation.recv.toString()));
+        assertThat(KubeCMDClient.getUser(kubernetes.getInfraNamespace(), getSharedAddressSpace().getMetadata().getName(), cred.getUsername()).getRetCode(), is(true));
+        assertTrue(resourcesManager.getUser(getSharedAddressSpace(), testUser.getSpec().getUsername()).toString().contains(Operation.recv.toString()));
 
 
         //delete user
-        assertThat(KubeCMDClient.deleteUser(kubernetes.getInfraNamespace(), brokered.getMetadata().getName(), cred.getUsername()).getRetCode(), is(true));
-        assertThat(KubeCMDClient.getUser(kubernetes.getInfraNamespace(), brokered.getMetadata().getName(), cred.getUsername()).getRetCode(), is(false));
-    }
-
-
-    @Test
-    void testUpdateUserPermissionsUserAPI() throws Exception {
-        Address queue = new AddressBuilder()
-                .withNewMetadata()
-                .withNamespace(standard.getMetadata().getNamespace())
-                .withName(AddressUtils.generateAddressMetadataName(standard, "test-queue"))
-                .endMetadata()
-                .withNewSpec()
-                .withType("queue")
-                .withAddress("test-queue")
-                .withPlan(DestinationPlan.STANDARD_LARGE_QUEUE)
-                .endSpec()
-                .build();
-        resourcesManager.setAddresses(queue);
-
-        UserCredentials cred = new UserCredentials("pepa", "pepapw");
-        User testUser = UserUtils.createUserResource(cred)
-                .editSpec()
-                .withAuthorization(Collections.singletonList(
-                        new UserAuthorizationBuilder()
-                                .withAddresses(queue.getSpec().getAddress())
-                                .withOperations(Operation.send).build()))
-                .endSpec()
-                .done();
-
-        users.put(brokered, testUser);
-        testUser = resourcesManager.createOrUpdateUser(standard, testUser);
-
-        AmqpClient client = getAmqpClientFactory().createQueueClient(standard);
-        client.getConnectOptions().setCredentials(cred);
-        assertThat(client.sendMessages(queue.getSpec().getAddress(), Arrays.asList("kuk", "puk")).get(1, TimeUnit.MINUTES), is(2));
-
-        testUser = new DoneableUser(testUser)
-                .editSpec()
-                .withAuthorization(Collections.singletonList(new UserAuthorizationBuilder()
-                        .withAddresses(queue.getSpec().getAddress())
-                        .withOperations(Operation.recv).build()))
-                .endSpec()
-                .done();
-
-        resourcesManager.createOrUpdateUser(standard, testUser);
-        Throwable exception = assertThrows(ExecutionException.class,
-                () -> client.sendMessages(queue.getSpec().getAddress(), Arrays.asList("kuk", "puk")).get(10, TimeUnit.SECONDS));
-        assertTrue(exception.getCause() instanceof UnauthorizedAccessException);
-        assertThat(client.recvMessages(queue.getSpec().getAddress(), 2).get(1, TimeUnit.MINUTES).size(), is(2));
+        assertThat(KubeCMDClient.deleteUser(kubernetes.getInfraNamespace(), getSharedAddressSpace().getMetadata().getName(), cred.getUsername()).getRetCode(), is(true));
+        assertThat(KubeCMDClient.getUser(kubernetes.getInfraNamespace(), getSharedAddressSpace().getMetadata().getName(), cred.getUsername()).getRetCode(), is(false));
     }
 
     @Test
     void testUserWithSimilarNamesAndAlreadyExistingUser() throws Exception {
         UserCredentials cred = new UserCredentials("user2", "user2");
-        User testUser = resourcesManager.createOrUpdateUser(brokered, cred);
+        User testUser = resourcesManager.createOrUpdateUser(getSharedAddressSpace(), cred);
 
         UserCredentials cred2 = new UserCredentials("user23", "test_user23");
-        User testUser2 = resourcesManager.createOrUpdateUser(brokered, cred2);
+        User testUser2 = resourcesManager.createOrUpdateUser(getSharedAddressSpace(), cred2);
 
-        users.put(brokered, testUser);
-        users.put(brokered, testUser2);
+        users.put(getSharedAddressSpace(), testUser);
+        users.put(getSharedAddressSpace(), testUser2);
 
         assertThrows(KubernetesClientException.class, () ->
-                kubernetes.getUserClient(brokered.getMetadata().getNamespace()).create(resourcesManager.getUser(brokered, testUser.getSpec().getUsername())));
+                kubernetes.getUserClient(getSharedAddressSpace().getMetadata().getNamespace()).create(resourcesManager.getUser(getSharedAddressSpace(), testUser.getSpec().getUsername())));
     }
 
     @Test
@@ -286,7 +174,7 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
         UserCredentials cred = new UserCredentials("UserPepinator", "ff^%fh16");
         User testUser = UserUtils.createUserResource(cred)
                 .editMetadata()
-                .withName(brokered.getMetadata().getName() + "." + "userpepinator")
+                .withName(getSharedAddressSpace().getMetadata().getName() + "." + "userpepinator")
                 .endMetadata()
                 .editSpec()
                 .withAuthorization(Collections.singletonList(
@@ -296,7 +184,7 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
                 .endSpec()
                 .done();
 
-        assertThrows(KubernetesClientException.class, () -> resourcesManager.createOrUpdateUser(brokered, testUser));
+        assertThrows(KubernetesClientException.class, () -> resourcesManager.createOrUpdateUser(getSharedAddressSpace(), testUser));
     }
 
     @Test
@@ -311,14 +199,14 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
                                 .withOperations(Operation.send).build()))
                 .endSpec()
                 .done();
-        users.put(brokered, testUser);
-        resourcesManager.createOrUpdateUser(brokered, testUser);
+        users.put(getSharedAddressSpace(), testUser);
+        resourcesManager.createOrUpdateUser(getSharedAddressSpace(), testUser);
 
         //first char of username must be a-z0-9 (other symbols respond with HTTP:400)
         UserCredentials cred2 = new UserCredentials("-hyphensymbolfirst", "password");
         User testUser2 = UserUtils.createUserResource(cred2)
                 .editMetadata()
-                .withName(brokered.getMetadata().getName() + "." + "userpepinator")
+                .withName(getSharedAddressSpace().getMetadata().getName() + "." + "userpepinator")
                 .endMetadata()
                 .editSpec()
                 .withAuthorization(Collections.singletonList(
@@ -327,7 +215,7 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
                                 .withOperations(Operation.send).build()))
                 .endSpec()
                 .done();
-        assertThrows(KubernetesClientException.class, () -> resourcesManager.createOrUpdateUser(brokered, testUser2));
+        assertThrows(KubernetesClientException.class, () -> resourcesManager.createOrUpdateUser(getSharedAddressSpace(), testUser2));
 
         //hyphen is allowed elsewhere in user name (response HTTP:201)
         UserCredentials cred3 = new UserCredentials("user-pepinator", "password");
@@ -339,14 +227,14 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
                                 .withOperations(Operation.send).build()))
                 .endSpec()
                 .done();
-        users.put(brokered, testUser3);
-        resourcesManager.createOrUpdateUser(brokered, testUser3);
+        users.put(getSharedAddressSpace(), testUser3);
+        resourcesManager.createOrUpdateUser(getSharedAddressSpace(), testUser3);
 
         //underscore is also allowed in user name (response HTTP:201)
         UserCredentials cred4 = new UserCredentials("user_pepinator", "password");
         User testUser4 = UserUtils.createUserResource(cred4)
                 .editMetadata()
-                .withName(brokered.getMetadata().getName() + "." + "testuser")
+                .withName(getSharedAddressSpace().getMetadata().getName() + "." + "testuser")
                 .endMetadata()
                 .editSpec()
                 .withAuthorization(Collections.singletonList(
@@ -355,14 +243,14 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
                                 .withOperations(Operation.send).build()))
                 .endSpec()
                 .done();
-        users.put(brokered, testUser4);
-        resourcesManager.createOrUpdateUser(brokered, testUser4);
+        users.put(getSharedAddressSpace(), testUser4);
+        resourcesManager.createOrUpdateUser(getSharedAddressSpace(), testUser4);
 
         //last char must also be a-z (other symbols respond with HTTP:400)
         UserCredentials cred5 = new UserCredentials("hyphensymbollast-", "password");
         User testUser5 = UserUtils.createUserResource(cred5)
                 .editMetadata()
-                .withName(brokered.getMetadata().getName() + "." + "userpepinator")
+                .withName(getSharedAddressSpace().getMetadata().getName() + "." + "userpepinator")
                 .endMetadata()
                 .editSpec()
                 .withAuthorization(Collections.singletonList(
@@ -372,7 +260,7 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
                 .endSpec()
                 .done();
         assertThrows(KubernetesClientException.class, () ->
-                resourcesManager.createOrUpdateUser(brokered, testUser5));
+                resourcesManager.createOrUpdateUser(getSharedAddressSpace(), testUser5));
 
         //username may start/end with 0-9 ()
         UserCredentials cred6 = new UserCredentials("01234usernamehere56789", "password");
@@ -384,8 +272,8 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
                                 .withOperations(Operation.send).build()))
                 .endSpec()
                 .done();
-        users.put(brokered, testUser6);
-        resourcesManager.createOrUpdateUser(brokered, testUser6);
+        users.put(getSharedAddressSpace(), testUser6);
+        resourcesManager.createOrUpdateUser(getSharedAddressSpace(), testUser6);
 
         //Foreign symbols may not be used in username (response HTTP:400)
         UserCredentials cred7 = new UserCredentials("invalid_Ö_username", "password");
@@ -399,41 +287,6 @@ class UserApiTest extends TestBase implements ITestIsolatedStandard {
                 .done();
         testUser7.getMetadata().setName("userpepinator");
         assertThrows(KubernetesClientException.class, () ->
-                resourcesManager.createOrUpdateUser(brokered, testUser5));
-    }
-
-    @Test
-    void testServiceaccountUser() throws Exception {
-        Address queue = new AddressBuilder()
-                .withNewMetadata()
-                .withNamespace(standard.getMetadata().getNamespace())
-                .withName(AddressUtils.generateAddressMetadataName(standard, "test-queue"))
-                .endMetadata()
-                .withNewSpec()
-                .withType("queue")
-                .withAddress("test-queue")
-                .withPlan(DestinationPlan.STANDARD_LARGE_QUEUE)
-                .endSpec()
-                .build();
-        resourcesManager.setAddresses(queue);
-        UserCredentials serviceAccount = new UserCredentials("test-service-account", "");
-        try {
-            resourcesManager.createUserServiceAccount(standard, serviceAccount);
-            UserCredentials messagingUser = new UserCredentials("@@serviceaccount@@",
-                    kubernetes.getServiceaccountToken(serviceAccount.getUsername(), environment.namespace()));
-            log.info("username: {}, password: {}", messagingUser.getUsername(), messagingUser.getPassword());
-
-            getClientUtils().assertCanConnect(standard, messagingUser, Collections.singletonList(queue), resourcesManager);
-
-            //delete user
-            assertThat("User deleting failed using oc cmd",
-                    KubeCMDClient.deleteUser(kubernetes.getInfraNamespace(), standard.getMetadata().getName(), serviceAccount.getUsername()).getRetCode(), is(true));
-            assertThat("User is still present",
-                    KubeCMDClient.getUser(kubernetes.getInfraNamespace(), standard.getMetadata().getName(), serviceAccount.getUsername()).getRetCode(), is(false));
-
-            getClientUtils().assertCannotConnect(standard, messagingUser, Collections.singletonList(queue), resourcesManager);
-        } finally {
-            kubernetes.deleteServiceAccount("test-service-account", environment.namespace());
-        }
+                resourcesManager.createOrUpdateUser(getSharedAddressSpace(), testUser5));
     }
 }
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/shared/standard/PlansTest.java b/systemtests/src/test/java/io/enmasse/systemtest/shared/standard/PlansTest.java
index e2aab6b21..18970f1d6 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/shared/standard/PlansTest.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/shared/standard/PlansTest.java
@@ -33,10 +33,10 @@ class PlansTest extends TestBase implements ITestSharedStandard {
         List<ResourceRequest> addressResources = Collections.singletonList(new ResourceRequest("broker", 0.1));
         String weakQueuePlanName = "pooled-standard-queue-weak";
         AddressPlan weakQueuePlan = PlanUtils.createAddressPlanObject(weakQueuePlanName, AddressType.QUEUE, addressResources);
-        resourcesManager.createAddressPlan(weakQueuePlan, false);
+        resourcesManager.createAddressPlan(weakQueuePlan);
 
         AddressSpacePlan standardPlan = resourcesManager.getAddressSpacePlan("standard");
-        resourcesManager.createAddressPlan(weakQueuePlan, false);
+        resourcesManager.createAddressPlan(weakQueuePlan);
         standardPlan = new DoneableAddressSpacePlan(standardPlan).addNewAddressPlan(weakQueuePlanName).done();
         resourcesManager.removeAddressSpacePlan(standardPlan);
 
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/shared/standard/StandardPersistentMessagesTest.java b/systemtests/src/test/java/io/enmasse/systemtest/shared/standard/StandardPersistentMessagesTest.java
deleted file mode 100644
index 954d7b114..000000000
--- a/systemtests/src/test/java/io/enmasse/systemtest/shared/standard/StandardPersistentMessagesTest.java
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
- * 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.standard;
-
-import io.enmasse.address.model.Address;
-import io.enmasse.address.model.AddressBuilder;
-import io.enmasse.systemtest.bases.common.PersistentMessagesTestBase;
-import io.enmasse.systemtest.bases.shared.ITestSharedStandard;
-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.utils.AddressUtils;
-import org.junit.jupiter.api.Test;
-
-public class StandardPersistentMessagesTest extends PersistentMessagesTestBase implements ITestSharedStandard {
-
-    @Override
-    public String getDefaultAddrSpaceIdentifier() {
-        return "standard-persistent";
-    }
-
-    @Override
-    public AddressSpaceType getAddressSpaceType() {
-        return AddressSpaceType.STANDARD;
-    }
-
-    @Override
-    public String getDefaultAddressSpacePlan() {
-        return AddressSpacePlans.STANDARD_UNLIMITED;
-    }
-
-    @Test
-    void testSmallQueue() throws Exception {
-        Address standardQueue = new AddressBuilder()
-                .withNewMetadata()
-                .withNamespace(getSharedAddressSpace().getMetadata().getNamespace())
-                .withName(AddressUtils.generateAddressMetadataName(getSharedAddressSpace(), "test-queue-standard"))
-                .endMetadata()
-                .withNewSpec()
-                .withType("queue")
-                .withAddress("test-queue-standard")
-                .withPlan(DestinationPlan.STANDARD_SMALL_QUEUE)
-                .endSpec()
-                .build();
-        doTestQueuePersistentMessages(standardQueue, 30);
-    }
-
-    @Test
-    void testLargeQueue() throws Exception {
-        Address standardLargeQueue = new AddressBuilder()
-                .withNewMetadata()
-                .withNamespace(getSharedAddressSpace().getMetadata().getNamespace())
-                .withName(AddressUtils.generateAddressMetadataName(getSharedAddressSpace(), "test-large-queue-standard"))
-                .endMetadata()
-                .withNewSpec()
-                .withType("queue")
-                .withAddress("test-large-queue-standard")
-                .withPlan(DestinationPlan.STANDARD_LARGE_QUEUE)
-                .endSpec()
-                .build();
-        doTestQueuePersistentMessages(standardLargeQueue, 30);
-    }
-
-    @Test
-    void testXLargeQueue() throws Exception {
-        Address standardXLargeQueue = new AddressBuilder()
-                .withNewMetadata()
-                .withNamespace(getSharedAddressSpace().getMetadata().getNamespace())
-                .withName(AddressUtils.generateAddressMetadataName(getSharedAddressSpace(), "test-xlarge-queue-standard"))
-                .endMetadata()
-                .withNewSpec()
-                .withType("queue")
-                .withAddress("test-xlarge-queue-standard")
-                .withPlan(DestinationPlan.STANDARD_XLARGE_QUEUE)
-                .endSpec()
-                .build();
-        doTestQueuePersistentMessages(standardXLargeQueue, 30);
-    }
-
-    @Test
-    void testTopic() throws Exception {
-        Address standardTopic = new AddressBuilder()
-                .withNewMetadata()
-                .withNamespace(getSharedAddressSpace().getMetadata().getNamespace())
-                .withName(AddressUtils.generateAddressMetadataName(getSharedAddressSpace(), "test-topic-standard"))
-                .endMetadata()
-                .withNewSpec()
-                .withType("topic")
-                .withAddress("test-topic-standard")
-                .withPlan(DestinationPlan.STANDARD_LARGE_TOPIC)
-                .endSpec()
-                .build();
-        Address standardSub = new AddressBuilder()
-                .withNewMetadata()
-                .withNamespace(getSharedAddressSpace().getMetadata().getNamespace())
-                .withName(AddressUtils.generateAddressMetadataName(getSharedAddressSpace(), "test-sub-standard"))
-                .endMetadata()
-                .withNewSpec()
-                .withType("subscription")
-                .withAddress("test-sub-standard")
-                .withTopic(standardTopic.getSpec().getAddress())
-                .withPlan(DestinationPlan.STANDARD_SMALL_SUBSCRIPTION)
-                .endSpec()
-                .build();
-        doTestTopicPersistentMessages(standardTopic, standardSub);
-    }
-
-}
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/shared/standard/api/UserApiTest.java b/systemtests/src/test/java/io/enmasse/systemtest/shared/standard/api/UserApiTest.java
new file mode 100644
index 000000000..fa79ef699
--- /dev/null
+++ b/systemtests/src/test/java/io/enmasse/systemtest/shared/standard/api/UserApiTest.java
@@ -0,0 +1,137 @@
+/*
+ * 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.standard.api;
+
+import io.enmasse.address.model.Address;
+import io.enmasse.address.model.AddressBuilder;
+import io.enmasse.address.model.AddressSpace;
+import io.enmasse.systemtest.UserCredentials;
+import io.enmasse.systemtest.amqp.AmqpClient;
+import io.enmasse.systemtest.amqp.UnauthorizedAccessException;
+import io.enmasse.systemtest.bases.TestBase;
+import io.enmasse.systemtest.bases.shared.ITestSharedStandard;
+import io.enmasse.systemtest.cmdclients.KubeCMDClient;
+import io.enmasse.systemtest.logs.CustomLogger;
+import io.enmasse.systemtest.model.addressplan.DestinationPlan;
+import io.enmasse.systemtest.utils.AddressUtils;
+import io.enmasse.systemtest.utils.UserUtils;
+import io.enmasse.user.model.v1.DoneableUser;
+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.Test;
+import org.slf4j.Logger;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class UserApiTest extends TestBase implements ITestSharedStandard {
+
+    private Map<AddressSpace, User> users = new HashMap<>();
+    private Logger LOGGER = CustomLogger.getLogger();
+
+    @AfterEach
+    void cleanUsers() {
+        users.forEach((addressSpace, user) -> {
+            try {
+                resourcesManager.removeUser(addressSpace, user);
+            } catch (Exception e) {
+                LOGGER.info("Clean: User not exists {}", user.getSpec().getUsername());
+            }
+        });
+    }
+
+    @Test
+    void testUpdateUserPermissionsUserAPI() throws Exception {
+        Address queue = new AddressBuilder()
+                .withNewMetadata()
+                .withNamespace(getSharedAddressSpace().getMetadata().getNamespace())
+                .withName(AddressUtils.generateAddressMetadataName(getSharedAddressSpace(), "test-queue"))
+                .endMetadata()
+                .withNewSpec()
+                .withType("queue")
+                .withAddress("test-queue")
+                .withPlan(DestinationPlan.STANDARD_LARGE_QUEUE)
+                .endSpec()
+                .build();
+        resourcesManager.setAddresses(queue);
+
+        UserCredentials cred = new UserCredentials("pepa", "pepapw");
+        User testUser = UserUtils.createUserResource(cred)
+                .editSpec()
+                .withAuthorization(Collections.singletonList(
+                        new UserAuthorizationBuilder()
+                                .withAddresses(queue.getSpec().getAddress())
+                                .withOperations(Operation.send).build()))
+                .endSpec()
+                .done();
+
+        users.put(getSharedAddressSpace(), testUser);
+        testUser = resourcesManager.createOrUpdateUser(getSharedAddressSpace(), testUser);
+
+        AmqpClient client = getAmqpClientFactory().createQueueClient(getSharedAddressSpace());
+        client.getConnectOptions().setCredentials(cred);
+        assertThat(client.sendMessages(queue.getSpec().getAddress(), Arrays.asList("kuk", "puk")).get(1, TimeUnit.MINUTES), is(2));
+
+        testUser = new DoneableUser(testUser)
+                .editSpec()
+                .withAuthorization(Collections.singletonList(new UserAuthorizationBuilder()
+                        .withAddresses(queue.getSpec().getAddress())
+                        .withOperations(Operation.recv).build()))
+                .endSpec()
+                .done();
+
+        resourcesManager.createOrUpdateUser(getSharedAddressSpace(), testUser);
+        Throwable exception = assertThrows(ExecutionException.class,
+                () -> client.sendMessages(queue.getSpec().getAddress(), Arrays.asList("kuk", "puk")).get(10, TimeUnit.SECONDS));
+        assertTrue(exception.getCause() instanceof UnauthorizedAccessException);
+        assertThat(client.recvMessages(queue.getSpec().getAddress(), 2).get(1, TimeUnit.MINUTES).size(), is(2));
+    }
+
+    @Test
+    void testServiceaccountUser() throws Exception {
+        Address queue = new AddressBuilder()
+                .withNewMetadata()
+                .withNamespace(getSharedAddressSpace().getMetadata().getNamespace())
+                .withName(AddressUtils.generateAddressMetadataName(getSharedAddressSpace(), "test-queue"))
+                .endMetadata()
+                .withNewSpec()
+                .withType("queue")
+                .withAddress("test-queue")
+                .withPlan(DestinationPlan.STANDARD_LARGE_QUEUE)
+                .endSpec()
+                .build();
+        resourcesManager.setAddresses(queue);
+        UserCredentials serviceAccount = new UserCredentials("test-service-account", "");
+        try {
+            resourcesManager.createUserServiceAccount(getSharedAddressSpace(), serviceAccount);
+            UserCredentials messagingUser = new UserCredentials("@@serviceaccount@@",
+                    kubernetes.getServiceaccountToken(serviceAccount.getUsername(), environment.namespace()));
+            LOGGER.info("username: {}, password: {}", messagingUser.getUsername(), messagingUser.getPassword());
+
+            getClientUtils().assertCanConnect(getSharedAddressSpace(), messagingUser, Collections.singletonList(queue), resourcesManager);
+
+            //delete user
+            assertThat("User deleting failed using oc cmd",
+                    KubeCMDClient.deleteUser(kubernetes.getInfraNamespace(), getSharedAddressSpace().getMetadata().getName(), serviceAccount.getUsername()).getRetCode(), is(true));
+            assertThat("User is still present",
+                    KubeCMDClient.getUser(kubernetes.getInfraNamespace(), getSharedAddressSpace().getMetadata().getName(), serviceAccount.getUsername()).getRetCode(), is(false));
+
+            getClientUtils().assertCannotConnect(getSharedAddressSpace(), messagingUser, Collections.singletonList(queue), resourcesManager);
+        } finally {
+            kubernetes.deleteServiceAccount("test-service-account", environment.namespace());
+        }
+    }
+}