schema-agreement: avoid fallback after mismatch error

Author: dkropachev

cassandra/cluster.py

@@ -194,6 +194,7 @@ def _connection_reduce_fn(val,import_fn):
 _GRAPH_PAGING_MIN_DSE_VERSION = Version('6.8.0')
 
 _NOT_SET = object()
+_SCHEMA_AGREEMENT_MISMATCHES_ATTR = '_schema_agreement_mismatches'
 
 
 class NoHostAvailable(Exception):
@@ -4337,6 +4338,7 @@ def _wait_for_schema_agreement(self, connection=None, preloaded_results=None, wa
                     if schema_mismatches is not None:
                         log.debug("[control connection] Error during schema agreement check after mismatch: %s",
                                   exc)
+                        setattr(exc, _SCHEMA_AGREEMENT_MISMATCHES_ATTR, schema_mismatches)
                         raise
 
                     fallback_wait = total_timeout - elapsed
@@ -4601,10 +4603,17 @@ def refresh_schema_and_set_result(control_conn, response_future, connection, **k
         use_session_fallback = False
         try:
             response_future.is_schema_agreed = control_conn._refresh_schema(connection, **kwargs)
-        except Exception:
+        except Exception as exc:
             log.exception("Exception refreshing schema in response to schema change:")
             response_future.is_schema_agreed = False
-            use_session_fallback = True
+            schema_mismatches = getattr(exc, _SCHEMA_AGREEMENT_MISMATCHES_ATTR, _NOT_SET)
+            if schema_mismatches is not _NOT_SET:
+                log.debug("Skipping session schema agreement fallback after control connection "
+                          "reported a schema disagreement: %s",
+                          schema_mismatches)
+                response_future.session.submit(control_conn.refresh_schema, **kwargs)
+            else:
+                use_session_fallback = True
 
         if use_session_fallback:
             log.debug("Falling back to session schema agreement check")

tests/unit/test_control_connection.py

@@ -19,7 +19,9 @@
 
 from cassandra import OperationTimedOut, SchemaTargetType, SchemaChangeType
 from cassandra.protocol import ResultMessage, RESULT_KIND_ROWS
-from cassandra.cluster import ControlConnection, _Scheduler, ProfileManager, EXEC_PROFILE_DEFAULT, ExecutionProfile
+from cassandra.cluster import (ControlConnection, _Scheduler, ProfileManager,
+                               EXEC_PROFILE_DEFAULT, ExecutionProfile,
+                               refresh_schema_and_set_result)
 from cassandra.pool import Host
 from cassandra.connection import (EndPoint, DefaultEndPoint, DefaultEndPointFactory,
                                   ConnectionException, ConnectionShutdown, ConnectionBusy)
@@ -352,6 +354,31 @@ def test_wait_for_schema_agreement_raises_connection_error_after_mismatch(self):
         with self.assertRaises(ConnectionShutdown):
             self.control_connection.wait_for_schema_agreement()
 
+    def test_schema_change_refresh_does_not_session_fallback_after_mismatch_then_connection_error(self):
+        session = Mock(is_shutdown=False)
+        session.wait_for_schema_agreement.return_value = True
+        self.cluster.sessions = [session]
+        self.cluster.metadata.refresh = Mock()
+
+        peer_columns = self.connection.peer_results[0]
+        mismatching_peer_rows = [list(row) for row in self.connection.peer_results[1]]
+        mismatching_peer_rows[1][2] = 'b'
+        self.connection.wait_for_responses.side_effect = [
+            _node_meta_results(self.connection.local_results, (peer_columns, mismatching_peer_rows)),
+            ConnectionShutdown("closed")]
+
+        response_future = Mock()
+        response_future.session = session
+        event = {'target_type': SchemaTargetType.TABLE, 'change_type': SchemaChangeType.CREATED,
+                 'keyspace': "keyspace1", "table": "table1"}
+
+        refresh_schema_and_set_result(self.control_connection, response_future, self.connection, **event)
+
+        session.wait_for_schema_agreement.assert_not_called()
+        self.cluster.metadata.refresh.assert_not_called()
+        assert not response_future.is_schema_agreed
+        response_future._set_final_result.assert_called_once_with(None)
+
     def test_wait_for_schema_agreement_does_not_exceed_configured_wait_with_session_fallback(self):
         session = Mock(is_shutdown=False)