fix(executions): avoid intermittent StackOverflowError

The use of `Sets.union` was causing such a deeply nested data structure with large amounts of running executions that we get StackOverflowError trying to iterate over it. Changed the implementation to much simpler and RxJava-y form.

orca-spring-batch/src/main/java/com/netflix/spinnaker/orca/batch/SpringBatchActiveExecutionTracker.java

@@ -19,7 +19,6 @@
 import java.util.*;
 import java.util.stream.Collectors;
 import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Sets;
 import com.netflix.spinnaker.orca.ActiveExecutionTracker;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.batch.core.launch.JobOperator;
@@ -118,9 +117,9 @@ public void recordRunningExecutions() {
     log.debug("Checking for running executions");
     Observable
       .from(jobOperator.getJobNames())
-      .map(this::runningExecutions)
-      .reduce(Sets::union)
-      .onErrorResumeNext(Observable.just(emptySet()))
+      .flatMapIterable(this::runningExecutions)
+      .toList()
+      .onErrorResumeNext(Observable.just(emptyList()))
       .subscribe(this::recordExecutions);
   }
 
@@ -180,7 +179,7 @@ private SortedSet<ExecutionRecord> readExecutions(Jedis jedis, String instance)
   /**
    * @param executions the executions currently running on _this_ instance.
    */
-  private void recordExecutions(Set<ExecutionRecord> executions) {
+  private void recordExecutions(Collection<ExecutionRecord> executions) {
     log.info("Currently running {} executions", executions.size());
     try (Jedis jedis = jedisPool.getResource()) {
       jedis.sadd(KEY_INSTANCES, currentInstance);