Changeset 3268

Timestamp:

01/12/09 08:49:33 (20 months ago)

Author:

jmoore

Message:

OmeroClustering : Various fixes for the jdbc version

• Properties must be passed via the descriptor not internal.cfg
• Needed to catch various exceptions and better handle cleanup

(cherry picked from commit 0df94cf5fd89fbc63f011fb3a25bf3f34df1b9be)

Location:

trunk

Files:

• components/blitz/src/ome/services/blitz/fire/Ring.java (modified) (11 diffs)
• components/blitz/test/ome/services/blitz/test/ClusteredRingTest.java (modified) (1 diff)
• etc/grid/templates.xml (modified) (1 diff)
• etc/internal.cfg (modified) (1 diff)

trunk/components/blitz/src/ome/services/blitz/fire/Ring.java

@@ -34 +34 @@
 import org.springframework.context.event.ContextClosedEvent;
 import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.dao.EmptyResultDataAccessException;
 import org.springframework.jdbc.BadSqlGrammarException;
 import org.springframework.jdbc.core.simple.ParameterizedRowMapper;
@@ -56 +57 @@
  *@since Beta4
  */
-public class Ring extends _ClusterDisp implements ApplicationListener, ApplicationEventPublisherAware {
+public class Ring extends _ClusterDisp implements ApplicationListener,
+        ApplicationEventPublisherAware {
 
     private final static Log log = LogFactory.getLog(Ring.class);
@@ -74 +76 @@
     private final SimpleJdbcTemplate jdbc;
 
-    private/*final */ApplicationEventPublisher publisher;
-    
+    private/* final */ApplicationEventPublisher publisher;
+
     private/* final */Ice.Communicator communicator;
 
@@ -98 +100 @@
      */
     private/* final */ClusterPrx cluster;
-    
+
     /**
      * Direct proxy value to the {@link SessionManager} in this blitz instance.
@@ -119 +121 @@
     /**
      * Passed to the {@link Discovery} instance for signalling completion from
-     * its background {@link Thread} 
+     * its background {@link Thread}
      */
     public void setApplicationEventPublisher(ApplicationEventPublisher arg0) {
         this.publisher = arg0;
     }
-    
+
     // Configuration and cluster usage
     // =========================================================================
@@ -139 +141 @@
         // The cluster we belong to.
         Ice.ObjectPrx prx = this.communicator.propertyToProxy("ClusterProxy");
+        if (prx == null) {
+            throw new RuntimeException("Could not obtain ClusterProxy. "
+                    + "Is multicast property properly set?");
+        }
         prx = prx.ice_datagram();
+        if (prx == null) {
+            throw new RuntimeException("Could no get datagram proxy. "
+                    + "Please check your multicast configuration.");
+        }
         cluster = ClusterPrxHelper.uncheckedCast(prx);
-        
+
         // Before we add our self we check the validity of the cluster.
         checkClusterAndAddSelf();
@@ -182 +192 @@
             log.info("Sent cluster node uuid: " + uuid);
         } catch (Exception e) {
-            log.warn("Exception while sending cluster node uuid: "+uuid, e);
-        }
-    }
-
-    /**
-     * Called when any node goes down. First we try to remove any redirect for that
-     * instance. Then we try to install ourselves.
+            log.warn("Exception while sending cluster node uuid: " + uuid, e);
+        }
+    }
+
+    /**
+     * Called when any node goes down. First we try to remove any redirect for
+     * that instance. Then we try to install ourselves.
      */
     public void down(String downUuid, Current __current) {
-        removeIfEquals(CONFIG+"redirect", downUuid);
-        if (putIfAbsent(CONFIG+"redirect", this.uuid)) {
-            log.info("Installed self as new redirect: "+uuid);
-        }
-    }
-    
+        removeIfEquals(CONFIG + "redirect", downUuid);
+        if (putIfAbsent(CONFIG + "redirect", this.uuid)) {
+            log.info("Installed self as new redirect: " + uuid);
+        }
+    }
+
     public void destroy() {
         try {
             this.clusterAdapter.deactivate();
-            cluster.down(this.uuid);
             remove(MANAGERS + uuid);
-            removeIfEquals(CONFIG+"redirect", uuid);
+            int count = jdbc.update("delete from session_ring where value = ?", uuid);
+            log.info("Removed "+count+" entries for "+uuid);
             log.info("Disconnected from OMERO.cluster");
         } catch (Exception e) {
             log.error("Error stopping ring " + this, e);
+        } finally {
+            cluster.down(this.uuid);
         }
     }
@@ -330 +342 @@
         return rv;
     }
-    
+
     public void assertNodes(Set<String> nodeUuids) {
         Set<String> managers = knownManagers();
         for (String manager : managers) {
-            System.out.println(manager+"?="+nodeUuids);
             if (!nodeUuids.contains(manager)) {
-                purgeNode(manager);
+                // Also verify this is not ourself, since
+                // possibly we haven't finished registration
+                // yet
+                if (!uuid.equals(manager)) {
+                    purgeNode(manager);
+                }
             }
         }
@@ -387 +403 @@
 
     public String get(String key) {
-        String value = (String) jdbc.queryForObject("select value "
-                + "from session_ring " + "where key = ?", String.class, key);
-        return value;
+        try {
+            String value = (String) jdbc
+                    .queryForObject("select value " + "from session_ring "
+                            + "where key = ?", String.class, key);
+            return value;
+        } catch (EmptyResultDataAccessException erdae) {
+            return null;
+        }
     }
 
@@ -401 +422 @@
                 log.info("Key not found for update: " + key);
             } else {
-                log.info(String.format("Updated key %s with value %s", key,
-                        value));
+                log.info(String.format("Updated key %s to %s", key, value));
             }
         }
@@ -444 +464 @@
      */
     public boolean removeIfEquals(String key, String value) {
-        int count = jdbc
-                .update("delete from session_ring where key = ? and value = ?", key, value);
+        int count = jdbc.update(
+                "delete from session_ring where key = ? and value = ?", key,
+                value);
         if (count == 0) {
             log.info("Key and value do not match: " + key + "=" + value);

trunk/components/blitz/test/ome/services/blitz/test/ClusteredRingTest.java

@@ -147 +147 @@
         fail();
     }
+    
+    @Test
+    public void testIfRedirectIsDeletedAnotherHostTakesOver() throws Exception {
+        fails();
+    }
 }

trunk/etc/grid/templates.xml

@@ -26 +26 @@
           <property name="Ice.MessageSizeMax" value="65536"/> <!-- 64 MB -->
           <property name="Ice.Default.CollocationOptimized" value="0"/>
+          <!-- used to communicate between instances. See omero/Internal.ice -->
+          <property name="Cluster.Endpoints" value="udp -h 239.0.0.5 -p 22222"/>
+          <property name="ClusterProxy" value="Cluster:udp -h 239.0.0.5 -p 22222"/>
         </properties>
       </server>

trunk/etc/internal.cfg

@@ -6 +6 @@
 IceGridAdmin.Username=root
 IceGridAdmin.Password=ome
-
-#
-# Used to communicate between instances
-# See: omero/Internal.ice
-#
-Cluster.Endpoints=udp -h 239.0.0.5 -p 22222
-ClusterProxy=Cluster:udp -h 239.0.0.5 -p 22222