Merge pull request #123 from oracle/async_review

Async review comments

Author: rjeberhard

src/main/java/oracle/kubernetes/operator/Main.java

@@ -811,12 +811,12 @@ public NextAction apply(Packet packet) {
               WlsClusterConfig wlsClusterConfig = scan.getClusterConfig(clusterName);
               if (wlsClusterConfig != null) {
                 for (WlsServerConfig wlsServerConfig : wlsClusterConfig.getServerConfigs()) {
+                  // done with the current cluster
+                  if (startedCount >= cs.getReplicas() && !startAll) 
+                    continue cluster;
+
                   String serverName = wlsServerConfig.getName();
                   if (!serverName.equals(asName) && !servers.contains(serverName)) {
-                    // done with the current cluster
-                    if (startedCount >= cs.getReplicas() && !startAll) 
-                      continue cluster;
-
                     List<V1EnvVar> env = cs.getEnv();
                     ServerStartup ssi = null;
                     ssl = spec.getServerStartup();
@@ -871,15 +871,15 @@ else if (StartupControlConstants.AUTO_STARTUPCONTROL.equals(sc)) {
                 int startedCount = 0;
                 WlsClusterConfig config = wlsClusterConfig.getValue();
                 for (WlsServerConfig wlsServerConfig : config.getServerConfigs()) {
+                  if (startedCount >= spec.getReplicas())
+                    break;
                   String serverName = wlsServerConfig.getName();
                   if (!serverName.equals(asName) && !servers.contains(serverName)) {
                     // start server
                     servers.add(serverName);
                     ssic.add(new ServerStartupInfo(wlsServerConfig, config, null, null));
+                    startedCount++;
                   }
-                  // outside the serverName check because these servers are already running
-                  if (++startedCount >= spec.getReplicas())
-                    break;
                 }
               }
             }

src/main/java/oracle/kubernetes/operator/work/FiberGate.java

@@ -55,17 +55,18 @@ public void onThrowable(Packet packet, Throwable throwable) {
 
   private static class WaitForOldFiberStep extends Step {
     private final AtomicReference<Fiber> old;
-    private WaitForOldFiberStep current;
+    private final AtomicReference<WaitForOldFiberStep> current;
 
     public WaitForOldFiberStep(Fiber old, Step next) {
       super(next);
       this.old = new AtomicReference<>(old);
-      current = this;
+      current = new AtomicReference<>(this);
     }
 
     @Override
     public NextAction apply(Packet packet) {
-      Fiber o = current != null ? current.old.getAndSet(null) : null;
+      WaitForOldFiberStep c = current.get();
+      Fiber o = c != null ? c.old.getAndSet(null) : null;
       if (o == null) {
         return doNext(packet);
       }
@@ -74,13 +75,13 @@ public NextAction apply(Packet packet) {
         boolean isWillCall = o.cancelAndExitCallback(true, new ExitCallback() {
           @Override
           public void onExit() {
-            current = o.getSPI(WaitForOldFiberStep.class);
+            current.set(o.getSPI(WaitForOldFiberStep.class));
             fiber.resume(packet);
           }
         });
 
         if (!isWillCall) {
-          current = o.getSPI(WaitForOldFiberStep.class);
+          current.set(o.getSPI(WaitForOldFiberStep.class));
           fiber.resume(packet);
         }
       });

src/main/java/oracle/kubernetes/operator/work/Packet.java

@@ -5,17 +5,18 @@
 
 import java.util.AbstractMap;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
 /**
  * Context of a single processing flow.  Acts as a map and as a registry of components.
  * 
  */
 public class Packet extends AbstractMap<String, Object> implements ComponentRegistry, ComponentEx {
-  private final Map<String, Component> components = new HashMap<String, Component>();
-  private final Map<String, Object> delegate = new HashMap<String, Object>();
+  private final ConcurrentMap<String, Component> components = new ConcurrentHashMap<String, Component>();
+  private final ConcurrentMap<String, Object> delegate = new ConcurrentHashMap<String, Object>();
 
   public Packet() {}
 
@@ -33,9 +34,6 @@ public Packet clone() {
   }
 
   public <S> S getSPI(Class<S> spiType) {
-    if (components == null) {
-      return null;
-    }
     for (Component c : components.values()) {
       S s = c.getSPI(spiType);
       if (s != null) {
@@ -66,6 +64,6 @@ public Set<Entry<String, Object>> entrySet() {
 
   @Override
   public Object put(String key, Object value) {
-    return delegate.put(key, value);
+    return value != null ? delegate.put(key, value) : delegate.remove(key);
   }
 }