From 5535dcfcfbdab8d8b8150e03dc11dbd9db0f3acc Mon Sep 17 00:00:00 2001
From: Trevor <grafnu@users.noreply.github.com>
Date: Fri, 25 Aug 2023 16:53:58 +0100
Subject: [PATCH] More UDMIS and Registrar fixes (#721)

---
 .github/workflows/testing.yml                 |   4 +-
 .../com/google/udmi/util/GeneralUtils.java    |   5 +-
 .../UdmiServicePod_configured.xml             |   2 +-
 udmis/bin/container                           |  42 +++----
 udmis/bin/deploy                              |  21 +---
 udmis/bin/update                              |   4 +-
 udmis/etc/prod_pod.json                       |   5 +-
 .../access/ClearBladeIotAccessProvider.java   | 103 +++++++++---------
 .../access/DynamicIotAccessProvider.java      |   6 +
 .../service/access/GcpIotAccessProvider.java  |   4 +-
 .../udmi/service/access/IotAccessBase.java    |  50 ++++++++-
 .../access/LocalIotAccessProvider.java        |   1 +
 .../udmi/service/core/DistributorPipe.java    |   6 +-
 .../bos/udmi/service/core/ProcessorBase.java  |   1 +
 .../udmi/service/core/ReflectProcessor.java   |  37 +++++--
 .../service/messaging/impl/MessageBase.java   |  11 +-
 .../messaging/impl/MessageDispatcherImpl.java |   2 +-
 .../service/messaging/impl/PubSubPipe.java    |  10 +-
 .../messaging/impl/SimpleMqttPipe.java        |  72 ++++++------
 .../service/access/IotAccessBaseTest.java     |  27 +++++
 .../bos/iot/core/proxy/MqttPublisher.java     |   2 +-
 .../daq/mqtt/registrar/LocalDevice.java       |   4 +-
 .../google/daq/mqtt/registrar/Registrar.java  |   3 +-
 .../daq/mqtt/sequencer/SequenceBase.java      |   4 +-
 .../daq/mqtt/util/IotReflectorClient.java     |  10 +-
 25 files changed, 262 insertions(+), 174 deletions(-)
 mode change 100755 => 120000 udmis/bin/deploy
 create mode 100644 udmis/src/test/java/com/google/bos/udmi/service/access/IotAccessBaseTest.java

diff --git a/.github/workflows/testing.yml b/.github/workflows/testing.yml
index d8240bc97a..02d9ac56c2 100644
--- a/.github/workflows/testing.yml
+++ b/.github/workflows/testing.yml
@@ -49,12 +49,13 @@ jobs:
       - name: Setup udmis container build
         if: ${{ github.event_name == 'push' }}
         run: |
+          set -x
           revhash=$(git rev-parse $GITHUB_REF)
           IMAGE_TAG=g${revhash:0:9}
           PUSH_REPO=$PUSH_REGISTRY/${{ github.repository }}
           PUSH_TAG=$PUSH_REPO:$IMAGE_TAG
           echo PUSH_TAG=$PUSH_TAG >> $GITHUB_ENV
-          udmis/bin/container prep --no-check $PUSH_REPO
+          udmis/bin/container prep --no-check ${PUSH_REPO%/*}
           echo Pushing built container as $PUSH_TAG | tee -a $GITHUB_STEP_SUMMARY
       - name: Build and push docker image
         if: ${{ github.event_name == 'push' }}
@@ -295,7 +296,6 @@ jobs:
           cd sites/udmi_site_model/out
           find . -type f | sort | xargs more | cat
       - name: itemized sequencer tests
-        if: ${{ always() }}
         run: |
           bin/test_itemized $GCP_TARGET_PROJECT
       - name: itemized output logs
diff --git a/common/src/main/java/com/google/udmi/util/GeneralUtils.java b/common/src/main/java/com/google/udmi/util/GeneralUtils.java
index 26d2fb2246..14524838e5 100644
--- a/common/src/main/java/com/google/udmi/util/GeneralUtils.java
+++ b/common/src/main/java/com/google/udmi/util/GeneralUtils.java
@@ -49,8 +49,9 @@ public class GeneralUtils {
   private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper()
       .enable(SerializationFeature.INDENT_OUTPUT)
       .setDateFormat(new ISO8601DateFormat())
+      .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
       .setSerializationInclusion(JsonInclude.Include.NON_NULL);
-  private static final ObjectMapper OBJECT_MAPPER_RAW =
+  public static final ObjectMapper OBJECT_MAPPER_RAW =
       OBJECT_MAPPER.copy()
           .enable(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS)
           .enable(Feature.ALLOW_TRAILING_COMMA)
@@ -59,7 +60,7 @@ public class GeneralUtils {
           .setSerializationInclusion(Include.NON_NULL);
   public static final ObjectMapper OBJECT_MAPPER_STRICT =
       OBJECT_MAPPER_RAW.copy()
-          .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
+          .enable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
           .disable(SerializationFeature.INDENT_OUTPUT);
 
   private static final String SEPARATOR = "\n  ";
diff --git a/udmis/.idea/runConfigurations/UdmiServicePod_configured.xml b/udmis/.idea/runConfigurations/UdmiServicePod_configured.xml
index 59670b69c6..d05214369a 100644
--- a/udmis/.idea/runConfigurations/UdmiServicePod_configured.xml
+++ b/udmis/.idea/runConfigurations/UdmiServicePod_configured.xml
@@ -19,4 +19,4 @@
       <option name="Make" enabled="true" />
     </method>
   </configuration>
-</component>
+</component>
\ No newline at end of file
diff --git a/udmis/bin/container b/udmis/bin/container
index c85b0bfc22..ee98c296da 100755
--- a/udmis/bin/container
+++ b/udmis/bin/container
@@ -6,7 +6,7 @@ DROOT=.
 
 function usage {
     echo Error: $*
-    echo Usage: $0 { prep, build, shell, run, push, deploy, update, status, logs, stop } [--no-check] [--selfie] [repo]
+    echo Usage: $0 { prep, build, shell, run, deploy, update, status, logs, stop } [--no-check] [--selfie] [repo]
     echo Project: $GCP_PROJECT
     echo Try starting with: $0 build
     exit 1
@@ -18,8 +18,6 @@ shift || usage missing command
 GCP_PROJECT=$(gcloud config get project)
 REPOSITORY=gcr.io/$GCP_PROJECT
 
-mkdir -p tmp
-
 IMAGE=udmis
 
 if [[ $1 == "--no-check" ]]; then
@@ -60,7 +58,8 @@ current_user=$USER@$HOSTNAME
 
 revparse=`git rev-parse HEAD`
 
-udmi_ref=$REPOSITORY:g${revparse:0:9}
+udmi_ver=g${revparse:0:9}
+udmi_ref=$REPOSITORY/udmi:$udmi_ver
 
 version=`git describe || echo $udmi_ref`
 
@@ -68,7 +67,7 @@ RUNARGS="--rm -ti -v $PWD/var:/udmi -v $HOME/.config:/root/.config --tmpfs /tmp"
 TEMPLATES=$(cd etc; ls k8s_*.yaml)
 
 if [[ $cmd == prep || $cmd == build ]]; then
-    rm -rf var tmp && mkdir -p var
+    rm -rf var tmp && mkdir -p var tmp
     bin/build
     build_time=`date --utc -Imin -r $LIBFILE`
     cp etc/prod_pod.json var/
@@ -84,29 +83,32 @@ EOF
     echo Next try: $0 build
 fi
 
-if [[ $cmd == prep ]]; then
-    true
-elif [[ $cmd == build ]]; then
+if [[ $cmd == build ]]; then
     echo Building Dockerfile.$IMAGE
     docker build -f Dockerfile.$IMAGE -t $IMAGE $DROOT
-    echo Next try: $0 push
-elif [[ $cmd == run ]]; then
-    docker run $RUNARGS $IMAGE
-elif [[ $cmd == shell ]]; then
-    docker run $RUNARGS $IMAGE bash
-elif [[ $cmd == push ]]; then
-    IMAGENAME=$REPOSITORY/$IMAGE
-    docker tag $IMAGE $IMAGENAME
-    docker push $IMAGENAME
-    hash=$(docker images --digests $IMAGENAME | fgrep latest | awk '{print $3}')
-    ihash=$IMAGENAME@$hash
+fi
+
+if [[ $cmd == build && $REPOSITORY != local ]]; then
+    docker tag $IMAGE $udmi_ref
+    docker push $udmi_ref
+    hash=$(docker images --digests ${udmi_ref%:*} | fgrep $udmi_ver | awk '{print $3}')
+    ihash=$udmi_ref@$hash
     for file in $TEMPLATES; do
         if fgrep -q @IMAGE-$IMAGE@ etc/$file; then
             sed < etc/$file > tmp/$file -e "s^@IMAGE-$IMAGE@^$ihash^"
             echo Updated tmp/$file with image $ihash
         fi
     done
-    echo Next try: $0 deploy
+fi
+
+if [[ $cmd == prep ]]; then
+    echo Next try: $0 build
+elif [[ $cmd == build ]]; then
+    echo Next try: $0 '{run, deploy, update}'
+elif [[ $cmd == run ]]; then
+    docker run $RUNARGS $IMAGE
+elif [[ $cmd == shell ]]; then
+    docker run $RUNARGS $IMAGE bash
 elif [[ $cmd == deploy ]]; then
     kubectl apply -f tmp/k8s_pod.yaml
     echo Next try: $0 status
diff --git a/udmis/bin/deploy b/udmis/bin/deploy
deleted file mode 100755
index 1735976d30..0000000000
--- a/udmis/bin/deploy
+++ /dev/null
@@ -1,20 +0,0 @@
-#!/bin/bash -e
-
-if [[ $# != 1 ]]; then
-    echo Usage: $0 project_id
-    false
-fi
-
-project_id=$1
-shift
-
-ROOT=$(dirname $0)/..
-cd $ROOT
-
-bin/setup $project_id
-
-bin/container build
-bin/container push
-bin/container deploy
-sleep 10
-kubectl get pods
diff --git a/udmis/bin/deploy b/udmis/bin/deploy
new file mode 120000
index 0000000000..77c430a5ac
--- /dev/null
+++ b/udmis/bin/deploy
@@ -0,0 +1 @@
+update
\ No newline at end of file
diff --git a/udmis/bin/update b/udmis/bin/update
index d0b26ec379..fc873822e7 100755
--- a/udmis/bin/update
+++ b/udmis/bin/update
@@ -8,13 +8,13 @@ fi
 project_id=$1
 shift
 
+CMD=$(basename $0)
 ROOT=$(dirname $0)/..
 cd $ROOT
 
 bin/setup $project_id
 
 bin/container build
-bin/container push
-bin/container update
+bin/container $CMD
 sleep 10
 kubectl get pods
diff --git a/udmis/etc/prod_pod.json b/udmis/etc/prod_pod.json
index 3b3cfc5b93..487171373f 100644
--- a/udmis/etc/prod_pod.json
+++ b/udmis/etc/prod_pod.json
@@ -24,12 +24,13 @@
     },
     "clearblade-iot-core": {
       "provider": "clearblade",
-      "project_id": "${CLEARBLADE_PROJECT}"
+      "project_id": "${CLEARBLADE_PROJECT}",
+      "options": "distributor=stately"
     },
     "gcp-iot-core": {
       "provider": "gcp",
       "project_id": "${GCP_PROJECT}",
-      "options": "${GCP_IOT_OPTIONS}"
+      "options": "${GCP_IOT_OPTIONS},distributor=stately"
     }
   },
   "distributors": {
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/access/ClearBladeIotAccessProvider.java b/udmis/src/main/java/com/google/bos/udmi/service/access/ClearBladeIotAccessProvider.java
index 1ace7685f0..5dcbd05eaf 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/access/ClearBladeIotAccessProvider.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/access/ClearBladeIotAccessProvider.java
@@ -7,10 +7,10 @@
 import static com.google.udmi.util.GeneralUtils.friendlyStackTrace;
 import static com.google.udmi.util.GeneralUtils.ifNotNullGet;
 import static com.google.udmi.util.GeneralUtils.ifNotNullThen;
+import static com.google.udmi.util.GeneralUtils.ifNotTrueThen;
 import static com.google.udmi.util.GeneralUtils.ifTrueThen;
 import static com.google.udmi.util.GeneralUtils.isTrue;
 import static com.google.udmi.util.JsonUtil.getDate;
-import static com.google.udmi.util.JsonUtil.getTimestamp;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 import static java.util.Optional.ofNullable;
@@ -53,10 +53,7 @@
 import com.google.common.collect.BiMap;
 import com.google.common.collect.ImmutableBiMap;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableSet;
 import com.google.udmi.util.GeneralUtils;
-import java.time.Duration;
-import java.time.Instant;
 import java.util.AbstractMap.SimpleEntry;
 import java.util.Base64;
 import java.util.Date;
@@ -66,7 +63,6 @@
 import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Set;
-import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -103,8 +99,11 @@ public class ClearBladeIotAccessProvider extends IotAccessBase {
    * Create a new instance for interfacing with GCP IoT Core.
    */
   public ClearBladeIotAccessProvider(IotAccess iotAccess) {
+    super(iotAccess);
     projectId = getProjectId(iotAccess);
     ifTrueThen(isEnabled(), this::fetchRegistryRegions);
+    ifNotTrueThen(isEnabled(),
+        () -> warn("Clearblade access provided disabled because project id is null or empty"));
   }
 
   private static Credential convertIot(DeviceCredential device) {
@@ -147,6 +146,27 @@ protected DeviceManagerClient getDeviceManagerClient() {
     return new DeviceManagerClient();
   }
 
+  @NotNull
+  protected Set<String> getRegistriesForRegion(String region) {
+    try {
+      DeviceManagerClient deviceManagerClient = getDeviceManagerClient();
+      ListDeviceRegistriesRequest request = ListDeviceRegistriesRequest.Builder.newBuilder()
+          .setParent(LocationName.of(projectId, region).getLocationFullName())
+          .build();
+      ListDeviceRegistriesResponse response = deviceManagerClient.listDeviceRegistries(request);
+      requireNonNull(response, "get registries response is null");
+      List<DeviceRegistry> deviceRegistries = response.getDeviceRegistriesList();
+      Set<String> registries =
+          ofNullable(deviceRegistries).orElseGet(ImmutableList::of).stream()
+              .map(registry -> registry.toBuilder().getId())
+              .collect(Collectors.toSet());
+      debug("Fetched " + registries.size() + " registries for region " + region);
+      return registries;
+    } catch (Exception e) {
+      throw new RuntimeException("While fetching registries for region " + region, e);
+    }
+  }
+
   @Override
   protected boolean isEnabled() {
     return !isNullOrEmpty(projectId);
@@ -276,6 +296,32 @@ private CloudModel deleteDevice(String registryId, Device device) {
     }
   }
 
+  @NotNull
+  private HashMap<String, CloudModel> fetchDevices(String deviceRegistryId, String gatewayId) {
+    String location = getRegistryLocation(deviceRegistryId);
+    DeviceManagerClient deviceManagerClient = getDeviceManagerClient();
+    GatewayListOptions gatewayListOptions = ifNotNullGet(gatewayId, this::getGatewayListOptions);
+    String registryFullName =
+        RegistryName.of(projectId, location, deviceRegistryId).getRegistryFullName();
+    String pageToken = null;
+    HashMap<String, CloudModel> collect = new HashMap<>();
+    do {
+      DevicesListRequest request = DevicesListRequest.Builder.newBuilder().setParent(
+              registryFullName)
+          .setGatewayListOptions(gatewayListOptions)
+          .setPageToken(pageToken)
+          .build();
+      DevicesListResponse response = deviceManagerClient.listDevices(request);
+      requireNonNull(response, "DeviceRegistriesList fetch failed");
+      Map<String, CloudModel> responseMap =
+          response.getDevicesList().stream().map(ClearBladeIotAccessProvider::convertToEntry)
+              .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+      collect.putAll(responseMap);
+      pageToken = response.getNextPageToken();
+    } while (pageToken != null);
+    return collect;
+  }
+
   private String getDeviceName(String registryId, String deviceId) {
     return DeviceName.of(projectId, getRegistryLocation(registryId), registryId, deviceId)
         .toString();
@@ -298,27 +344,6 @@ private String getProjectId(IotAccess iotAccess) {
     }
   }
 
-  @NotNull
-  protected Set<String> getRegistriesForRegion(String region) {
-    try {
-      DeviceManagerClient deviceManagerClient = getDeviceManagerClient();
-      ListDeviceRegistriesRequest request = ListDeviceRegistriesRequest.Builder.newBuilder()
-          .setParent(LocationName.of(projectId, region).getLocationFullName())
-          .build();
-      ListDeviceRegistriesResponse response = deviceManagerClient.listDeviceRegistries(request);
-      requireNonNull(response, "get registries response is null");
-      List<DeviceRegistry> deviceRegistries = response.getDeviceRegistriesList();
-      Set<String> registries =
-          ofNullable(deviceRegistries).orElseGet(ImmutableList::of).stream()
-              .map(registry -> registry.toBuilder().getId())
-              .collect(Collectors.toSet());
-      debug("Fetched " + registries.size() + " registries for region " + region);
-      return registries;
-    } catch (Exception e) {
-      throw new RuntimeException("While fetching registries for region " + region, e);
-    }
-  }
-
   private String getRegistryLocation(String registry) {
     return getRegistryRegion(registry);
   }
@@ -362,32 +387,6 @@ private CloudModel listRegistryDevices(String deviceRegistryId, String gatewayId
     }
   }
 
-  @NotNull
-  private HashMap<String, CloudModel> fetchDevices(String deviceRegistryId, String gatewayId) {
-    String location = getRegistryLocation(deviceRegistryId);
-    DeviceManagerClient deviceManagerClient = getDeviceManagerClient();
-    GatewayListOptions gatewayListOptions = ifNotNullGet(gatewayId, this::getGatewayListOptions);
-    String registryFullName =
-        RegistryName.of(projectId, location, deviceRegistryId).getRegistryFullName();
-    String pageToken = null;
-    HashMap<String, CloudModel> collect = new HashMap<>();
-    do {
-      DevicesListRequest request = DevicesListRequest.Builder.newBuilder().setParent(
-              registryFullName)
-          .setGatewayListOptions(gatewayListOptions)
-          .setPageToken(pageToken)
-          .build();
-      DevicesListResponse response = deviceManagerClient.listDevices(request);
-      requireNonNull(response, "DeviceRegistriesList fetch failed");
-      Map<String, CloudModel> responseMap =
-          response.getDevicesList().stream().map(ClearBladeIotAccessProvider::convertToEntry)
-              .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
-      collect.putAll(responseMap);
-      pageToken = response.getNextPageToken();
-    } while (pageToken != null);
-    return collect;
-  }
-
   private void unbindDevice(String registryId, String gatewayId, String proxyId) {
     try {
       debug(format("Unbind %s: %s from %s", registryId, proxyId, gatewayId));
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/access/DynamicIotAccessProvider.java b/udmis/src/main/java/com/google/bos/udmi/service/access/DynamicIotAccessProvider.java
index 9c0da43cd8..10530654e4 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/access/DynamicIotAccessProvider.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/access/DynamicIotAccessProvider.java
@@ -38,6 +38,7 @@ public class DynamicIotAccessProvider extends IotAccessBase {
    * Create a new instance for interfacing with multiple providers.
    */
   public DynamicIotAccessProvider(IotAccess iotAccess) {
+    super(iotAccess);
     providerList = Arrays.asList(iotAccess.project_id.split(","));
   }
 
@@ -135,6 +136,11 @@ public void sendCommandBase(String registryId, String deviceId, SubFolder folder
     getProviderFor(registryId).sendCommandBase(registryId, deviceId, folder, message);
   }
 
+  @Override
+  public void updateRegistryRegions(Map<String, String> regions) {
+    providers.values().forEach(provider -> provider.updateRegistryRegions(regions));
+  }
+
   @Override
   public void setProviderAffinity(String registryId, String deviceId, String providerId) {
     if (providerId != null) {
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/access/GcpIotAccessProvider.java b/udmis/src/main/java/com/google/bos/udmi/service/access/GcpIotAccessProvider.java
index b8ccc65899..2f77d9aeed 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/access/GcpIotAccessProvider.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/access/GcpIotAccessProvider.java
@@ -94,8 +94,8 @@ public class GcpIotAccessProvider extends IotAccessBase {
    * TODO: Need to implement page tokens for all requisite API calls.
    */
   public GcpIotAccessProvider(IotAccess iotAccess) {
-    String options = variableSubstitution(iotAccess.options, null);
-    if (!ENABLED_OPTION.equals(options)) {
+    super(iotAccess);
+    if (!options.containsKey(ENABLED_OPTION)) {
       warn("access provider disabled, missing option '%s'", ENABLED_OPTION);
       projectId = null;
       cloudIotService = null;
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/access/IotAccessBase.java b/udmis/src/main/java/com/google/bos/udmi/service/access/IotAccessBase.java
index fe6efafcb7..478919bfec 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/access/IotAccessBase.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/access/IotAccessBase.java
@@ -11,14 +11,17 @@
 import static java.util.Objects.requireNonNull;
 import static java.util.Optional.ofNullable;
 
+import com.google.bos.udmi.service.core.DistributorPipe;
 import com.google.bos.udmi.service.core.ProcessorBase.PreviousParseException;
 import com.google.bos.udmi.service.pod.ContainerBase;
+import com.google.bos.udmi.service.pod.UdmiServicePod;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import java.time.Duration;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.time.temporal.TemporalAmount;
+import java.util.Arrays;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
@@ -28,9 +31,11 @@
 import java.util.stream.Collectors;
 import org.jetbrains.annotations.NotNull;
 import udmi.schema.CloudModel;
+import udmi.schema.Envelope;
 import udmi.schema.Envelope.SubFolder;
 import udmi.schema.IotAccess;
 import udmi.schema.IotAccess.IotProvider;
+import udmi.schema.UdmiState;
 
 /**
  * Generic interface for accessing iot device management.
@@ -53,8 +58,14 @@ public abstract class IotAccessBase extends ContainerBase {
       IotProvider.GCP, GcpIotAccessProvider.class,
       IotProvider.LOCAL, LocalIotAccessProvider.class
   );
+  final Map<String, Object> options;
   private CompletableFuture<Map<String, String>> registryRegions;
   private Instant regionRetry = Instant.now();
+  private DistributorPipe distributor;
+
+  public IotAccessBase(IotAccess iotAccess) {
+    options = parseOptions(iotAccess);
+  }
 
   /**
    * Factory constructor for new instances.
@@ -79,6 +90,17 @@ private static String getBackoffKey(String registryId, String deviceId) {
 
   protected abstract Entry<Long, String> fetchConfig(String registryId, String deviceId);
 
+  /**
+   * Update the cached registry regions with any incremental updates.
+   */
+  public void updateRegistryRegions(Map<String, String> regions) {
+    try {
+      registryRegions.get().putAll(regions);
+    } catch (Exception e) {
+      throw new RuntimeException("While updating registry regions", e);
+    }
+  }
+
   protected Map<String, String> fetchRegistryRegions() {
     Map<String, String> regionMap = CLOUD_REGIONS.stream().flatMap(
         region -> getRegistriesForRegion(region).stream()
@@ -104,9 +126,13 @@ protected String getRegistryRegion(String registryId) {
 
   protected synchronized String populateRegistryRegions(String registryId) {
     if (regionRetry.isBefore(Instant.now())) {
+      regionRetry = Instant.now().plus(REGION_RETRY_BACKOFF);
+      Map<String, String> previousRegions =
+          ifNotNullGet(registryRegions, regions -> regions.getNow(ImmutableMap.of()));
       registryRegions = new CompletableFuture<>();
       registryRegions.complete(fetchRegistryRegions());
-      regionRetry = Instant.now().plus(REGION_RETRY_BACKOFF);
+      Map<String, String> currentRegions = registryRegions.getNow(ImmutableMap.of());
+      ifNotNullThen(previousRegions, () -> disseminateDifference(previousRegions, currentRegions));
     }
     return getCompletedRegistryRegion(registryId);
   }
@@ -126,6 +152,17 @@ private String checkedUpdate(String registryId, String deviceId, Long version, S
     return updateConfig(registryId, deviceId, updated, version);
   }
 
+  private void disseminateDifference(Map<String, String> previousRegions,
+      Map<String, String> currentRegions) {
+    Map<String, String> newRegions = currentRegions.entrySet().stream()
+        .filter(entry -> !previousRegions.containsKey(entry.getKey())).collect(
+            Collectors.toMap(Entry::getKey, Entry::getValue));
+    UdmiState udmiState = new UdmiState();
+    udmiState.regions = newRegions;
+    Envelope envelope = new Envelope();
+    distributor.distribute(envelope, udmiState);
+  }
+
   private String getCompletedRegistryRegion(String registryId) {
     try {
       return registryRegions.get().get(registryId);
@@ -169,6 +206,7 @@ private String safeMunge(Function<String, String> munger, Entry<Long, String> co
   @Override
   public void activate() {
     super.activate();
+    distributor = UdmiServicePod.maybeGetComponent((String) options.get("distributor"));
     if (isEnabled()) {
       populateRegistryRegions(REFLECT_REGISTRY);
     }
@@ -258,6 +296,16 @@ public AbortLoopException(String message) {
     }
   }
 
+  Map<String, Object> parseOptions(IotAccess iotAccess) {
+    String options = variableSubstitution(iotAccess.options, null);
+    if (options == null) {
+      return ImmutableMap.of();
+    }
+    String[] parts = options.split(",");
+    return Arrays.stream(parts).map(String::trim).map(option -> option.split("=", 2))
+        .collect(Collectors.toMap(x -> x[0], x -> x.length > 1 ? x[1] : true));
+  }
+
   abstract String fetchRegistryMetadata(String registryId, String metadataKey);
 
 }
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/access/LocalIotAccessProvider.java b/udmis/src/main/java/com/google/bos/udmi/service/access/LocalIotAccessProvider.java
index ec2ed41b41..a9cee3bb50 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/access/LocalIotAccessProvider.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/access/LocalIotAccessProvider.java
@@ -30,6 +30,7 @@ public class LocalIotAccessProvider extends IotAccessBase {
    * Create a new instance for interfacing with multiple providers.
    */
   public LocalIotAccessProvider(IotAccess iotAccess) {
+    super(iotAccess);
   }
 
   @Override
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/core/DistributorPipe.java b/udmis/src/main/java/com/google/bos/udmi/service/core/DistributorPipe.java
index c29223aa63..d076cd701b 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/core/DistributorPipe.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/core/DistributorPipe.java
@@ -58,10 +58,10 @@ public void shutdown() {
   /**
    * Distribute a message (broadcast).
    */
-  public void distribute(Envelope envelope, UdmiState toolState) {
+  public void distribute(Envelope envelope, Object toolState) {
     try {
-      debug("Distributing %s for %s %s", toolState.getClass().getSimpleName(),
-          envelope.deviceId, envelope.transactionId);
+      debug("Distributing %s for %s/%s", toolState.getClass().getSimpleName(),
+          envelope.deviceRegistryId, envelope.deviceId);
       envelope.gatewayId = clientId;
       dispatcher.publish(dispatcher.makeMessageBundle(envelope, toolState));
     } catch (Exception e) {
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/core/ProcessorBase.java b/udmis/src/main/java/com/google/bos/udmi/service/core/ProcessorBase.java
index 53f1c33291..db1ddc2f54 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/core/ProcessorBase.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/core/ProcessorBase.java
@@ -108,6 +108,7 @@ protected void exceptionHandler(Exception e) {
     String message = Common.getExceptionMessage(e);
     String payload = friendlyStackTrace(e);
     error(format("Received message exception: %s", payload));
+    e.printStackTrace();
     BundleException bundleException = new BundleException(message, toStringMap(envelope), payload);
     reflectError(SubType.EVENT, bundleException);
   }
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/core/ReflectProcessor.java b/udmis/src/main/java/com/google/bos/udmi/service/core/ReflectProcessor.java
index 5ec2b6a715..425a1499d5 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/core/ReflectProcessor.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/core/ReflectProcessor.java
@@ -16,6 +16,7 @@
 import static com.google.udmi.util.JsonUtil.fromStringStrict;
 import static com.google.udmi.util.JsonUtil.loadFileStrictRequired;
 import static com.google.udmi.util.JsonUtil.stringify;
+import static com.google.udmi.util.JsonUtil.stringifyTerse;
 import static com.google.udmi.util.JsonUtil.toMap;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
@@ -28,6 +29,7 @@
 import java.io.File;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.stream.Collectors;
 import udmi.schema.CloudModel;
 import udmi.schema.CloudModel.Operation;
 import udmi.schema.Envelope;
@@ -44,6 +46,7 @@
 public class ReflectProcessor extends ProcessorBase {
 
   public static final String PAYLOAD_KEY = "payload";
+  public static final String HOSTNAME = System.getenv("HOSTNAME");
   static final String DEPLOY_FILE = "var/deployed_version.json";
   static final SetupUdmiConfig DEPLOYED_CONFIG =
       loadFileStrictRequired(SetupUdmiConfig.class, new File(DEPLOY_FILE));
@@ -137,13 +140,19 @@ private void processReflection(Envelope reflection, Envelope envelope,
       Map<String, Object> payload) {
     debug("Processing reflection %s/%s %s", envelope.subType, envelope.subFolder,
         envelope.transactionId);
-    iotAccess.setProviderAffinity(envelope.deviceRegistryId, envelope.deviceId, reflection.source);
+    updateProviderAffinity(envelope, reflection.source);
     CloudModel result = getReflectionResult(envelope, payload);
     ifNotNullThen(result,
         v -> debug("Reflection result %s: %s", envelope.transactionId, envelope.subType));
     ifNotNullThen(result, v -> sendReflectCommand(reflection, envelope, result));
   }
 
+  private void processStateUpdate(Envelope envelope, StateUpdate stateUpdate) {
+    debug("Updating last_start b/c of device state query %s/%s",
+        envelope.deviceRegistryId, envelope.deviceId);
+    updateLastStart(envelope, stateUpdate);
+  }
+
   private CloudModel queryCloudDevice(Envelope attributes) {
     return iotAccess.fetchDevice(attributes.deviceRegistryId, attributes.deviceId);
   }
@@ -170,12 +179,6 @@ private CloudModel queryDeviceState(Envelope attributes) {
     }
   }
 
-  private void processStateUpdate(Envelope envelope, StateUpdate stateUpdate) {
-    debug("Updating last_start b/c of device state query %s/%s",
-        envelope.deviceRegistryId, envelope.deviceId);
-    updateLastStart(envelope, stateUpdate);
-  }
-
   private CloudModel reflectModel(Envelope attributes, CloudModel request) {
     return requireNonNull(
         iotAccess.modelDevice(attributes.deviceRegistryId, attributes.deviceId, request),
@@ -221,6 +224,7 @@ private void reflectStateHandler(Envelope envelope, UdmiState toolState) {
     udmiConfig.last_state = toolState.timestamp;
     udmiConfig.setup = new SetupUdmiConfig();
     copyFields(DEPLOYED_CONFIG, udmiConfig.setup, false);
+    udmiConfig.setup.hostname = HOSTNAME;
     udmiConfig.setup.udmi_version = UDMI_VERSION;
     udmiConfig.setup.functions_min = FUNCTIONS_VERSION_MIN;
     udmiConfig.setup.functions_max = FUNCTIONS_VERSION_MAX;
@@ -246,6 +250,18 @@ private void sendReflectCommand(Envelope reflection, Envelope message, Object pa
     iotAccess.sendCommand(reflectRegistry, deviceRegistry, SubFolder.UDMI, stringify(message));
   }
 
+  private void updateProviderAffinity(Envelope envelope, String source) {
+    debug("Setting affinity for %s/%s to %s", envelope.deviceRegistryId, envelope.deviceId, source);
+    iotAccess.setProviderAffinity(envelope.deviceRegistryId, envelope.deviceId, source);
+  }
+
+  private void updateRegistryRegions(Map<String, String> regions) {
+    debug("Updating registry regions: ",
+        regions.entrySet().stream().map(entry -> format("%s:%s", entry.getKey(), entry.getValue()))
+            .collect(Collectors.joining(", ")));
+    iotAccess.updateRegistryRegions(regions);
+  }
+
   @Override
   public void activate() {
     debug("Deployment configuration: " + stringify(DEPLOYED_CONFIG));
@@ -253,8 +269,9 @@ public void activate() {
   }
 
   void updateAwareness(Envelope envelope, UdmiState toolState) {
-    debug("Processing UdmiState for %s/%s %s", envelope.deviceId, envelope.source,
-        stringify(toolState));
-    iotAccess.setProviderAffinity(envelope.deviceRegistryId, envelope.deviceId, envelope.source);
+    debug("Processing UdmiState for %s/%s: %s", envelope.deviceRegistryId, envelope.deviceId,
+        stringifyTerse(toolState));
+    ifNotNullThen(toolState.setup, setup -> updateProviderAffinity(envelope, envelope.source));
+    ifNotNullThen(toolState.regions, this::updateRegistryRegions);
   }
 }
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageBase.java b/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageBase.java
index b03a6f0df4..9261e0baf5 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageBase.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageBase.java
@@ -26,13 +26,10 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingDeque;
-import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 import org.jetbrains.annotations.Nullable;
 import org.jetbrains.annotations.TestOnly;
@@ -198,8 +195,8 @@ private void messageLoop(String id) {
             return;
           }
           envelope = bundle.envelope;
-          debug("Processing %s/%s %s %s -> %s", envelope.subType, envelope.subFolder,
-              envelope.transactionId, queueIdentifier(), dispatcher);
+          debug("Processing %s %s/%s %s", this, envelope.subType, envelope.subFolder,
+              envelope.transactionId);
           if (ERROR_MESSAGE_MARKER.equals(envelope.transactionId)) {
             throw new RuntimeException("Exception due to test-induced error");
           }
@@ -254,7 +251,7 @@ private void receiveException(Map<String, String> attributesMap, String messageS
   public void activate(Consumer<Bundle> bundleConsumer) {
     dispatcher = bundleConsumer;
     ensureSourceQueue();
-    debug("Handling %s to %08x", this, Objects.hash(bundleConsumer));
+    debug("Handling %s", this);
     handleQueue();
     activated = true;
   }
@@ -315,7 +312,7 @@ public void terminate() {
 
   @Override
   public String toString() {
-    return format("MessagePipe %s", queueIdentifier());
+    return format("MessagePipe %s => %s", queueIdentifier(), Objects.hash(dispatcher));
   }
 
   /**
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageDispatcherImpl.java b/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageDispatcherImpl.java
index d24919ae3e..1d5c579afa 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageDispatcherImpl.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageDispatcherImpl.java
@@ -68,7 +68,7 @@ public class MessageDispatcherImpl extends ContainerBase implements MessageDispa
 
   private final MessagePipe messagePipe;
   private final Map<Object, Envelope> messageEnvelopes = new ConcurrentHashMap<>();
-  private final Map<Class<?>, Consumer<Object>> handlers = new HashMap<>();
+  private final Map<Class<?>, Consumer<Object>> handlers = new ConcurrentHashMap<>();
   private final Map<Class<?>, AtomicInteger> handlerCounts = new ConcurrentHashMap<>();
   private final String projectId;
   private final ThreadLocal<Envelope> threadEnvelope = new ThreadLocal<>();
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/PubSubPipe.java b/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/PubSubPipe.java
index 239337af76..022d8e1b6e 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/PubSubPipe.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/PubSubPipe.java
@@ -64,6 +64,9 @@ public PubSubPipe(EndpointConfiguration configuration) {
           "no project id defined in configuration as 'hostname'");
       publisher = ifNotNullGet(configuration.send_id, this::getPublisher);
       subscriber = ifNotNullGet(configuration.recv_id, this::getSubscriber);
+      String subscriptionName = ifNotNullGet(subscriber, Subscriber::getSubscriptionNameString);
+      String topicName = ifNotNullGet(publisher, Publisher::getTopicNameString);
+      debug("PubSub %s s -> %s", super.toString(), subscriptionName, topicName);
     } catch (Exception e) {
       throw new RuntimeException("While creating PubSub pipe", e);
     }
@@ -138,13 +141,6 @@ public void receiveMessage(PubsubMessage message, AckReplyConsumer reply) {
     debug("Receive message took %ss", seconds);
   }
 
-  @Override
-  public String toString() {
-    String subscriptionName = ifNotNullGet(subscriber, Subscriber::getSubscriptionNameString);
-    String topicName = ifNotNullGet(publisher, Publisher::getTopicNameString);
-    return format("PubSub %s -> %s", subscriptionName, topicName);
-  }
-
   Publisher getPublisher(String topicName) {
     try {
       ProjectTopicName projectTopicName = ProjectTopicName.of(projectId, topicName);
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/SimpleMqttPipe.java b/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/SimpleMqttPipe.java
index afd23538d3..3cd972c82a 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/SimpleMqttPipe.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/SimpleMqttPipe.java
@@ -3,6 +3,7 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.udmi.util.GeneralUtils.friendlyStackTrace;
 import static com.google.udmi.util.GeneralUtils.ifNotNullThen;
+import static com.google.udmi.util.GeneralUtils.ifTrueThen;
 import static com.google.udmi.util.JsonUtil.stringify;
 import static com.google.udmi.util.JsonUtil.toMap;
 import static com.google.udmi.util.JsonUtil.toStringMap;
@@ -51,39 +52,41 @@ public SimpleMqttPipe(EndpointConfiguration config) {
     namespace = config.hostname;
     endpoint = config;
     mqttClient = createMqttClient();
-    tryConnect();
+    tryConnect(false);
     scheduledFuture = Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(
-        SimpleMqttPipe.this::tryConnect, RECONNECT_SEC, RECONNECT_SEC, TimeUnit.SECONDS);
+        () -> SimpleMqttPipe.this.tryConnect(true), 0, RECONNECT_SEC, TimeUnit.SECONDS);
   }
 
   public static MessagePipe fromConfig(EndpointConfiguration config) {
     return new SimpleMqttPipe(config);
   }
 
-  private synchronized void connect() {
+  private void connect(boolean forceDisconnect) {
     try {
-      if (mqttClient.isConnected()) {
-        return;
+      synchronized (mqttClient) {
+        if (mqttClient.isConnected()) {
+          return;
+        }
+        debug("Attempting connection of mqtt client %s", clientId);
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
+        options.setMaxInflight(PUBLISH_THREAD_COUNT * 2);
+        options.setConnectionTimeout(INITIALIZE_TIME_MS);
+
+        ifNotNullThen(endpoint.auth_provider, provider -> {
+          Basic basicAuth = checkNotNull(provider.basic, "basic auth not defined");
+          options.setUserName(checkNotNull(basicAuth.username, "MQTT username not defined"));
+          options.setPassword(
+              checkNotNull(basicAuth.password, "MQTT password not defined").toCharArray());
+        });
+
+        mqttClient.connect(options);
+        info("Connection established to mqtt server as " + clientId);
+        subscribeToMessages();
       }
-      debug("Attempting connection of mqtt client %s", clientId);
-      MqttConnectOptions options = new MqttConnectOptions();
-      options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
-      options.setMaxInflight(PUBLISH_THREAD_COUNT * 2);
-      options.setConnectionTimeout(INITIALIZE_TIME_MS);
-
-      ifNotNullThen(endpoint.auth_provider, provider -> {
-        Basic basicAuth = checkNotNull(provider.basic, "basic auth not defined");
-        options.setUserName(checkNotNull(basicAuth.username, "MQTT username not defined"));
-        options.setPassword(
-            checkNotNull(basicAuth.password, "MQTT password not defined").toCharArray());
-      });
-
-      mqttClient.connect(options);
-      info("Connection established to mqtt server as " + clientId);
-      subscribeToMessages();
     } catch (Exception e) {
       // Sometimes a forced disconnect is necessary else the connection attempt gets stuck somehow.
-      forceDisconnect();
+      ifTrueThen(forceDisconnect, this::forcedisconnect);
       throw new RuntimeException("While connecting mqtt client", e);
     }
   }
@@ -101,7 +104,7 @@ private MqttClient createMqttClient() {
     }
   }
 
-  private void forceDisconnect() {
+  private void forcedisconnect() {
     try {
       mqttClient.disconnectForcibly();
     } catch (Exception e) {
@@ -127,23 +130,25 @@ private String makeMqttTopic(Bundle bundle) {
         : format(TOPIC_FORMAT, namespace, envelope.subType, envelope.subFolder);
   }
 
-  private synchronized void subscribeToMessages() {
+  private void subscribeToMessages() {
     String topic = format(TOPIC_FORMAT, namespace, TOPIC_WILDCARD, TOPIC_WILDCARD);
     try {
-      boolean connected = mqttClient.isConnected();
-      trace("Subscribing %s, active=%s connected=%s", clientId, isActive(), connected);
-      if (isActive() && connected) {
-        mqttClient.subscribe(topic);
-        info("Subscribed %s to topic %s", clientId, topic);
+      synchronized (mqttClient) {
+        boolean connected = mqttClient.isConnected();
+        trace("Subscribing %s, active=%s connected=%s", clientId, isActive(), connected);
+        if (isActive() && connected) {
+          mqttClient.subscribe(topic);
+          info("Subscribed %s to topic %s", clientId, topic);
+        }
       }
     } catch (Exception e) {
       throw new RuntimeException("While subscribing to mqtt topic: " + topic, e);
     }
   }
 
-  private void tryConnect() {
+  private void tryConnect(boolean forceDisconnect) {
     try {
-      connect();
+      connect(forceDisconnect);
     } catch (Exception e) {
       error("While attempting scheduled connect for %s: %s", clientId, friendlyStackTrace(e));
     }
@@ -156,11 +161,8 @@ public void activate(Consumer<Bundle> bundleConsumer) {
   }
 
   @Override
-  public synchronized void publish(Bundle bundle) {
+  public void publish(Bundle bundle) {
     try {
-      if (!mqttClient.isConnected()) {
-        connect();
-      }
       mqttClient.publish(makeMqttTopic(bundle), makeMqttMessage(bundle));
     } catch (Exception e) {
       throw new RuntimeException("While publishing to mqtt client " + clientId, e);
diff --git a/udmis/src/test/java/com/google/bos/udmi/service/access/IotAccessBaseTest.java b/udmis/src/test/java/com/google/bos/udmi/service/access/IotAccessBaseTest.java
new file mode 100644
index 0000000000..3695b95845
--- /dev/null
+++ b/udmis/src/test/java/com/google/bos/udmi/service/access/IotAccessBaseTest.java
@@ -0,0 +1,27 @@
+package com.google.bos.udmi.service.access;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.Test;
+import udmi.schema.IotAccess;
+
+class IotAccessBaseTest {
+
+  @Test
+  public void emptyOptions() {
+    IotAccess access = new IotAccess();
+    LocalIotAccessProvider localIotAccessProvider = new LocalIotAccessProvider(access);
+    assertEquals(ImmutableMap.of(), localIotAccessProvider.options, "empty options");
+  }
+
+  @Test
+  public void optionsParser() {
+    IotAccess access = new IotAccess();
+    access.options = "enable, foo=bar, x=";
+    LocalIotAccessProvider localIotAccessProvider = new LocalIotAccessProvider(access);
+    ImmutableMap<String, Object> expected =
+        ImmutableMap.of("enable", true, "foo", "bar", "x", "");
+    assertEquals(expected, localIotAccessProvider.options, "parsed options object");
+  }
+}
\ No newline at end of file
diff --git a/validator/src/main/java/com/google/bos/iot/core/proxy/MqttPublisher.java b/validator/src/main/java/com/google/bos/iot/core/proxy/MqttPublisher.java
index 01e4b9f792..7efb3ab987 100644
--- a/validator/src/main/java/com/google/bos/iot/core/proxy/MqttPublisher.java
+++ b/validator/src/main/java/com/google/bos/iot/core/proxy/MqttPublisher.java
@@ -153,7 +153,7 @@ private void publishCore(String deviceId, String topic, String payload, Instant
 
   private synchronized void delayStateUpdate(String deviceId) {
     long now = System.currentTimeMillis();
-    long last = lastStateTime.getOrDefault(deviceId, now);
+    long last = lastStateTime.getOrDefault(deviceId, now - STATE_RATE_LIMIT_MS);
     long delta = now - last;
     long delay = STATE_RATE_LIMIT_MS - delta;
     if (delay > 0) {
diff --git a/validator/src/main/java/com/google/daq/mqtt/registrar/LocalDevice.java b/validator/src/main/java/com/google/daq/mqtt/registrar/LocalDevice.java
index d515e7ab5f..bd745ec538 100644
--- a/validator/src/main/java/com/google/daq/mqtt/registrar/LocalDevice.java
+++ b/validator/src/main/java/com/google/daq/mqtt/registrar/LocalDevice.java
@@ -9,10 +9,12 @@
 import static com.google.daq.mqtt.registrar.Registrar.NORMALIZED_JSON;
 import static com.google.daq.mqtt.util.MessageUpgrader.METADATA_SCHEMA;
 import static com.google.udmi.util.Common.VERSION_KEY;
+import static com.google.udmi.util.GeneralUtils.OBJECT_MAPPER_RAW;
 import static com.google.udmi.util.GeneralUtils.OBJECT_MAPPER_STRICT;
 import static com.google.udmi.util.GeneralUtils.compressJsonString;
 import static com.google.udmi.util.GeneralUtils.isTrue;
 import static com.google.udmi.util.GeneralUtils.writeString;
+import static com.google.udmi.util.JsonUtil.OBJECT_MAPPER;
 import static com.google.udmi.util.JsonUtil.asMap;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
@@ -608,7 +610,7 @@ public void validateEnvelope(String registryId, String siteName) {
       envelope.deviceNumId = makeNumId(envelope);
       String envelopeJson = OBJECT_MAPPER_STRICT.writeValueAsString(envelope);
       ProcessingReport processingReport = schemas.get(ENVELOPE_JSON)
-          .validate(OBJECT_MAPPER_STRICT.readTree(envelopeJson));
+          .validate(OBJECT_MAPPER.readTree(envelopeJson));
       if (!processingReport.isSuccess()) {
         processingReport.forEach(action -> {
           throw new RuntimeException("against schema", action.asException());
diff --git a/validator/src/main/java/com/google/daq/mqtt/registrar/Registrar.java b/validator/src/main/java/com/google/daq/mqtt/registrar/Registrar.java
index b1b5a898d3..e8d93137f4 100644
--- a/validator/src/main/java/com/google/daq/mqtt/registrar/Registrar.java
+++ b/validator/src/main/java/com/google/daq/mqtt/registrar/Registrar.java
@@ -737,7 +737,8 @@ private Set<String> fetchCloudDevices() {
       boolean requiresCloud = updateCloudIoT || (idleLimit != null);
       if (requiresCloud) {
         Set<String> devices = cloudIotManager.fetchDeviceIds();
-        System.err.printf("Fetched %d devices from cloud registry %s%n", devices.size(),
+        int size = ifNotNullGet(devices, Set::size, -1);
+        System.err.printf("Fetched %d devices from cloud registry %s%n", size,
             cloudIotManager.getRegistryId());
         return devices;
       } else {
diff --git a/validator/src/main/java/com/google/daq/mqtt/sequencer/SequenceBase.java b/validator/src/main/java/com/google/daq/mqtt/sequencer/SequenceBase.java
index 88e3f571ee..4e186dc07f 100644
--- a/validator/src/main/java/com/google/daq/mqtt/sequencer/SequenceBase.java
+++ b/validator/src/main/java/com/google/daq/mqtt/sequencer/SequenceBase.java
@@ -1310,7 +1310,7 @@ private void processMessage(Map<String, String> attributes, Map<String, Object>
     String commandSignature = format("%s/%s/%s", deviceId, subTypeRaw, subFolderRaw);
     trace("received command " + commandSignature);
 
-    if (SubType.REPLY.value().equals(subTypeRaw)) {
+    if (!SequenceBase.getDeviceId().equals(deviceId)) {
       return;
     }
 
@@ -1319,7 +1319,7 @@ private void processMessage(Map<String, String> attributes, Map<String, Object>
       return;
     }
 
-    if (!SequenceBase.getDeviceId().equals(deviceId)) {
+    if (SubType.REPLY.value().equals(subTypeRaw)) {
       return;
     }
 
diff --git a/validator/src/main/java/com/google/daq/mqtt/util/IotReflectorClient.java b/validator/src/main/java/com/google/daq/mqtt/util/IotReflectorClient.java
index d8dea74e46..ae2f76bdb4 100644
--- a/validator/src/main/java/com/google/daq/mqtt/util/IotReflectorClient.java
+++ b/validator/src/main/java/com/google/daq/mqtt/util/IotReflectorClient.java
@@ -148,7 +148,13 @@ private CloudModel fetchCloudModel(String deviceId) {
   private Map<String, Object> transaction(String deviceId, String topic,
       String message, QuerySpeed speed) {
     // TODO: Publish should return future to avoid race conditions.
-    return waitForReply(messageClient.publish(deviceId, topic, message), speed);
+    String transactionId = messageClient.publish(deviceId, topic, message);
+    Map<String, Object> objectMap = waitForReply(transactionId, speed);
+    String error = (String) objectMap.get("error");
+    if (error != null) {
+      throw new RuntimeException(format("UDMIS error %s: %s", transactionId, error));
+    }
+    return objectMap;
   }
 
   private Map<String, Object> waitForReply(String sentId, QuerySpeed speed) {
@@ -178,7 +184,7 @@ private void processReplies() {
 
         String error = (String) messageBundle.message.get(ERROR_KEY);
         if (error != null) {
-          throw new RuntimeException("UDMIS error: " + error);
+          System.err.println("UDMIS error: " + error);
         }
 
         String transactionId = messageBundle.attributes.get(TRANSACTION_KEY);