Attempt to process correlated events the order in which the arrive

aholmberg · aholmberg · commit 5777c2046985 · 2015-07-09T16:53:51.000-05:00
Events are mostly processed in order. "mostly" because there is still a
race by way of the cluster thread pool executor. Meta class should be
robust against this.
diff --git a/cassandra/cluster.py b/cassandra/cluster.py
@@ -2065,6 +2065,8 @@ def __init__(self, cluster, timeout,
         self._reconnection_handler = None
         self._reconnection_lock = RLock()
 
+        self._event_schedule_times = {}
+
     def connect(self):
         if self._is_shutdown:
             return
@@ -2501,12 +2503,26 @@ def _update_location_info(self, host, datacenter, rack):
         self._cluster.load_balancing_policy.on_up(host)
         return True
 
+    def _delay_for_event_type(self, event_type, delay_window):
+        # this serves to order processing correlated events (received within the window)
+        # the window and randomization still have the desired effect of skew across client instances
+        next_time = self._event_schedule_times.get(event_type, 0)
+        now = self._time.time()
+        if now <= next_time:
+            this_time = next_time + 0.01
+            delay = this_time - now
+        else:
+            delay = random() * delay_window
+            this_time = now + delay
+        self._event_schedule_times[event_type] = this_time
+        return delay
+
     def _handle_topology_change(self, event):
         change_type = event["change_type"]
         addr, port = event["address"]
         if change_type == "NEW_NODE" or change_type == "MOVED_NODE":
             if self._topology_event_refresh_window >= 0:
-                delay = random() * self._topology_event_refresh_window
+                delay = self._delay_for_event_type('topology_change', self._topology_event_refresh_window)
                 self._cluster.scheduler.schedule_unique(delay, self.refresh_node_list_and_token_map)
         elif change_type == "REMOVED_NODE":
             host = self._cluster.metadata.get_host(addr)
@@ -2517,7 +2533,7 @@ def _handle_status_change(self, event):
         addr, port = event["address"]
         host = self._cluster.metadata.get_host(addr)
         if change_type == "UP":
-            delay = 1 + random() * 0.5  # randomness to avoid thundering herd problem on events
+            delay = 1 + self._delay_for_event_type('status_change', 0.5)  # randomness to avoid thundering herd problem on events
             if host is None:
                 # this is the first time we've seen the node
                 self._cluster.scheduler.schedule_unique(delay, self.refresh_node_list_and_token_map)
@@ -2540,7 +2556,7 @@ def _handle_schema_change(self, event):
         usertype = event.get('type')
         function = event.get('function')
         aggregate = event.get('aggregate')
-        delay = random() * self._schema_event_refresh_window
+        delay = self._delay_for_event_type('schema_change', self._schema_event_refresh_window)
         self._cluster.scheduler.schedule_unique(delay, self.refresh_schema, keyspace, table, usertype, function, aggregate)
 
     def wait_for_schema_agreement(self, connection=None, preloaded_results=None, wait_time=None):
diff --git a/cassandra/metadata.py b/cassandra/metadata.py
@@ -236,21 +236,16 @@ def aggregate_changed(self, keyspace, aggregate, aggregate_results):
                 pass
 
     def table_changed(self, keyspace, table, cf_results, col_results, triggers_result):
-        try:
-            keyspace_meta = self.keyspaces[keyspace]
-        except KeyError:
-            # we're trying to update a table in a keyspace we don't know about
-            log.warn("Tried to update schema for table '%s' in unknown keyspace '%s'",
-                      table, keyspace)
-            return
-
-        if not cf_results:
-            # the table was removed
-            keyspace_meta._drop_table_metadata(table)
-        else:
+        if cf_results:
             assert len(cf_results) == 1
+            keyspace_meta = self.keyspaces[keyspace]
             table_meta = self._build_table_metadata(keyspace_meta, cf_results[0], {table: col_results}, {table: triggers_result})
             keyspace_meta._add_table_metadata(table_meta)
+        else:
+            try:
+                self.keyspaces[keyspace]._drop_table_metadata(table)
+            except KeyError:
+                pass
 
     def _keyspace_added(self, ksname):
         if self.token_map:
diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py
@@ -410,12 +410,12 @@ def test_handle_schema_change(self):
             }
             self.cluster.scheduler.reset_mock()
             self.control_connection._handle_schema_change(event)
-            self.cluster.scheduler.schedule_unique.assert_called_once_with(0.0, self.control_connection.refresh_schema, 'ks1', 'table1', None, None, None)
+            self.cluster.scheduler.schedule_unique.assert_called_once_with(ANY, self.control_connection.refresh_schema, 'ks1', 'table1', None, None, None)
 
             self.cluster.scheduler.reset_mock()
             event['table'] = None
             self.control_connection._handle_schema_change(event)
-            self.cluster.scheduler.schedule_unique.assert_called_once_with(0.0, self.control_connection.refresh_schema, 'ks1', None, None, None, None)
+            self.cluster.scheduler.schedule_unique.assert_called_once_with(ANY, self.control_connection.refresh_schema, 'ks1', None, None, None, None)
 
     def test_refresh_disabled(self):
         cluster = MockCluster()
