Repository: EnMasseProject/enmasse
Pull request #2605 (2019-04-09 11:27:41)

Fix plural spelling of resources


Patch:
diff --git a/address-space-controller/src/main/java/io/enmasse/controller/StatusController.java b/address-space-controller/src/main/java/io/enmasse/controller/StatusController.java
index 3be2887e2..5b2b564ea 100644
--- a/address-space-controller/src/main/java/io/enmasse/controller/StatusController.java
+++ b/address-space-controller/src/main/java/io/enmasse/controller/StatusController.java
@@ -19,10 +19,7 @@ import org.slf4j.LoggerFactory;
 
 import static io.enmasse.controller.InfraConfigs.parseCurrentInfraConfig;
 
-import java.util.HashSet;
-import java.util.List;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
 import java.util.stream.Collectors;
 
 public class StatusController implements Controller {
@@ -45,9 +42,29 @@ public class StatusController implements Controller {
     public AddressSpace reconcile(AddressSpace addressSpace) {
         checkComponentsReady(addressSpace);
         checkAuthServiceReady(addressSpace);
+        checkExposedEndpoints(addressSpace);
         return addressSpace;
     }
 
+    private void checkExposedEndpoints(AddressSpace addressSpace) {
+        Map<String, EndpointSpec> exposedEndpoints = new HashMap<>();
+        for (EndpointSpec endpointSpec : addressSpace.getSpec().getEndpoints()) {
+            if (endpointSpec.getExpose() != null && endpointSpec.getExpose().getType().equals(ExposeType.route)) {
+                exposedEndpoints.put(endpointSpec.getName(), endpointSpec);
+            }
+        }
+
+        for (EndpointStatus endpointStatus : addressSpace.getStatus().getEndpointStatuses()) {
+            if (exposedEndpoints.containsKey(endpointStatus.getName())) {
+                if (endpointStatus.getExternalHost() == null) {
+                    String msg = String.format("Endpoint '%s' is not yet exposed", endpointStatus.getName());
+                    addressSpace.getStatus().setReady(false);
+                    addressSpace.getStatus().appendMessage(msg);
+                }
+            }
+        }
+    }
+
     private InfraConfig getInfraConfig(AddressSpace addressSpace) {
         AddressSpaceResolver addressSpaceResolver = new AddressSpaceResolver(schemaProvider.getSchema());
         return addressSpaceResolver.getInfraConfig(addressSpace.getSpec().getType(), addressSpace.getSpec().getPlan());
diff --git a/address-space-controller/src/main/java/io/enmasse/controller/keycloak/Keycloak.java b/address-space-controller/src/main/java/io/enmasse/controller/keycloak/Keycloak.java
index 899452feb..5b2965251 100644
--- a/address-space-controller/src/main/java/io/enmasse/controller/keycloak/Keycloak.java
+++ b/address-space-controller/src/main/java/io/enmasse/controller/keycloak/Keycloak.java
@@ -44,7 +44,14 @@ public class Keycloak implements KeycloakApi {
         if (keycloakMap.get(authenticationService.getMetadata().getName()) == null) {
             keycloakMap.put(authenticationService.getMetadata().getName(), keycloakFactory.createInstance(authenticationService));
         }
-        return consumer.handle(keycloakMap.get(authenticationService.getMetadata().getName()));
+        org.keycloak.admin.client.Keycloak keycloak = keycloakMap.get(authenticationService.getMetadata().getName());
+        try {
+            return consumer.handle(keycloak);
+        } catch (Exception e) {
+            keycloakMap.remove(authenticationService.getMetadata().getName());
+            keycloak.close();
+            throw e;
+        }
     }
 
     @Override
diff --git a/api-model/src/main/java/io/enmasse/common/model/CustomResources.java b/api-model/src/main/java/io/enmasse/common/model/CustomResources.java
index d51cae2a6..b71e948a4 100644
--- a/api-model/src/main/java/io/enmasse/common/model/CustomResources.java
+++ b/api-model/src/main/java/io/enmasse/common/model/CustomResources.java
@@ -15,6 +15,11 @@ public final class CustomResources {
         String singular = kind.toLowerCase();
         String listKind = kind + "List";
         String plural = singular + "s";
+        if (singular.endsWith("s")) {
+            plural = singular + "es";
+        } else if (singular.endsWith("y")) {
+            plural = singular.substring(0, singular.length() - 1) + "ies";
+        }
         return new CustomResourceDefinitionBuilder()
                         .editOrNewMetadata()
                         .withName(plural + "." + group)
diff --git a/api-model/src/main/java/io/enmasse/user/api/DelegateUserApi.java b/api-model/src/main/java/io/enmasse/user/api/DelegateUserApi.java
index c0bfb3245..02d35cc8c 100644
--- a/api-model/src/main/java/io/enmasse/user/api/DelegateUserApi.java
+++ b/api-model/src/main/java/io/enmasse/user/api/DelegateUserApi.java
@@ -20,7 +20,7 @@ public class DelegateUserApi implements UserApi {
     }
 
     @Override
-    public boolean isAvailable(AuthenticationService authenticationService) {
+    public boolean isAvailable(AuthenticationService authenticationService) throws Exception {
         return userApiMap.get(authenticationService.getSpec().getType()).isAvailable(authenticationService);
     }
 
@@ -45,32 +45,32 @@ public class DelegateUserApi implements UserApi {
     }
 
     @Override
-    public boolean realmExists(AuthenticationService authenticationService, String realm) {
+    public boolean realmExists(AuthenticationService authenticationService, String realm) throws Exception {
         return userApiMap.get(authenticationService.getSpec().getType()).realmExists(authenticationService, realm);
     }
 
     @Override
-    public UserList listUsers(AuthenticationService authenticationService, String namespace) {
+    public UserList listUsers(AuthenticationService authenticationService, String namespace) throws Exception {
         return userApiMap.get(authenticationService.getSpec().getType()).listUsers(authenticationService, namespace);
     }
 
     @Override
-    public UserList listUsersWithLabels(AuthenticationService authenticationService, String namespace, Map<String, String> labels) {
+    public UserList listUsersWithLabels(AuthenticationService authenticationService, String namespace, Map<String, String> labels) throws Exception {
         return userApiMap.get(authenticationService.getSpec().getType()).listUsersWithLabels(authenticationService, namespace, labels);
     }
 
     @Override
-    public UserList listAllUsers(AuthenticationService authenticationService) {
+    public UserList listAllUsers(AuthenticationService authenticationService) throws Exception {
         return userApiMap.get(authenticationService.getSpec().getType()).listAllUsers(authenticationService);
     }
 
     @Override
-    public UserList listAllUsersWithLabels(AuthenticationService authenticationService, Map<String, String> labels) {
+    public UserList listAllUsersWithLabels(AuthenticationService authenticationService, Map<String, String> labels) throws Exception {
         return userApiMap.get(authenticationService.getSpec().getType()).listAllUsersWithLabels(authenticationService, labels);
     }
 
     @Override
-    public void deleteUsers(AuthenticationService authenticationService, String namespace) {
+    public void deleteUsers(AuthenticationService authenticationService, String namespace) throws Exception {
         userApiMap.get(authenticationService.getSpec().getType()).deleteUsers(authenticationService, namespace);
     }
 }
diff --git a/api-model/src/main/java/io/enmasse/user/api/UserApi.java b/api-model/src/main/java/io/enmasse/user/api/UserApi.java
index 6d405d886..12a99835b 100644
--- a/api-model/src/main/java/io/enmasse/user/api/UserApi.java
+++ b/api-model/src/main/java/io/enmasse/user/api/UserApi.java
@@ -12,18 +12,18 @@ import java.util.Map;
 import java.util.Optional;
 
 public interface UserApi {
-    boolean isAvailable(AuthenticationService authenticationService);
+    boolean isAvailable(AuthenticationService authenticationService) throws Exception;
     Optional<User> getUserWithName(AuthenticationService authenticationService, String realm, String name) throws Exception;
     void createUser(AuthenticationService authenticationService, String realm, User user) throws Exception;
     boolean replaceUser(AuthenticationService authenticationService, String realm, User user) throws Exception;
     void deleteUser(AuthenticationService authenticationService, String realm, User user) throws Exception;
 
-    boolean realmExists(AuthenticationService authenticationService, String realm);
+    boolean realmExists(AuthenticationService authenticationService, String realm) throws Exception;
 
-    UserList listUsers(AuthenticationService authenticationService, String namespace);
-    UserList listUsersWithLabels(AuthenticationService authenticationService, String namespace, Map<String, String> labels);
-    UserList listAllUsers(AuthenticationService authenticationService);
-    UserList listAllUsersWithLabels(AuthenticationService authenticationService, Map<String, String> labels);
+    UserList listUsers(AuthenticationService authenticationService, String namespace) throws Exception;
+    UserList listUsersWithLabels(AuthenticationService authenticationService, String namespace, Map<String, String> labels) throws Exception;
+    UserList listAllUsers(AuthenticationService authenticationService) throws Exception;
+    UserList listAllUsersWithLabels(AuthenticationService authenticationService, Map<String, String> labels) throws Exception;
 
-    void deleteUsers(AuthenticationService authenticationService, String namespace);
+    void deleteUsers(AuthenticationService authenticationService, String namespace) throws Exception;
 }
diff --git a/api-server/src/main/resources/swagger.json b/api-server/src/main/resources/swagger.json
index 060e57442..84f8bd879 100644
--- a/api-server/src/main/resources/swagger.json
+++ b/api-server/src/main/resources/swagger.json
@@ -2063,9 +2063,6 @@
                 "spec": {
                     "type": "object",
                     "properties": {
-                        "resources": {
-                            "type": "object"
-                        },
                         "affinity": {
                             "type": "object"
                         },
@@ -2077,6 +2074,17 @@
                         },
                         "priorityClassName": {
                             "type": "string"
+                        },
+                        "containers": {
+                            "type": "array",
+                            "items": {
+                                "type": "object",
+                                "properties": {
+                                    "resources": {
+                                        "type": "object"
+                                    }
+                                }
+                            }
                         }
                     }
                 }
diff --git a/documentation/common/restapi-reference.adoc b/documentation/common/restapi-reference.adoc
index a990e326e..e58982a0f 100644
--- a/documentation/common/restapi-reference.adoc
+++ b/documentation/common/restapi-reference.adoc
@@ -1488,14 +1488,24 @@ __optional__|object
 |Name|Schema
 |**affinity** +
 __optional__|object
+|**containers** +
+__optional__|< <<_io_enmasse_admin_v1beta1_infraconfigpodspec_containers,containers>> > array
 |**priorityClassName** +
 __optional__|string
-|**resources** +
-__optional__|object
 |**tolerations** +
 __optional__|< object > array
 |===
 
+[[_io_enmasse_admin_v1beta1_infraconfigpodspec_containers]]
+**containers**
+
+[options="header", cols=".^3a,.^4a"]
+|===
+|Name|Schema
+|**resources** +
+__optional__|object
+|===
+
 
 [[_io_enmasse_admin_v1beta1_standardinfraconfig]]
 === io.enmasse.admin.v1beta1.StandardInfraConfig
diff --git a/iot/examples/k8s-tls/create b/iot/examples/k8s-tls/create
index babfc7649..22220a6a1 100755
--- a/iot/examples/k8s-tls/create
+++ b/iot/examples/k8s-tls/create
@@ -30,7 +30,7 @@ cat "$OUTDIR/ca-cert.pem" "$OUTDIR/root-cert.pem" > "$OUTDIR/trusted-certs.pem"
 # Export information
 
 CA_SUBJECT=$(openssl x509 -in "$OUTDIR/ca-cert.pem" -noout -subject -nameopt RFC2253 | sed s/^subject=//)
-PK=$(openssl x509 -in "$OUTDIR/ca-cert.pem" -noout -pubkey | sed /^---/d | sed -z 's/\n//g')
+PK=$(openssl x509 -in "$OUTDIR/ca-cert.pem" -noout -pubkey | sed /^---/d)
 
 echo $CA_SUBJECT
 echo $PK
diff --git a/iot/iot-auth-service/Dockerfile b/iot/iot-auth-service/Dockerfile
index aab70a287..0dc55a8a5 100644
--- a/iot/iot-auth-service/Dockerfile
+++ b/iot/iot-auth-service/Dockerfile
@@ -1,4 +1,4 @@
-FROM enmasseproject/java-base:11-0
+FROM enmasseproject/java-base:11-1
 
 ARG version
 ARG maven_version
@@ -7,4 +7,5 @@ ENV VERSION=${version} COMMIT=${commit} MAVEN_VERSION=${maven_version}
 
 ADD target/iot-auth-service-${MAVEN_VERSION}.jar /iot-auth-service.jar
 
+ENV JAVA_LAUNCH_PROFILE=openjdk-11
 CMD ["/opt/run-java/launch_java.sh", "/iot-auth-service.jar"]
diff --git a/iot/iot-device-registry-file/Dockerfile b/iot/iot-device-registry-file/Dockerfile
index f4ee88aa1..5d22fb7e1 100644
--- a/iot/iot-device-registry-file/Dockerfile
+++ b/iot/iot-device-registry-file/Dockerfile
@@ -1,4 +1,4 @@
-FROM enmasseproject/java-base:11-0
+FROM enmasseproject/java-base:11-1
 
 ARG version
 ARG maven_version
@@ -7,4 +7,5 @@ ENV VERSION=${version} COMMIT=${commit} MAVEN_VERSION=${maven_version}
 
 ADD target/iot-device-registry-file-${MAVEN_VERSION}.jar /iot-device-registry-file.jar
 
+ENV JAVA_LAUNCH_PROFILE=openjdk-11
 CMD ["/opt/run-java/launch_java.sh", "/iot-device-registry-file.jar"]
diff --git a/iot/iot-http-adapter/Dockerfile b/iot/iot-http-adapter/Dockerfile
index 0d92cff53..9c0ec5797 100644
--- a/iot/iot-http-adapter/Dockerfile
+++ b/iot/iot-http-adapter/Dockerfile
@@ -1,4 +1,4 @@
-FROM enmasseproject/java-base:11-0
+FROM enmasseproject/java-base:11-1
 
 ARG version
 ARG maven_version
@@ -7,4 +7,5 @@ ENV VERSION=${version} COMMIT=${commit} MAVEN_VERSION=${maven_version}
 
 ADD target/iot-http-adapter-${MAVEN_VERSION}.jar /iot-http-adapter.jar
 
+ENV JAVA_LAUNCH_PROFILE=openjdk-11
 CMD ["/opt/run-java/launch_java.sh", "/iot-http-adapter.jar"]
diff --git a/iot/iot-mqtt-adapter/Dockerfile b/iot/iot-mqtt-adapter/Dockerfile
index 79213cf33..c704a1709 100644
--- a/iot/iot-mqtt-adapter/Dockerfile
+++ b/iot/iot-mqtt-adapter/Dockerfile
@@ -1,4 +1,4 @@
-FROM enmasseproject/java-base:11-0
+FROM enmasseproject/java-base:11-1
 
 ARG version
 ARG maven_version
@@ -7,4 +7,5 @@ ENV VERSION=${version} COMMIT=${commit} MAVEN_VERSION=${maven_version}
 
 ADD target/iot-mqtt-adapter-${MAVEN_VERSION}.jar /iot-mqtt-adapter.jar
 
+ENV JAVA_LAUNCH_PROFILE=openjdk-11
 CMD ["/opt/run-java/launch_java.sh", "/iot-mqtt-adapter.jar"]
diff --git a/iot/iot-tenant-service/Dockerfile b/iot/iot-tenant-service/Dockerfile
index 4440c0cf7..1b6bd833c 100644
--- a/iot/iot-tenant-service/Dockerfile
+++ b/iot/iot-tenant-service/Dockerfile
@@ -1,4 +1,4 @@
-FROM enmasseproject/java-base:11-0
+FROM enmasseproject/java-base:11-1
 
 ARG version
 ARG maven_version
@@ -7,4 +7,5 @@ ENV VERSION=${version} COMMIT=${commit} MAVEN_VERSION=${maven_version}
 
 ADD target/iot-tenant-service-${MAVEN_VERSION}.jar /iot-tenant-service.jar
 
+ENV JAVA_LAUNCH_PROFILE=openjdk-11
 CMD ["/opt/run-java/launch_java.sh", "/iot-tenant-service.jar"]
diff --git a/keycloak-user-api/src/main/java/io/enmasse/user/keycloak/KeycloakUserApi.java b/keycloak-user-api/src/main/java/io/enmasse/user/keycloak/KeycloakUserApi.java
index d7a5f38b1..4a3bff35d 100644
--- a/keycloak-user-api/src/main/java/io/enmasse/user/keycloak/KeycloakUserApi.java
+++ b/keycloak-user-api/src/main/java/io/enmasse/user/keycloak/KeycloakUserApi.java
@@ -58,14 +58,21 @@ public class KeycloakUserApi implements UserApi {
     }
 
     interface KeycloakHandler<T> {
-        T handle(Keycloak keycloak);
+        T handle(Keycloak keycloak) throws Exception;
     }
 
-    private synchronized <T> T withKeycloak(AuthenticationService authenticationService, KeycloakHandler<T> consumer) {
+    private synchronized <T> T withKeycloak(AuthenticationService authenticationService, KeycloakHandler<T> consumer) throws Exception {
         if (keycloakMap.get(authenticationService.getMetadata().getName()) == null) {
             keycloakMap.put(authenticationService.getMetadata().getName(), keycloakFactory.createInstance(authenticationService));
         }
-        return consumer.handle(keycloakMap.get(authenticationService.getMetadata().getName()));
+        Keycloak keycloak = keycloakMap.get(authenticationService.getMetadata().getName());
+        try {
+            return consumer.handle(keycloak);
+        } catch (Exception e) {
+            keycloakMap.remove(authenticationService.getMetadata().getName());
+            keycloak.close();
+            throw e;
+        }
     }
 
     interface RealmHandler<T> {
@@ -73,12 +80,10 @@ public class KeycloakUserApi implements UserApi {
     }
 
     private synchronized <T> T withRealm(AuthenticationService authenticationService, String realmName, RealmHandler<T> consumer) throws Exception {
-        if (keycloakMap.get(authenticationService.getMetadata().getName()) == null) {
-            keycloakMap.put(authenticationService.getMetadata().getName(), keycloakFactory.createInstance(authenticationService));
-        }
-        Keycloak keycloak = keycloakMap.get(authenticationService.getMetadata().getName());
-        RealmResource realmResource = waitForRealm(keycloak, realmName, apiTimeout);
-        return consumer.handle(realmResource);
+        return withKeycloak(authenticationService, keycloak -> {
+            RealmResource realmResource = waitForRealm(keycloak, realmName, apiTimeout);
+            return consumer.handle(realmResource);
+        });
     }
 
     private RealmResource waitForRealm(Keycloak keycloak, String realmName, Duration timeout) throws Exception {
@@ -117,12 +122,8 @@ public class KeycloakUserApi implements UserApi {
     }
 
     @Override
-    public synchronized boolean isAvailable(AuthenticationService authenticationService) {
-        if (keycloakMap.get(authenticationService.getMetadata().getName()) == null) {
-            Keycloak keycloak = keycloakFactory.createInstance(authenticationService);
-            keycloakMap.put(authenticationService.getMetadata().getName(), keycloak);
-        }
-        return keycloakMap.get(authenticationService.getMetadata().getName()) != null;
+    public synchronized boolean isAvailable(AuthenticationService authenticationService) throws Exception {
+        return withKeycloak(authenticationService, Objects::nonNull);
     }
 
     @Override
@@ -461,11 +462,11 @@ public class KeycloakUserApi implements UserApi {
     }
 
     @Override
-    public boolean realmExists(AuthenticationService authenticationService, String realmName) {
+    public boolean realmExists(AuthenticationService authenticationService, String realmName) throws Exception {
         return withKeycloak(authenticationService, kc -> getRealmResource(kc, realmName) != null);
     }
 
-    private UserList queryUsers(AuthenticationService authenticationService, final Predicate<RealmRepresentation> realmPredicate, final Predicate<UserRepresentation> userPredicate) {
+    private UserList queryUsers(AuthenticationService authenticationService, final Predicate<RealmRepresentation> realmPredicate, final Predicate<UserRepresentation> userPredicate) throws Exception {
         return withKeycloak(authenticationService, keycloak -> {
 
             List<RealmRepresentation> realmReps = keycloak.realms().findAll();
@@ -493,7 +494,7 @@ public class KeycloakUserApi implements UserApi {
      * List all users, from all namespaces.
      */
     @Override
-    public UserList listAllUsers(AuthenticationService authenticationService) {
+    public UserList listAllUsers(AuthenticationService authenticationService) throws Exception {
         return queryUsers(authenticationService,
                         realm -> getAttribute(realm, "namespace").isPresent(),
                         user -> true);
@@ -503,7 +504,7 @@ public class KeycloakUserApi implements UserApi {
      * List users from a single namespace.
      */
     @Override
-    public UserList listUsers(AuthenticationService authenticationService, final String namespace) {
+    public UserList listUsers(AuthenticationService authenticationService, final String namespace) throws Exception {
         return queryUsers(authenticationService,
                         realm -> hasAttribute(realm, "namespace", namespace),
                         user -> true);
@@ -635,21 +636,21 @@ public class KeycloakUserApi implements UserApi {
     }
 
     @Override
-    public UserList listUsersWithLabels(AuthenticationService authenticationService, String namespace, Map<String, String> labels) {
+    public UserList listUsersWithLabels(AuthenticationService authenticationService, String namespace, Map<String, String> labels) throws Exception {
         return queryUsers(authenticationService,
                         realm -> hasAttribute(realm, "namespace", namespace),
                         user -> matchesLabels(user, labels));
     }
 
     @Override
-    public UserList listAllUsersWithLabels(AuthenticationService authenticationService, final Map<String, String> labels) {
+    public UserList listAllUsersWithLabels(AuthenticationService authenticationService, final Map<String, String> labels) throws Exception {
         return queryUsers(authenticationService,
                         realm -> getAttribute(realm, "namespace").isPresent(),
                         user -> matchesLabels(user, labels));
     }
 
     @Override
-    public void deleteUsers(AuthenticationService authenticationService, String namespace) {
+    public void deleteUsers(AuthenticationService authenticationService, String namespace) throws Exception {
         withKeycloak(authenticationService, keycloak -> {
             List<RealmRepresentation> realmReps = keycloak.realms().findAll();
             for (RealmRepresentation realmRep : realmReps) {
diff --git a/router/Dockerfile b/router/Dockerfile
index ac572ff28..a60edf28d 100644
--- a/router/Dockerfile
+++ b/router/Dockerfile
@@ -1,4 +1,4 @@
-FROM enmasseproject/qdrouterd-base:1.6.0-rc2
+FROM enmasseproject/qdrouterd-base:1.6.0-patched
 ARG version
 ARG maven_version
 ARG commit
diff --git a/templates/crds/brokeredinfraconfigs.crd.yaml b/templates/crds/brokeredinfraconfigs.crd.yaml
index f3a326bf9..60b682ec4 100644
--- a/templates/crds/brokeredinfraconfigs.crd.yaml
+++ b/templates/crds/brokeredinfraconfigs.crd.yaml
@@ -55,10 +55,17 @@ spec:
                           type: object
                         tolerations:
                           type: array
+                          items:
+                            type: object
                         priorityClassName:
                           type: string
-                        resources:
-                          type: object
+                        containers:
+                          type: array
+                          items:
+                            type: object
+                            properties:
+                              resources:
+                                type: object
                 resources:
                   type: object
                   properties:
diff --git a/templates/crds/standardinfraconfigs.crd.yaml b/templates/crds/standardinfraconfigs.crd.yaml
index 6c00e9ecc..77d7cf5e9 100644
--- a/templates/crds/standardinfraconfigs.crd.yaml
+++ b/templates/crds/standardinfraconfigs.crd.yaml
@@ -60,11 +60,17 @@ spec:
                           type: object
                         tolerations:
                           type: array
+                          items:
+                            type: object
                         priorityClassName:
                           type: string
-                        resources:
-                          type: object
-
+                        containers:
+                          type: array
+                          items:
+                            type: object
+                            properties:
+                              resources:
+                                type: object
             broker:
               type: object
               properties:


Test patch:
diff --git a/api-model/src/test/java/io/enmasse/common/model/CustomResourcesTest.java b/api-model/src/test/java/io/enmasse/common/model/CustomResourcesTest.java
new file mode 100644
index 000000000..7f8df3a66
--- /dev/null
+++ b/api-model/src/test/java/io/enmasse/common/model/CustomResourcesTest.java
@@ -0,0 +1,23 @@
+/*
+ * 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.common.model;
+
+import io.fabric8.kubernetes.api.model.apiextensions.CustomResourceDefinition;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class CustomResourcesTest {
+    @Test
+    public void testPural() {
+        CustomResourceDefinition a = CustomResources.createCustomResource("enmasse.io", "v1alpha1", "Address");
+        CustomResourceDefinition b = CustomResources.createCustomResource("enmasse.io", "v1alpha1", "Addressy");
+        CustomResourceDefinition c = CustomResources.createCustomResource("enmasse.io", "v1alpha1", "Addr");
+
+        assertEquals("addresses", a.getSpec().getNames().getPlural());
+        assertEquals("addressies", b.getSpec().getNames().getPlural());
+        assertEquals("addrs", c.getSpec().getNames().getPlural());
+    }
+}
diff --git a/api-server/src/test/java/io/enmasse/api/server/HTTPServerTest.java b/api-server/src/test/java/io/enmasse/api/server/HTTPServerTest.java
index 8d3250eea..c421839f8 100644
--- a/api-server/src/test/java/io/enmasse/api/server/HTTPServerTest.java
+++ b/api-server/src/test/java/io/enmasse/api/server/HTTPServerTest.java
@@ -66,7 +66,7 @@ public class HTTPServerTest {
     }
 
     @BeforeEach
-    public void setup(VertxTestContext context) {
+    public void setup(VertxTestContext context) throws Exception {
         vertx = Vertx.vertx();
         addressSpaceApi = new TestAddressSpaceApi();
         addressSpace = createAddressSpace("ns", "myinstance");
diff --git a/systemtests/pom.xml b/systemtests/pom.xml
index 35b2e9f81..868219761 100644
--- a/systemtests/pom.xml
+++ b/systemtests/pom.xml
@@ -242,14 +242,14 @@
         <profile>
             <id>shared-standard-release</id>
             <properties>
-                <groups>shared-standard &amp; !noneAuth &amp; !smoke &amp; !marathon</groups>
+                <groups>shared-standard &amp; !noneAuth &amp; !smoke &amp; !marathon &amp; !shared-iot</groups>
                 <skip.tests>false</skip.tests>
             </properties>
         </profile>
         <profile>
             <id>shared-brokered-release</id>
             <properties>
-                <groups>shared-brokered &amp; !noneAuth &amp; !smoke &amp; !marathon</groups>
+                <groups>shared-brokered &amp; !noneAuth &amp; !smoke &amp; !marathon &amp; !shared-iot</groups>
                 <skip.tests>false</skip.tests>
             </properties>
         </profile>
@@ -260,5 +260,12 @@
                 <skip.tests>false</skip.tests>
             </properties>
         </profile>
+        <profile>
+            <id>iot-release</id>
+            <properties>
+                <groups>shared-iot &amp; !noneAuth &amp; !smoke &amp; !marathon</groups>
+                <skip.tests>false</skip.tests>
+            </properties>
+        </profile>
     </profiles>
 </project>
diff --git a/systemtests/scripts/docker-logs.sh b/systemtests/scripts/pod-logs.sh
similarity index 76%
rename from systemtests/scripts/docker-logs.sh
rename to systemtests/scripts/pod-logs.sh
index 974678c9c..70772a56c 100755
--- a/systemtests/scripts/docker-logs.sh
+++ b/systemtests/scripts/pod-logs.sh
@@ -17,7 +17,9 @@ while [[ true ]]; do
     for POD in ${PODS}; do
         CONTAINERS=$(${CMD} get pods ${POD} -o=jsonpath='{.spec.containers[*].name}')
         for CONTAINER in ${CONTAINERS}; do
-            ${CMD} logs ${POD} --container=${CONTAINER} > ${LOG_DIR}/${POD}.${CONTAINER}.log
+            if [[ $(${CMD} logs ${POD} --container=${CONTAINER}) ]]; then
+                ${CMD} logs ${POD} --container=${CONTAINER} > ${LOG_DIR}/${POD}.${CONTAINER}.log
+            fi
         done
     done
     sleep 2
diff --git a/systemtests/scripts/run_test_component.sh b/systemtests/scripts/run_test_component.sh
index 20f573ae9..924587f9f 100755
--- a/systemtests/scripts/run_test_component.sh
+++ b/systemtests/scripts/run_test_component.sh
@@ -34,8 +34,8 @@ export KUBERNETES_NAMESPACE=${OPENSHIFT_PROJECT}
 export_required_env
 
 #start docker logging
-DOCKER_LOG_DIR="${ARTIFACTS_DIR}/docker-logs"
-${CURDIR}/docker-logs.sh ${DOCKER_LOG_DIR} ${KUBERNETES_NAMESPACE} > /dev/null 2> /dev/null &
+DOCKER_LOG_DIR="${ARTIFACTS_DIR}/pod-logs"
+${CURDIR}/pod-logs.sh ${DOCKER_LOG_DIR} ${KUBERNETES_NAMESPACE} > /dev/null 2> /dev/null &
 LOGS_PID=$!
 info "process for syncing docker logs is running with PID: ${LOGS_PID}"
 
diff --git a/systemtests/scripts/run_test_kubernetes.sh b/systemtests/scripts/run_test_kubernetes.sh
index 37361c145..ae892df8a 100755
--- a/systemtests/scripts/run_test_kubernetes.sh
+++ b/systemtests/scripts/run_test_kubernetes.sh
@@ -91,8 +91,8 @@ get_kubernetes_info ${LOG_DIR} pods default "-before"
 
 if [[ -z "${DISABLE_LOG_SYNC}" ]]; then
     #start docker logging
-    DOCKER_LOG_DIR="${ARTIFACTS_DIR}/docker-logs"
-    ${CURDIR}/docker-logs.sh ${DOCKER_LOG_DIR} ${KUBERNETES_NAMESPACE} > /dev/null 2> /dev/null &
+    DOCKER_LOG_DIR="${ARTIFACTS_DIR}/pod-logs"
+    ${CURDIR}/pod-logs.sh ${DOCKER_LOG_DIR} ${KUBERNETES_NAMESPACE} > /dev/null 2> /dev/null &
     LOGS_PID=$!
     echo "process for syncing docker logs is running with PID: ${LOGS_PID}"
 fi
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/CertBundle.java b/systemtests/src/main/java/io/enmasse/systemtest/CertBundle.java
index 9e75a1439..faaf79e74 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/CertBundle.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/CertBundle.java
@@ -4,30 +4,24 @@
  */
 package io.enmasse.systemtest;
 
-import java.io.File;
+import java.util.Base64;
 
 public class CertBundle {
 
-    private File caCert;
-    private File crtFile;
+    private String caCert;
     private String key;
     private String cert;
 
-    public CertBundle(File caCert, File crtFile, String key, String cert) {
+    public CertBundle(String caCert, String key, String cert) {
         this.caCert = caCert;
-        this.crtFile = crtFile;
         this.key = key;
         this.cert = cert;
     }
 
-    public File getCaCert() {
+    public String getCaCert() {
         return caCert;
     }
 
-    public File getCrtFile() {
-        return crtFile;
-    }
-
     public String getKey() {
         return key;
     }
@@ -36,4 +30,13 @@ public class CertBundle {
         return cert;
     }
 
+    public String getCertB64() {
+        return Base64.getEncoder().encodeToString(cert.getBytes());
+    }
+
+    public String getKeyB64() {
+        return Base64.getEncoder().encodeToString(key.getBytes());
+    }
+
+
 }
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/Kubernetes.java b/systemtests/src/main/java/io/enmasse/systemtest/Kubernetes.java
index c2201a4a5..825382a06 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/Kubernetes.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/Kubernetes.java
@@ -45,6 +45,10 @@ public abstract class Kubernetes {
         return globalNamespace;
     }
 
+    public KubernetesClient getClient() {
+        return client;
+    }
+
     private static int getPort(Service service, String portName) {
         List<ServicePort> ports = service.getSpec().getPorts();
         for (ServicePort port : ports) {
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/amqp/AbstractSender.java b/systemtests/src/main/java/io/enmasse/systemtest/amqp/AbstractSender.java
new file mode 100644
index 000000000..6712067b8
--- /dev/null
+++ b/systemtests/src/main/java/io/enmasse/systemtest/amqp/AbstractSender.java
@@ -0,0 +1,59 @@
+/*
+ * Copyright 2018-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.amqp;
+
+import java.util.concurrent.CompletableFuture;
+import org.slf4j.Logger;
+
+import io.enmasse.systemtest.CustomLogger;
+import io.vertx.proton.ProtonConnection;
+import io.vertx.proton.ProtonSender;
+
+public abstract class AbstractSender<T> extends ClientHandlerBase<T> {
+
+    private static Logger log = CustomLogger.getLogger();
+
+    public AbstractSender(AmqpConnectOptions clientOptions, LinkOptions linkOptions, CompletableFuture<Void> connectPromise, CompletableFuture<T> resultPromise, String containerId) {
+        super(clientOptions, linkOptions, connectPromise, resultPromise, containerId);
+    }
+
+    protected abstract String getCurrentState();
+    protected abstract void sendMessages(ProtonConnection connection, ProtonSender sender);
+
+    @Override
+    public void connectionOpened(ProtonConnection connection) {
+        ProtonSender sender = connection.createSender(linkOptions.getTarget().getAddress());
+        sender.setTarget(linkOptions.getTarget());
+        sender.setQoS(clientOptions.getQos());
+        sender.openHandler(result -> {
+            if (result.succeeded()) {
+                log.info("Sender link '" + sender.getTarget().getAddress() + "' opened, sending messages");
+                connectPromise.complete(null);
+                sendMessages(connection, sender);
+            } else {
+                handleError(connection, sender.getRemoteCondition());
+            }
+        });
+        sender.closeHandler(result -> handleError(connection, sender.getRemoteCondition()));
+        sender.open();
+    }
+
+    @Override
+    protected void connectionClosed(ProtonConnection conn) {
+        conn.close();
+        resultPromise.completeExceptionally(new RuntimeException("Connection closed" + getCurrentState()));
+        connectPromise.completeExceptionally(new RuntimeException("Connection closed after " + getCurrentState()));
+    }
+
+    @Override
+    protected void connectionDisconnected(ProtonConnection conn) {
+        conn.close();
+        resultPromise.completeExceptionally(new RuntimeException("Connection disconnected after "  + getCurrentState()));
+        connectPromise.completeExceptionally(new RuntimeException("Connection disconnected after "  + getCurrentState()));
+    }
+
+
+}
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/amqp/AmqpClient.java b/systemtests/src/main/java/io/enmasse/systemtest/amqp/AmqpClient.java
index c2d6b772e..685db3659 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/amqp/AmqpClient.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/amqp/AmqpClient.java
@@ -8,6 +8,8 @@ package io.enmasse.systemtest.amqp;
 import io.enmasse.systemtest.Count;
 import io.enmasse.systemtest.VertxFactory;
 import io.vertx.core.Vertx;
+import io.vertx.proton.ProtonDelivery;
+
 import org.apache.qpid.proton.amqp.messaging.AmqpValue;
 import org.apache.qpid.proton.amqp.messaging.Source;
 import org.apache.qpid.proton.amqp.messaging.Target;
@@ -102,11 +104,11 @@ public class AmqpClient implements AutoCloseable {
         }
     }
 
-    public Future<Integer> sendMessages(String address, List<String> messages) {
+    public CompletableFuture<Integer> sendMessages(String address, List<String> messages) {
         return sendMessages(address, messages, new Count<>(messages.size()));
     }
 
-    public Future<Integer> sendMessages(String address, List<String> messages, Predicate<Message> predicate) {
+    public CompletableFuture<Integer> sendMessages(String address, List<String> messages, Predicate<Message> predicate) {
         List<Message> messageList = messages.stream()
                 .map(body -> {
                     Message message = Message.Factory.create();
@@ -118,11 +120,11 @@ public class AmqpClient implements AutoCloseable {
         return sendMessages(address, messageList, predicate);
     }
 
-    public Future<Integer> sendMessages(String address, Message... messages) {
+    public CompletableFuture<Integer> sendMessages(String address, Message... messages) {
         return sendMessages(address, Arrays.asList(messages), new Count<>(messages.length));
     }
 
-    public Future<Integer> sendMessages(String address, Iterable<Message> messages, Predicate<Message> predicate) {
+    public CompletableFuture<Integer> sendMessages(String address, Iterable<Message> messages, Predicate<Message> predicate) {
 
         CompletableFuture<Integer> resultPromise = new CompletableFuture<>();
         Vertx vertx = VertxFactory.create();
@@ -138,4 +140,23 @@ public class AmqpClient implements AutoCloseable {
         }
         return resultPromise;
     }
+
+    public CompletableFuture<List<ProtonDelivery>> sendMessage(String address, Message message) {
+
+        CompletableFuture<List<ProtonDelivery>> resultPromise = new CompletableFuture<>();
+        Vertx vertx = VertxFactory.create();
+        String containerId = "systemtest-sender-" + address;
+        CompletableFuture<Void> connectPromise = new CompletableFuture<>();
+        vertx.deployVerticle(new SingleSender(options, new LinkOptions(options.getTerminusFactory().getSource(address), options.getTerminusFactory().getTarget(address), Optional.empty()), connectPromise, resultPromise, containerId, message));
+
+        try {
+            connectPromise.get(2, TimeUnit.MINUTES);
+        } catch (Exception e) {
+            resultPromise.completeExceptionally(e);
+        }
+
+        return resultPromise
+                .whenComplete((res, err) -> vertx.close());
+
+    }
 }
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/amqp/Sender.java b/systemtests/src/main/java/io/enmasse/systemtest/amqp/Sender.java
index b6f1d2eb8..767b90341 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/amqp/Sender.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/amqp/Sender.java
@@ -4,24 +4,23 @@
  */
 package io.enmasse.systemtest.amqp;
 
-import io.enmasse.systemtest.CustomLogger;
+import org.apache.qpid.proton.amqp.messaging.Accepted;
+import org.apache.qpid.proton.message.Message;
+
 import io.vertx.proton.ProtonConnection;
 import io.vertx.proton.ProtonQoS;
 import io.vertx.proton.ProtonSender;
-import org.apache.qpid.proton.amqp.messaging.Accepted;
-import org.apache.qpid.proton.message.Message;
-import org.slf4j.Logger;
 
 import java.util.Iterator;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Predicate;
 
-class Sender extends ClientHandlerBase<Integer> {
-    private static Logger log = CustomLogger.getLogger();
+class Sender extends AbstractSender<Integer> {
+
     private final AtomicInteger numSent = new AtomicInteger(0);
-    private final Iterator<Message> messageQueue;
-    private final Predicate<Message> predicate;
+    protected final Iterator<Message> messageQueue;
+    protected final Predicate<Message> predicate;
 
     public Sender(AmqpConnectOptions clientOptions,
                   LinkOptions linkOptions,
@@ -36,35 +35,13 @@ class Sender extends ClientHandlerBase<Integer> {
     }
 
     @Override
-    public void connectionOpened(ProtonConnection connection) {
-        ProtonSender sender = connection.createSender(linkOptions.getTarget().getAddress());
-        sender.setTarget(linkOptions.getTarget());
-        sender.setQoS(clientOptions.getQos());
-        sender.openHandler(result -> {
-            if (result.succeeded()) {
-                log.info("Sender link '" + sender.getTarget().getAddress() + "' opened, sending messages");
-                connectPromise.complete(null);
-                sendNext(connection, sender);
-            } else {
-                handleError(connection, sender.getRemoteCondition());
-            }
-        });
-        sender.closeHandler(result -> handleError(connection, sender.getRemoteCondition()));
-        sender.open();
-    }
-
-    @Override
-    protected void connectionClosed(ProtonConnection conn) {
-        conn.close();
-        resultPromise.completeExceptionally(new RuntimeException("Connection closed after " + numSent.get() + " messages sent"));
-        connectPromise.completeExceptionally(new RuntimeException("Connection closed after " + numSent.get() + " messages sent"));
+    protected String getCurrentState() {
+        return "after " + numSent.get() + " messages sent";
     }
 
     @Override
-    protected void connectionDisconnected(ProtonConnection conn) {
-        conn.close();
-        resultPromise.completeExceptionally(new RuntimeException("Connection disconnected after " + numSent.get() + " messages sent"));
-        connectPromise.completeExceptionally(new RuntimeException("Connection disconnected after " + numSent.get() + " messages sent"));
+    protected void sendMessages(ProtonConnection connection, ProtonSender sender) {
+        sendNext(connection, sender);
     }
 
     private void sendNext(ProtonConnection connection, ProtonSender sender) {
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/amqp/SingleSender.java b/systemtests/src/main/java/io/enmasse/systemtest/amqp/SingleSender.java
new file mode 100644
index 000000000..cee709ce0
--- /dev/null
+++ b/systemtests/src/main/java/io/enmasse/systemtest/amqp/SingleSender.java
@@ -0,0 +1,80 @@
+/*
+ * 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.amqp;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.qpid.proton.message.Message;
+
+import io.vertx.proton.ProtonConnection;
+import io.vertx.proton.ProtonDelivery;
+import io.vertx.proton.ProtonSender;
+
+public class SingleSender extends AbstractSender<List<ProtonDelivery>> {
+
+    /**
+     * Recorded deliveries.
+     */
+    private List<ProtonDelivery> deliveries;
+
+    /**
+     * The message to send.
+     */
+    private final Message message;
+
+    /**
+     * The current state, in a human readable format.
+     */
+    private String state = "initialized";
+
+    public SingleSender(final AmqpConnectOptions clientOptions,
+            final LinkOptions linkOptions,
+            final CompletableFuture<Void> connectPromise,
+            final CompletableFuture<List<ProtonDelivery>> resultPromise,
+            final String containerId, final Message message) {
+
+        super(clientOptions, linkOptions, connectPromise, resultPromise, containerId);
+
+        this.deliveries = new LinkedList<>();
+        this.message = message;
+
+    }
+
+    @Override
+    protected String getCurrentState() {
+        return this.state;
+    }
+
+    @Override
+    protected void sendMessages(final ProtonConnection connection, final ProtonSender sender) {
+        this.state = "ready to send";
+
+        switch (sender.getQoS()) {
+            case AT_MOST_ONCE:
+                sender.send(this.message);
+                this.state = "message sent";
+                this.resultPromise.complete(Collections.emptyList());
+                break;
+            case AT_LEAST_ONCE:
+                sender.send(this.message, del -> {
+                    this.state = "message disposition received";
+                    this.deliveries.add(del);
+                    if (del.remotelySettled()) {
+                        this.state = "message settled";
+                        // complete, sending a copy
+                        this.resultPromise.complete(new ArrayList<>(this.deliveries));
+                    }
+                });
+                this.state = "message sent";
+                break;
+        }
+    }
+
+}
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/apiclients/ApiClient.java b/systemtests/src/main/java/io/enmasse/systemtest/apiclients/ApiClient.java
index 1e0dcbd9f..20126bdf9 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/apiclients/ApiClient.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/apiclients/ApiClient.java
@@ -85,6 +85,13 @@ public abstract class ApiClient {
         responseHandler(ar, promise, (responseCode)-> responseCode == expectedCode, String.valueOf(expectedCode), warnMessage, throwException);
     }
 
+    protected <T> void responseHandler(AsyncResult<HttpResponse<T>> ar, CompletableFuture<T> promise, Predicate<Integer> expectedCodePredicate,
+            String warnMessage, boolean throwException) {
+
+        responseHandler(ar, promise, expectedCodePredicate, expectedCodePredicate.toString(), warnMessage, throwException);
+
+    }
+
     protected <T> void responseHandler(AsyncResult<HttpResponse<T>> ar, CompletableFuture<T> promise, Predicate<Integer> expectedCodePredicate,
                                        String expectedCodeOrCodes, String warnMessage, boolean throwException) {
         try {
@@ -92,7 +99,7 @@ public abstract class ApiClient {
                 HttpResponse<T> response = ar.result();
                 T body = response.body();
                 if (expectedCodePredicate.negate().test(response.statusCode())) {
-                    log.error("expected-code: {}, response-code: {}, body: {}", expectedCodeOrCodes, response.statusCode(), response.body());
+                    log.error("expected-code: {}, response-code: {}, body: {}, op: {}", expectedCodeOrCodes, response.statusCode(), response.body(), warnMessage);
                     promise.completeExceptionally(new RuntimeException("Status " + response.statusCode() + " body: " + (body != null ? body.toString() : null)));
                 } else if (response.statusCode() < HTTP_OK || response.statusCode() >= HttpURLConnection.HTTP_MULT_CHOICE) {
                     if(throwException) {
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/apiclients/MsgCliApiClient.java b/systemtests/src/main/java/io/enmasse/systemtest/apiclients/MsgCliApiClient.java
index cd257b267..1f7cbf864 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/apiclients/MsgCliApiClient.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/apiclients/MsgCliApiClient.java
@@ -126,7 +126,7 @@ public class MsgCliApiClient extends ApiClient {
         JsonArray ids = response.getJsonArray("clients");
         String uuid = ids.getString(0);
 
-        Thread.sleep(5000);
+        Thread.sleep(10000);
 
         response = getClientInfo(uuid);
         log.info(response.toString());
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/apiclients/Predicates.java b/systemtests/src/main/java/io/enmasse/systemtest/apiclients/Predicates.java
new file mode 100644
index 000000000..d25b5b993
--- /dev/null
+++ b/systemtests/src/main/java/io/enmasse/systemtest/apiclients/Predicates.java
@@ -0,0 +1,59 @@
+/*
+ * 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.apiclients;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+public final class Predicates {
+    private Predicates() {}
+
+    public static <T> Predicate<T> is(T value) {
+        return new Predicate<>() {
+
+            @Override
+            public boolean test(T t) {
+                return value.equals(t);
+            }
+
+            @Override
+            public String toString() {
+                return value.toString();
+            }
+        };
+    }
+
+    public static <T> Predicate<T> any() {
+        return v -> true;
+    }
+
+    public static <T> Predicate<T> in(final T value) {
+        return is(value);
+    }
+
+    @SafeVarargs
+    public static <T> Predicate<T> in(final T...values) {
+        return in(Arrays.asList(values));
+    }
+
+    public static <T> Predicate<T> in(final Collection<T> values) {
+
+        return new Predicate<>() {
+
+            @Override
+            public boolean test(T t) {
+                return values.contains(t);
+            }
+
+            @Override
+            public String toString() {
+                return values.stream().map(Object::toString).collect(Collectors.joining(", "));
+            }
+        };
+    }
+}
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/iot/HttpAdapterClient.java b/systemtests/src/main/java/io/enmasse/systemtest/iot/HttpAdapterClient.java
index 883df798a..695e4293d 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/iot/HttpAdapterClient.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/iot/HttpAdapterClient.java
@@ -2,12 +2,19 @@
  * 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.iot;
 
+import static io.enmasse.systemtest.iot.MessageType.EVENT;
+import static io.enmasse.systemtest.iot.MessageType.TELEMETRY;
+import static java.time.Duration.ofSeconds;
+
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.Base64;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 import java.util.function.Predicate;
 
 import javax.ws.rs.core.HttpHeaders;
@@ -21,6 +28,8 @@ import io.enmasse.systemtest.Kubernetes;
 import io.enmasse.systemtest.apiclients.ApiClient;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.json.JsonObject;
+import io.vertx.ext.web.client.HttpRequest;
+import io.vertx.ext.web.client.HttpResponse;
 import io.vertx.ext.web.client.WebClient;
 import io.vertx.ext.web.client.WebClientOptions;
 
@@ -28,9 +37,6 @@ public class HttpAdapterClient extends ApiClient {
 
     protected static Logger log = CustomLogger.getLogger();
 
-    private static final String TELEMETRY_PATH = "/telemetry";
-    private static final String EVENT_PATH = "/event";
-
     public HttpAdapterClient(Kubernetes kubernetes, Endpoint endpoint, String username, String password) {
         super(kubernetes, () -> endpoint, "");
         this.authzString = getBasicAuth(username, password);
@@ -49,28 +55,70 @@ public class HttpAdapterClient extends ApiClient {
                 .setVerifyHost(false));
     }
 
-    public void sendTelemetry(JsonObject payload, Predicate<Integer> expectedCodePredicate, String expectedCodeOrCodes) throws Exception {
-        CompletableFuture<Buffer> responsePromise = new CompletableFuture<>();
-        log.info("POST-telemetry: body {}", payload.toString());
-        client.post(endpoint.getPort(), endpoint.getHost(), TELEMETRY_PATH)
-            .putHeader(HttpHeaders.AUTHORIZATION, authzString)
-            .putHeader(HttpHeaders.CONTENT_TYPE, ContentType.APPLICATION_JSON.getMimeType())
-            .timeout(120000)
-            .sendJsonObject(payload,
-                    ar -> responseHandler(ar, responsePromise, expectedCodePredicate, expectedCodeOrCodes, "Error sending telemetry data", false));
-        responsePromise.get(150000, TimeUnit.SECONDS);
+    public void close () {
+        this.client.close();
+    }
+
+    private static String contentType (final JsonObject payload) {
+        return payload != null ? ContentType.APPLICATION_JSON.getMimeType() : "application/vnd.eclipse-hono-empty-notification";
+    }
+
+    public HttpResponse<?> send(MessageType messageType, JsonObject payload, Predicate<Integer> expectedCodePredicate, Consumer<HttpRequest<?>> requestCustomizer,
+            final Duration responseTimeout) throws Exception {
+
+        CompletableFuture<HttpResponse<?>> responsePromise = new CompletableFuture<>();
+        var ms = responseTimeout.toMillis();
+
+        log.info("POST-{}: body {}", messageType.name().toLowerCase(), payload);
+
+        // create new request
+
+        var request = client.post(endpoint.getPort(), endpoint.getHost(), messageType.path())
+                .putHeader(HttpHeaders.AUTHORIZATION, authzString)
+                .putHeader(HttpHeaders.CONTENT_TYPE, contentType(payload))
+                .timeout(ms);
+
+        // allow to customize request
+
+        if (requestCustomizer != null) {
+            requestCustomizer.accept(request);
+        }
+
+        // execute request with payload
+
+        request.sendJsonObject(payload, ar -> {
+
+            // if the request failed ...
+            if (!ar.succeeded()) {
+                // ... fail the response promise
+                responsePromise.completeExceptionally(ar.cause());
+            }
+
+            final CompletableFuture<Buffer> nf = new CompletableFuture<>();
+            responseHandler(ar, nf, expectedCodePredicate, "Error sending " + messageType.name().toLowerCase() + " data", false);
+
+            // use the result from the responseHandler
+            // and map it to the responsePromise
+            nf.whenComplete((res, err) -> {
+                if ( err != null )  {
+                    responsePromise.completeExceptionally(err);
+                } else {
+                    responsePromise.complete(ar.result());
+                }
+            });
+        });
+
+        // the next line gives the timeout a bit extra, as the HTTP timeout should
+        // kick in, we would prefer the timeout via the future.
+        return responsePromise.get(((long)(ms*1.1)), TimeUnit.MILLISECONDS);
+    }
+
+    public HttpResponse<?> sendTelemetry(JsonObject payload, Predicate<Integer> expectedCodePredicate) throws Exception {
+        return send(TELEMETRY, payload, expectedCodePredicate, null, ofSeconds(15));
     }
 
-    public void sendEvent(JsonObject payload, Predicate<Integer> expectedCodePredicate, String expectedCodeOrCodes) throws Exception {
-        CompletableFuture<Buffer> responsePromise = new CompletableFuture<>();
-        log.info("POST-event: body {}", payload.toString());
-        client.post(endpoint.getPort(), endpoint.getHost(), EVENT_PATH)
-            .putHeader(HttpHeaders.AUTHORIZATION, authzString)
-            .putHeader(HttpHeaders.CONTENT_TYPE, ContentType.APPLICATION_JSON.getMimeType())
-            .timeout(120000)
-            .sendJsonObject(payload,
-                    ar -> responseHandler(ar, responsePromise, expectedCodePredicate, expectedCodeOrCodes, "Error sending event data", false));
-        responsePromise.get(150000, TimeUnit.SECONDS);
+    public HttpResponse<?> sendEvent(JsonObject payload, Predicate<Integer> expectedCodePredicate) throws Exception {
+        return send(EVENT, payload, expectedCodePredicate, null, ofSeconds(15));
     }
 
     private String getBasicAuth(final String user, final String password) {
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/iot/MessageType.java b/systemtests/src/main/java/io/enmasse/systemtest/iot/MessageType.java
new file mode 100644
index 000000000..e97d21a37
--- /dev/null
+++ b/systemtests/src/main/java/io/enmasse/systemtest/iot/MessageType.java
@@ -0,0 +1,24 @@
+/*
+ * 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.iot;
+
+public enum MessageType {
+
+    TELEMETRY("/telemetry"),
+    EVENT("/event"),
+    COMMAND("/control");
+
+    private String path;
+
+    private MessageType(final String path) {
+        this.path = path;
+    }
+
+    public String path() {
+        return this.path;
+    }
+
+}
\ 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 19a69c583..7837e1bc6 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/utils/AddressSpaceUtils.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/utils/AddressSpaceUtils.java
@@ -152,10 +152,12 @@ public class AddressSpaceUtils {
     }
 
     public static AddressSpace waitForAddressSpaceReady(AddressApiClient apiClient, AddressSpace addressSpace) throws Exception {
-        TimeoutBudget budget;
+        return waitForAddressSpaceReady(apiClient, addressSpace, new TimeoutBudget(15, TimeUnit.MINUTES));
+    }
 
+    public static AddressSpace waitForAddressSpaceReady(AddressApiClient apiClient, AddressSpace addressSpace, TimeoutBudget budget) throws Exception {
         boolean isReady = false;
-        budget = new TimeoutBudget(10, TimeUnit.MINUTES);
+        budget = new TimeoutBudget(15, TimeUnit.MINUTES);
         while (budget.timeLeft() >= 0 && !isReady) {
             addressSpace = jsonToAdressSpace(apiClient.getAddressSpace(addressSpace.getMetadata().getName()));
             isReady = isAddressSpaceReady(addressSpace);
@@ -174,7 +176,7 @@ public class AddressSpaceUtils {
 
     public static void waitForAddressSpacePlanApplied(AddressApiClient apiClient, AddressSpace addressSpace) throws Exception {
         AddressSpace addressSpaceObject = null;
-        TimeoutBudget budget = new TimeoutBudget(10, TimeUnit.MINUTES);
+        TimeoutBudget budget = new TimeoutBudget(15, TimeUnit.MINUTES);
 
         boolean isPlanApplied = false;
         while (budget.timeLeft() >= 0 && !isPlanApplied) {
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/utils/CertificateUtils.java b/systemtests/src/main/java/io/enmasse/systemtest/utils/CertificateUtils.java
index 583dae0ec..37cfcd19e 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/utils/CertificateUtils.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/utils/CertificateUtils.java
@@ -5,9 +5,8 @@
 package io.enmasse.systemtest.utils;
 
 import java.io.File;
-import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
-import java.util.Base64;
 import java.util.UUID;
 
 import org.apache.commons.io.FileUtils;
@@ -18,8 +17,6 @@ import io.enmasse.systemtest.executor.Executor;
 
 public class CertificateUtils {
 
-    //TODO refactor CertProviderTest to use this class
-
     private static void createSelfSignedCert(File cert, File key) throws Exception {
         new Executor().execute(Arrays.asList("openssl", "req", "-new", "-days", "11000", "-x509", "-batch", "-nodes",
                 "-out", cert.getAbsolutePath(), "-keyout", key.getAbsolutePath()));
@@ -32,30 +29,35 @@ public class CertificateUtils {
     }
 
     public static File signCsr(File caKey, File caCert, File csrKey, File csrCsr) throws Exception {
-        File crtFile = createTempFile(FilenameUtils.removeExtension(csrKey.getName()), "crt");
+        File crtFile = File.createTempFile(FilenameUtils.removeExtension(csrKey.getName()), "crt");
         new Executor().execute(Arrays.asList("openssl", "x509", "-req", "-days", "11000", "-in",
                 csrCsr.getAbsolutePath(), "-CA", caCert.getAbsolutePath(), "-CAkey", caKey.getAbsolutePath(),
                 "-CAcreateserial", "-out", crtFile.getAbsolutePath()));
         return crtFile;
     }
 
-    private static File createTempFile(String prefix, String suffix) throws IOException {
-        File file = File.createTempFile(prefix, suffix);
-        return file;
-    }
-
     public static CertBundle createCertBundle() throws Exception {
-        File caCert = createTempFile("certAuthority", "crt");
-        File caKey = createTempFile("certAuthority", "key");
+        File caCert = File.createTempFile("certAuthority", "crt");
+        File caKey = File.createTempFile("certAuthority", "key");
         createSelfSignedCert(caCert, caKey);
         String randomName = UUID.randomUUID().toString();
-        File keyFile = createTempFile(randomName, "key");
-        File csrFile = createTempFile(randomName, "csr");
+        File keyFile = File.createTempFile(randomName, "key");
+        File csrFile = File.createTempFile(randomName, "csr");
         createCsr(keyFile, csrFile);
         File crtFile = signCsr(caKey, caCert, keyFile, csrFile);
-        String key = Base64.getEncoder().encodeToString(FileUtils.readFileToByteArray(keyFile));
-        String cert = Base64.getEncoder().encodeToString(FileUtils.readFileToByteArray(crtFile));
-        return new CertBundle(caCert, crtFile, key, cert);
+        try {
+            return new CertBundle(FileUtils.readFileToString(caCert, StandardCharsets.UTF_8),
+                    FileUtils.readFileToString(keyFile, StandardCharsets.UTF_8),
+                    FileUtils.readFileToString(crtFile, StandardCharsets.UTF_8));
+        } finally {
+            deleteFiles(caCert, caKey, keyFile, csrFile, crtFile);
+        }
+
     }
 
+    private static void deleteFiles(File... files) {
+        for(File file : files) {
+            FileUtils.deleteQuietly(file);
+        }
+    }
 }
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/utils/IoTUtils.java b/systemtests/src/main/java/io/enmasse/systemtest/utils/IoTUtils.java
index 3bcf9a512..2dc174ecf 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/utils/IoTUtils.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/utils/IoTUtils.java
@@ -16,6 +16,8 @@ import io.enmasse.systemtest.timemeasuring.SystemtestsOperation;
 import io.enmasse.systemtest.timemeasuring.TimeMeasuringSystem;
 import org.slf4j.Logger;
 
+import static io.enmasse.systemtest.utils.AddressSpaceUtils.jsonToAdressSpace;
+
 import java.util.concurrent.TimeUnit;
 
 public class IoTUtils {
@@ -39,7 +41,7 @@ public class IoTUtils {
         }
     }
 
-    public static void waitForIoTProjectReady(IoTProjectApiClient apiClient, IoTProject project) throws Exception {
+    public static void waitForIoTProjectReady(IoTProjectApiClient apiClient, AddressApiClient addressSpaceApiClient, IoTProject project) throws Exception {
         boolean isReady = false;
         TimeoutBudget budget = new TimeoutBudget(10, TimeUnit.MINUTES);
         while (budget.timeLeft() >= 0 && !isReady) {
@@ -54,6 +56,15 @@ public class IoTUtils {
             String jsonStatus = project != null && project.getStatus() != null ? project.getStatus().toString() : "Project doesn't have status";
             throw new IllegalStateException("IoTProject " + project.getMetadata().getName() + " is not in Ready state within timeout: " + jsonStatus);
         }
+
+        if ( project.getSpec().getDownstreamStrategy() != null
+                && project.getSpec().getDownstreamStrategy().getManagedStrategy() != null
+                && project.getSpec().getDownstreamStrategy().getManagedStrategy().getAddressSpace() != null
+                && project.getSpec().getDownstreamStrategy().getManagedStrategy().getAddressSpace().getName() != null
+                ) {
+            var addressSpaceName = project.getSpec().getDownstreamStrategy().getManagedStrategy().getAddressSpace().getName();
+            AddressSpaceUtils.waitForAddressSpaceReady(addressSpaceApiClient, jsonToAdressSpace(addressSpaceApiClient.getAddressSpace(addressSpaceName)), budget);
+        }
     }
 
     private static void waitForIoTProjectDeleted(Kubernetes kubernetes, AddressApiClient addressApiClient, IoTProject project) throws Exception {
diff --git a/systemtests/src/main/java/io/enmasse/systemtest/utils/TestUtils.java b/systemtests/src/main/java/io/enmasse/systemtest/utils/TestUtils.java
index 8237fbd7b..e3fe517bf 100644
--- a/systemtests/src/main/java/io/enmasse/systemtest/utils/TestUtils.java
+++ b/systemtests/src/main/java/io/enmasse/systemtest/utils/TestUtils.java
@@ -10,6 +10,10 @@ import io.enmasse.address.model.AddressSpace;
 import io.enmasse.address.model.BrokerState;
 import io.enmasse.address.model.BrokerStatus;
 import io.enmasse.admin.model.v1.AddressPlan;
+import io.enmasse.iot.model.v1.DoneableIoTProject;
+import io.enmasse.iot.model.v1.IoTCrd;
+import io.enmasse.iot.model.v1.IoTProject;
+import io.enmasse.iot.model.v1.IoTProjectList;
 import io.enmasse.systemtest.*;
 import io.enmasse.systemtest.apiclients.AddressApiClient;
 import io.enmasse.systemtest.timemeasuring.SystemtestsOperation;
@@ -563,4 +567,20 @@ public class TestUtils {
             }
         }, budget);
     }
+
+    public static void waitForIoTProjectReady(final String namespace, final String name, final TimeoutBudget budget) throws Exception {
+        var projectClient = Kubernetes.getInstance().getClient().customResources(IoTCrd.project(), IoTProject.class, IoTProjectList.class, DoneableIoTProject.class);
+
+        final BooleanSupplier s = () -> {
+
+            var config = projectClient.inNamespace(namespace).withName(name).get();
+            if (config == null || config.getStatus() == null) {
+                return false;
+            }
+            return config.getStatus().isReady();
+
+        };
+
+        TestUtils.waitUntilCondition(String.format("IoTProject to become ready: %s/%s", namespace, name), s, budget);
+    }
 }
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/bases/IoTTestBase.java b/systemtests/src/test/java/io/enmasse/systemtest/bases/IoTTestBase.java
index 293ca8170..3c096fba2 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/bases/IoTTestBase.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/bases/IoTTestBase.java
@@ -92,6 +92,17 @@ public abstract class IoTTestBase extends TestBase {
         return null;
     }
 
+    /**
+     * Get the Hono tenant name from the project configuration.
+     */
+    protected String tenantId() {
+        var project = getSharedIoTProject();
+        if (project == null) {
+            return null;
+        }
+        return String.format("%s.%s", project.getMetadata().getNamespace(), project.getMetadata().getName());
+    }
+
     protected void createIoTConfig(IoTConfig config) throws Exception {
         String operationID = TimeMeasuringSystem.startOperation(SystemtestsOperation.CREATE_IOT_CONFIG);
         if (iotConfigApiClient.existsIoTConfig(config.getMetadata().getName())) {
@@ -119,7 +130,7 @@ public abstract class IoTTestBase extends TestBase {
                 iotProjects.add(project);
             }
         }
-        IoTUtils.waitForIoTProjectReady(iotProjectApiClient, project);
+        IoTUtils.waitForIoTProjectReady(iotProjectApiClient, addressApiClient, project);
         IoTUtils.syncIoTProject(project, iotProjectApiClient);
         TimeMeasuringSystem.stopOperation(operationID);
     }
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/bases/IoTTestBaseWithShared.java b/systemtests/src/test/java/io/enmasse/systemtest/bases/IoTTestBaseWithShared.java
index 8a76c72c0..27675fd87 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/bases/IoTTestBaseWithShared.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/bases/IoTTestBaseWithShared.java
@@ -9,6 +9,9 @@ import io.enmasse.iot.model.v1.IoTConfigBuilder;
 import io.enmasse.iot.model.v1.IoTProject;
 import io.enmasse.systemtest.CertBundle;
 import io.enmasse.systemtest.CustomLogger;
+import io.enmasse.systemtest.Environment;
+import io.enmasse.systemtest.UserCredentials;
+import io.enmasse.systemtest.amqp.AmqpClientFactory;
 import io.enmasse.systemtest.utils.CertificateUtils;
 import io.enmasse.systemtest.utils.IoTUtils;
 import org.junit.jupiter.api.AfterEach;
@@ -17,17 +20,25 @@ import org.junit.jupiter.api.extension.ExtensionContext;
 import org.slf4j.Logger;
 
 import java.nio.ByteBuffer;
-import java.util.Base64;
+import java.util.UUID;
 
 public abstract class IoTTestBaseWithShared extends IoTTestBase {
 
     protected static Logger log = CustomLogger.getLogger();
 
+    private final String addressSpace = "shared-address-space";
+
     protected IoTConfig sharedConfig;
     protected IoTProject sharedProject;
 
+    private UserCredentials credentials;
+    protected AmqpClientFactory iotAmqpClientFactory;
+
     @BeforeEach
     public void setUpSharedIoTProject() throws Exception {
+
+        this.credentials = new UserCredentials(UUID.randomUUID().toString(), UUID.randomUUID().toString());
+
         if (sharedConfig == null) {
             CertBundle certBundle = CertificateUtils.createCertBundle();
             sharedConfig = new IoTConfigBuilder()
@@ -39,8 +50,8 @@ public abstract class IoTTestBaseWithShared extends IoTTestBase {
                     .withNewMqtt()
                     .withNewEndpoint()
                     .withNewKeyCertificateStrategy()
-                    .withCertificate(ByteBuffer.wrap(Base64.getDecoder().decode(certBundle.getCert())))
-                    .withKey(ByteBuffer.wrap(Base64.getDecoder().decode(certBundle.getKey())))
+                    .withCertificate(ByteBuffer.wrap(certBundle.getCert().getBytes()))
+                    .withKey(ByteBuffer.wrap(certBundle.getKey().getBytes()))
                     .endKeyCertificateStrategy()
                     .endEndpoint()
                     .endMqtt()
@@ -52,9 +63,25 @@ public abstract class IoTTestBaseWithShared extends IoTTestBase {
         }
 
         if (sharedProject == null) {
-            sharedProject = IoTUtils.getBasicIoTProjectObject("shared-iot-project", "shard-address-space");
+            sharedProject = IoTUtils.getBasicIoTProjectObject("shared-iot-project", this.addressSpace);
             createIoTProject(sharedProject);
         }
+
+        this.iotAmqpClientFactory = createAmqpClientFactory();
+    }
+
+    public String getAddressSpace() {
+        return this.addressSpace;
+    }
+
+    /**
+     * Create a new {@link AmqpClientFactory} for the IoTProject's address space.
+     */
+    private AmqpClientFactory createAmqpClientFactory() throws Exception {
+
+        getUserApiClient().createUser(this.addressSpace, this.credentials);
+        return new AmqpClientFactory(kubernetes, Environment.getInstance(), getAddressSpace(this.addressSpace), this.credentials);
+
     }
 
     @AfterEach
@@ -79,6 +106,13 @@ public abstract class IoTTestBaseWithShared extends IoTTestBase {
                 log.warn("Remove shared iotproject when test failed - SKIPPED!");
             }
         }
+
+        // always clean up random user
+        getUserApiClient().deleteUser(this.addressSpace, this.credentials.getUsername());
+        if (this.iotAmqpClientFactory != null) {
+            this.iotAmqpClientFactory.close();
+            this.iotAmqpClientFactory = null;
+        }
     }
 
     @Override
@@ -91,5 +125,4 @@ public abstract class IoTTestBaseWithShared extends IoTTestBase {
         return sharedProject;
     }
 
-
 }
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/bases/TestBase.java b/systemtests/src/test/java/io/enmasse/systemtest/bases/TestBase.java
index d745515f5..00f79c542 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/bases/TestBase.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/bases/TestBase.java
@@ -280,7 +280,7 @@ public abstract class TestBase implements ITestBase, ITestSeparator {
     }
 
     protected void appendAddresses(AddressSpace addressSpace, Address... destinations) throws Exception {
-        TimeoutBudget budget = new TimeoutBudget(10, TimeUnit.MINUTES);
+        TimeoutBudget budget = new TimeoutBudget(15, TimeUnit.MINUTES);
         appendAddresses(addressSpace, budget, destinations);
     }
 
@@ -289,7 +289,7 @@ public abstract class TestBase implements ITestBase, ITestSeparator {
     }
 
     protected void appendAddresses(AddressSpace addressSpace, boolean wait, Address... destinations) throws Exception {
-        TimeoutBudget budget = new TimeoutBudget(10, TimeUnit.MINUTES);
+        TimeoutBudget budget = new TimeoutBudget(15, TimeUnit.MINUTES);
         appendAddresses(addressSpace, wait, budget, destinations);
     }
 
@@ -299,13 +299,13 @@ public abstract class TestBase implements ITestBase, ITestSeparator {
     }
 
     protected void appendAddresses(AddressSpace addressSpace, boolean wait, int batchSize, Address... destinations) throws Exception {
-        TimeoutBudget timeout = new TimeoutBudget(10, TimeUnit.MINUTES);
+        TimeoutBudget timeout = new TimeoutBudget(15, TimeUnit.MINUTES);
         AddressUtils.appendAddresses(addressApiClient, kubernetes, timeout, addressSpace, wait, batchSize, destinations);
         logCollector.collectConfigMaps();
     }
 
     protected void setAddresses(AddressSpace addressSpace, int expectedCode, Address... addresses) throws Exception {
-        TimeoutBudget budget = new TimeoutBudget(10, TimeUnit.MINUTES);
+        TimeoutBudget budget = new TimeoutBudget(15, TimeUnit.MINUTES);
         logCollector.collectRouterState("setAddresses");
 
         if (expectedCode == HTTP_CONFLICT) {
@@ -335,7 +335,7 @@ public abstract class TestBase implements ITestBase, ITestSeparator {
     }
 
     protected void setAddresses(AddressSpace addressSpace, AddressApiClient apiClient, Address... addresses) throws Exception {
-        AddressUtils.setAddresses(apiClient, kubernetes, new TimeoutBudget(10, TimeUnit.MINUTES), addressSpace, true, HTTP_CREATED, addresses);
+        AddressUtils.setAddresses(apiClient, kubernetes, new TimeoutBudget(15, TimeUnit.MINUTES), addressSpace, true, HTTP_CREATED, addresses);
         logCollector.collectConfigMaps();
     }
 
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/bases/clients/ClusterClientTestBase.java b/systemtests/src/test/java/io/enmasse/systemtest/bases/clients/ClusterClientTestBase.java
index c182c9e2a..6c830bf4b 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/bases/clients/ClusterClientTestBase.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/bases/clients/ClusterClientTestBase.java
@@ -91,6 +91,8 @@ public abstract class ClusterClientTestBase extends TestBaseWithShared {
         assertThat(String.format("Return code of sender is not 0: %s", response.toString()),
                 response.getInteger("ecode"), is(0));
 
+        Thread.sleep(2000);
+
         response = cliApiClient.sendAndGetStatus(receiver);
         assertThat(String.format("Return code of receiver is not 0: %s", response.toString()),
                 response.getInteger("ecode"), is(0));
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/common/certs/CertProviderTest.java b/systemtests/src/test/java/io/enmasse/systemtest/common/certs/CertProviderTest.java
index eee0d5498..5f3f09271 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/common/certs/CertProviderTest.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/common/certs/CertProviderTest.java
@@ -8,21 +8,15 @@ import static io.enmasse.systemtest.TestTag.isolated;
 import static org.junit.jupiter.api.Assertions.fail;
 
 import java.io.ByteArrayInputStream;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
 import java.io.InputStream;
 import java.security.KeyStore;
 import java.security.SecureRandom;
 import java.security.cert.CertificateFactory;
 import java.security.cert.X509Certificate;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Base64;
 import java.util.Collections;
-import java.util.List;
 import java.util.Optional;
-import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
@@ -31,12 +25,8 @@ import javax.net.ssl.SSLSocketFactory;
 import javax.net.ssl.TrustManagerFactory;
 
 import io.enmasse.address.model.*;
-import org.apache.commons.io.FileUtils;
-import org.apache.commons.io.FilenameUtils;
 import org.eclipse.paho.client.mqttv3.IMqttClient;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
@@ -58,10 +48,10 @@ import io.enmasse.systemtest.amqp.AmqpClient;
 import io.enmasse.systemtest.apiclients.OpenshiftCertValidatorApiClient;
 import io.enmasse.systemtest.bases.TestBase;
 import io.enmasse.systemtest.common.api.ApiServerTest;
-import io.enmasse.systemtest.executor.Executor;
 import io.enmasse.systemtest.standard.QueueTest;
 import io.enmasse.systemtest.utils.AddressSpaceUtils;
 import io.enmasse.systemtest.utils.AddressUtils;
+import io.enmasse.systemtest.utils.CertificateUtils;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.net.PemTrustOptions;
@@ -74,23 +64,11 @@ class CertProviderTest extends TestBase {
     private static Logger log = CustomLogger.getLogger();
     private static String ENDPOINT_PREFIX = "test-endpoint-";
 
-    private List<File> createdFiles;
-
     private AddressSpace addressSpace;
     private UserCredentials user;
     private Address queue;
     private Address topic;
 
-    @BeforeEach
-    void setUp() {
-        createdFiles = new ArrayList<>();
-    }
-
-    @AfterEach
-    void tearDown() {
-        createdFiles.forEach(FileUtils::deleteQuietly);
-    }
-
     @Test
     void testSelfSigned() throws Exception {
 
@@ -163,12 +141,12 @@ class CertProviderTest extends TestBase {
 
     @Test
     void testCertBundle() throws Exception {
-        CertBundle certBundle = createCertBundle();
+        CertBundle certBundle = CertificateUtils.createCertBundle();
 
         createTestEnv(new CertSpecBuilder()
                 .withProvider(CertProvider.certBundle.name())
-                .withTlsKey(certBundle.getKey())
-                .withTlsCert(certBundle.getCert())
+                .withTlsKey(certBundle.getKeyB64())
+                .withTlsCert(certBundle.getCertB64())
                 .build());
 
         AmqpClient amqpClient = amqpClientFactory.createQueueClient(addressSpace);
@@ -178,11 +156,11 @@ class CertProviderTest extends TestBase {
                 .setSsl(true)
                 .setHostnameVerificationAlgorithm("")
                 .setPemTrustOptions(new PemTrustOptions()
-                        .addCertPath(certBundle.getCaCert().getAbsolutePath()))
+                        .addCertValue(Buffer.buffer(certBundle.getCaCert())))
                 .setTrustAll(false);
 
         MqttConnectOptions mqttOptions = new MqttConnectOptions();
-        mqttOptions.setSocketFactory(getSocketFactory(new FileInputStream(certBundle.getCaCert())));
+        mqttOptions.setSocketFactory(getSocketFactory(new ByteArrayInputStream(certBundle.getCaCert().getBytes())));
         mqttOptions.setUserName(user.getUsername());
         mqttOptions.setPassword(user.getPassword().toCharArray());
         IMqttClient mqttClient = mqttClientFactory.build()
@@ -196,19 +174,19 @@ class CertProviderTest extends TestBase {
     @Test
     @Disabled("Disabled due to #2427")
     void testConsoleCertBundleTrustEndpointCert() throws Exception {
-        CertBundle certBundle = createCertBundle();
+        CertBundle certBundle = CertificateUtils.createCertBundle();
 
         createTestEnv(new CertSpecBuilder()
                 .withProvider(CertProvider.certBundle.name())
-                .withTlsKey(certBundle.getKey())
-                .withTlsCert(certBundle.getCert())
+                .withTlsKey(certBundle.getKeyB64())
+                .withTlsCert(certBundle.getCertB64())
                 .build(), false);
 
         WebClient webClient = WebClient.create(VertxFactory.create(), new WebClientOptions()
                 .setSsl(true)
                 .setTrustAll(false)
                 .setPemTrustOptions(new PemTrustOptions()
-                        .addCertPath(certBundle.getCrtFile().getAbsolutePath()))
+                        .addCertValue(Buffer.buffer(certBundle.getCert())))
                 .setVerifyHost(false));
 
         testConsole(webClient);
@@ -217,19 +195,19 @@ class CertProviderTest extends TestBase {
     @Test
     @Disabled("Disabled due to #2427")
     void testConsoleCertBundleTrustCaCert() throws Exception {
-        CertBundle certBundle = createCertBundle();
+        CertBundle certBundle = CertificateUtils.createCertBundle();
 
         createTestEnv(new CertSpecBuilder()
                 .withProvider(CertProvider.certBundle.name())
-                .withTlsKey(certBundle.getKey())
-                .withTlsCert(certBundle.getCert())
+                .withTlsKey(certBundle.getKeyB64())
+                .withTlsCert(certBundle.getCertB64())
                 .build(), false);
 
         WebClient webClient = WebClient.create(VertxFactory.create(), new WebClientOptions()
                 .setSsl(true)
                 .setTrustAll(false)
                 .setPemTrustOptions(new PemTrustOptions()
-                        .addCertPath(certBundle.getCaCert().getAbsolutePath()))
+                        .addCertValue(Buffer.buffer(certBundle.getCaCert())))
                 .setVerifyHost(false));
 
         testConsole(webClient);
@@ -357,31 +335,6 @@ class CertProviderTest extends TestBase {
         }
     }
 
-    private void createSelfSignedCert(File cert, File key) throws Exception {
-        new Executor().execute(Arrays.asList("openssl", "req", "-new", "-days", "11000", "-x509", "-batch", "-nodes",
-                "-out", cert.getAbsolutePath(), "-keyout", key.getAbsolutePath()));
-    }
-
-    public void createCsr(File keyFile, File csrFile) throws Exception {
-        String subjString = "/O=enmasse-systemtests";
-        new Executor().execute(Arrays.asList("openssl", "req", "-new", "-batch", "-nodes", "-keyout",
-                keyFile.getAbsolutePath(), "-subj", subjString, "-out", csrFile.getAbsolutePath()));
-    }
-
-    public File signCsr(File caKey, File caCert, File csrKey, File csrCsr) throws Exception {
-        File crtFile = createTempFile(FilenameUtils.removeExtension(csrKey.getName()), "crt");
-        new Executor().execute(Arrays.asList("openssl", "x509", "-req", "-days", "11000", "-in",
-                csrCsr.getAbsolutePath(), "-CA", caCert.getAbsolutePath(), "-CAkey", caKey.getAbsolutePath(),
-                "-CAcreateserial", "-out", crtFile.getAbsolutePath()));
-        return crtFile;
-    }
-
-    private File createTempFile(String prefix, String suffix) throws IOException {
-        File file = File.createTempFile(prefix, suffix);
-        createdFiles.add(file);
-        return file;
-    }
-
     private SSLSocketFactory getSocketFactory(InputStream caCrtFile) throws Exception {
         CertificateFactory cf = CertificateFactory.getInstance("X.509");
 
@@ -399,18 +352,5 @@ class CertProviderTest extends TestBase {
         return context.getSocketFactory();
     }
 
-    private CertBundle createCertBundle() throws Exception {
-        File caCert = createTempFile("certAuthority", "crt");
-        File caKey = createTempFile("certAuthority", "key");
-        createSelfSignedCert(caCert, caKey);
-        String randomName = UUID.randomUUID().toString();
-        File keyFile = createTempFile(randomName, "key");
-        File csrFile = createTempFile(randomName, "csr");
-        createCsr(keyFile, csrFile);
-        File crtFile = signCsr(caKey, caCert, keyFile, csrFile);
-        String key = Base64.getEncoder().encodeToString(FileUtils.readFileToByteArray(keyFile));
-        String cert = Base64.getEncoder().encodeToString(FileUtils.readFileToByteArray(crtFile));
-        return new CertBundle(caCert, crtFile, key, cert);
-    }
 
 }
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/iot/control/CommandAndControlTest.java b/systemtests/src/test/java/io/enmasse/systemtest/iot/control/CommandAndControlTest.java
new file mode 100644
index 000000000..ebc352b85
--- /dev/null
+++ b/systemtests/src/test/java/io/enmasse/systemtest/iot/control/CommandAndControlTest.java
@@ -0,0 +1,207 @@
+/*
+ * 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.iot.control;
+
+import io.enmasse.systemtest.Endpoint;
+import io.enmasse.systemtest.amqp.QueueTerminusFactory;
+import io.enmasse.systemtest.bases.IoTTestBaseWithShared;
+import io.enmasse.systemtest.iot.CredentialsRegistryClient;
+import io.enmasse.systemtest.iot.DeviceRegistryClient;
+import io.enmasse.systemtest.iot.HttpAdapterClient;
+import io.enmasse.systemtest.utils.TestUtils;
+import io.vertx.proton.ProtonDelivery;
+
+import org.apache.qpid.proton.amqp.Binary;
+import org.apache.qpid.proton.amqp.messaging.Accepted;
+import org.apache.qpid.proton.amqp.messaging.Data;
+import org.apache.qpid.proton.amqp.messaging.Released;
+import org.apache.qpid.proton.message.Message;
+import org.hamcrest.core.Is;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import static java.net.HttpURLConnection.HTTP_ACCEPTED;
+import static java.net.HttpURLConnection.HTTP_OK;
+import static java.time.Duration.ofSeconds;
+import static org.hamcrest.CoreMatchers.anyOf;
+import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
+import static org.hamcrest.collection.IsIterableContainingInOrder.contains;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
+import static org.hamcrest.core.IsNull.notNullValue;
+import static org.junit.Assert.assertThat;
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+import static io.enmasse.systemtest.TestTag.sharedIot;
+import static io.enmasse.systemtest.TimeoutBudget.ofDuration;
+import static io.enmasse.systemtest.apiclients.Predicates.any;
+import static io.enmasse.systemtest.apiclients.Predicates.is;
+import static io.enmasse.systemtest.iot.MessageType.TELEMETRY;
+
+@Tag(sharedIot)
+class CommandAndControlTest extends IoTTestBaseWithShared {
+
+    private Endpoint deviceRegistryEndpoint;
+    private Endpoint httpAdapterEndpoint;
+    private DeviceRegistryClient registryClient;
+    private CredentialsRegistryClient credentialsClient;
+
+    private String deviceId;
+
+    private String authId;
+
+    private String password;
+
+    private HttpAdapterClient httpClient;
+
+    @BeforeEach
+    protected void initClient() {
+        this.deviceRegistryEndpoint = kubernetes.getExternalEndpoint("device-registry");
+        this.httpAdapterEndpoint = kubernetes.getExternalEndpoint("iot-http-adapter");
+        this.registryClient = new DeviceRegistryClient(kubernetes, this.deviceRegistryEndpoint);
+        this.credentialsClient = new CredentialsRegistryClient(kubernetes, this.deviceRegistryEndpoint);
+    }
+
+    @BeforeEach
+    protected void initDevice() {
+        this.deviceId = UUID.randomUUID().toString();
+        this.authId = UUID.randomUUID().toString();
+        this.password = UUID.randomUUID().toString();
+        this.httpClient = new HttpAdapterClient(kubernetes, this.httpAdapterEndpoint, this.authId + "@" + tenantId(), this.password);
+    }
+
+    @AfterEach
+    protected void closeHttpClient() {
+        if (this.httpClient != null) {
+            this.httpClient.close();
+            this.httpClient = null;
+        }
+    }
+
+    @Test
+    void testOneShotCommand() throws Exception {
+
+        // set up new random device
+        this.registryClient.registerDevice(tenantId(), this.deviceId);
+        this.credentialsClient.addCredentials(tenantId(), this.deviceId, this.authId, this.password);
+
+        final var commandPayload = UUID.randomUUID().toString();
+        final int ttd = 30;
+
+        final AtomicReference<Future<List<ProtonDelivery>>> sentFuture = new AtomicReference<>();
+
+        try (var t1 = this.iotAmqpClientFactory.createQueueClient()) {
+
+            // setup consumer
+            var f1 = t1.recvMessages(new QueueTerminusFactory().getSource("telemetry/" + tenantId()), msg -> {
+
+                var ttdValue = msg.getApplicationProperties().getValue().get("ttd");
+
+                if (ttdValue == null) {
+                    // this was the initial message, without waiting for commands
+                    return false;
+                }
+
+                var deviceId = msg.getApplicationProperties().getValue().get("device_id").toString();
+
+                // prepare message
+
+                var commandMessage = Message.Factory.create();
+                commandMessage.setSubject("CMD1");
+                commandMessage.setMessageId(UUID.randomUUID().toString());
+
+                commandMessage.setContentType("application/octet-stream");
+                commandMessage.setBody(new Data(Binary.create(ByteBuffer.wrap(commandPayload.getBytes()))));
+
+                // send one shot command
+
+                log.info("Sending out command message");
+                var f2 = t1.sendMessage("control/" + tenantId() + "/" + deviceId, commandMessage)
+                        .whenComplete((res, err) -> {
+                            String strres = null;
+                            if (res != null) {
+                                strres = res.stream().map(ProtonDelivery::getRemoteState).map(Object::toString).collect(Collectors.joining(", "));
+                            }
+                            log.info("Message result - res: {}, err:", // no need for final {}, as this is an exception
+                                    strres, err);
+                        });
+                sentFuture.set(f2);
+                log.info("Message underway");
+
+                // stop listening for more messages
+
+                return true;
+
+            }, Optional.empty()).getResult();
+
+            // wait for the first telemetry message to succeed
+
+            TestUtils.waitUntilCondition("First successful telemetry message", () -> {
+                try {
+                    var response = this.httpClient.sendTelemetry(null, any());
+                    return response.statusCode() == HTTP_ACCEPTED;
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+            }, ofDuration(ofSeconds(60)));
+
+            log.info("First telemetry message accepted");
+
+            // consumer link should be ready now ... send telemetry with "ttd"
+
+            var response = this.httpClient.send(TELEMETRY, null, is(HTTP_OK /* OK for command responses */), request -> {
+                // set "time to disconnect"
+                request.putHeader("hono-ttd", Integer.toString(ttd));
+            }, Duration.ofSeconds(ttd + 5));
+
+            log.info("Telemetry with TTD processed");
+
+            // wait for the future of the sent message
+
+            var messageFuture = sentFuture.get();
+            assertThat(messageFuture, notNullValue());
+
+            // assert message - cloud side
+
+            var m1 = f1.get(10, TimeUnit.SECONDS);
+            assertThat(m1, hasSize(2));
+            var msg = m1.get(1);
+
+            // assert command message deliveries - cloud side
+
+            final List<ProtonDelivery> deliveries = messageFuture.get(5, TimeUnit.SECONDS);
+            assertThat(deliveries, hasSize(1));
+            assertThat(deliveries.stream().map(ProtonDelivery::getRemoteState).collect(Collectors.toList()),
+                    contains(
+                            anyOf(
+                                    instanceOf(Released.class), // remove once issue eclipse/hono#1149 is fixed
+                                    instanceOf(Accepted.class))));
+
+            // assert message - device side
+
+            final var actualCommand = response.bodyAsString();
+            assertThat(response.getHeader("hono-command"), Is.is("CMD1"));
+            assertThat(actualCommand, Is.is(commandPayload));
+
+            // message must have "ttd" set
+
+            var ttdValue = msg.getApplicationProperties().getValue().get("ttd");
+            assertThat(ttdValue, instanceOf(Number.class));
+            assertThat(ttdValue, Is.is(30));
+
+
+        }
+    }
+
+}
diff --git a/systemtests/src/test/java/io/enmasse/systemtest/iot/registry/DeviceRegistryTest.java b/systemtests/src/test/java/io/enmasse/systemtest/iot/registry/DeviceRegistryTest.java
index a2d1e874a..8e087f68b 100644
--- a/systemtests/src/test/java/io/enmasse/systemtest/iot/registry/DeviceRegistryTest.java
+++ b/systemtests/src/test/java/io/enmasse/systemtest/iot/registry/DeviceRegistryTest.java
@@ -22,6 +22,10 @@ import java.time.Instant;
 import java.util.Map;
 
 import static io.enmasse.systemtest.TestTag.sharedIot;
+import static io.enmasse.systemtest.apiclients.Predicates.in;
+import static java.net.HttpURLConnection.HTTP_ACCEPTED;
+import static java.net.HttpURLConnection.HTTP_UNAUTHORIZED;
+import static java.net.HttpURLConnection.HTTP_UNAVAILABLE;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 
@@ -164,32 +168,14 @@ class DeviceRegistryTest extends IoTTestBaseWithShared {
         HttpAdapterClient httpAdapterClient = new HttpAdapterClient(kubernetes, httpAdapterEndpoint, authId + "@" + tenantId(), password);
         JsonObject payload = new JsonObject(Map.of("data", "dummy"));
 
-        String expectedResponse = authFail ? "401" : "503, 202";
-        log.info("Sending event data expected response. " + expectedResponse);
-        httpAdapterClient.sendEvent(payload, (responseCode) -> {
-            log.info("Received response: " + responseCode);
-            if (authFail) {
-                return responseCode == HttpURLConnection.HTTP_UNAUTHORIZED;
-            } else {
-                return responseCode == HttpURLConnection.HTTP_UNAVAILABLE || responseCode == HttpURLConnection.HTTP_ACCEPTED;
-            }
-        }, expectedResponse);
-
-        expectedResponse = authFail ? "401" : "503";
-        log.info("Sending telemetry data expected response: " + expectedResponse);
-        httpAdapterClient.sendTelemetry(payload, (responseCode) -> {
-            log.info("Received response: " + responseCode);
-            if (authFail) {
-                return responseCode == HttpURLConnection.HTTP_UNAUTHORIZED;
-            } else {
-                return responseCode == HttpURLConnection.HTTP_UNAVAILABLE;
-            }
-        }, expectedResponse);
+        var expectedResponse = authFail ? in(HTTP_UNAUTHORIZED): in(HTTP_UNAVAILABLE, HTTP_ACCEPTED);
+        log.info("Sending event data expected response: {}", expectedResponse);
+        httpAdapterClient.sendEvent(payload, expectedResponse );
 
-    }
+        expectedResponse = authFail ? in(HTTP_UNAUTHORIZED): in(HTTP_UNAVAILABLE);
+        log.info("Sending telemetry data expected response: {}", expectedResponse);
+        httpAdapterClient.sendTelemetry(payload, expectedResponse);
 
-    private String tenantId() {
-        return String.format("%s.%s", sharedProject.getMetadata().getNamespace(), sharedProject.getMetadata().getName());
     }
 
 }