schema-agreement: propagate post-mismatch connection errors

Author: dkropachev

cassandra/cluster.py

@@ -4333,12 +4333,12 @@ def _wait_for_schema_agreement(self, connection=None, preloaded_results=None, wa
                         log.debug("[control connection] Aborting wait for schema match due to shutdown")
                         return None
 
+                    elapsed = self._time.time() - start
                     if schema_mismatches is not None:
-                        log.warning("Node %s is reporting a schema disagreement: %s",
-                                    connection.endpoint, schema_mismatches)
-                        return False
+                        log.debug("[control connection] Error during schema agreement check after mismatch: %s",
+                                  exc)
+                        raise
 
-                    elapsed = self._time.time() - start
                     fallback_wait = total_timeout - elapsed
                     fallback = self._wait_for_schema_agreement_through_session(fallback_wait)
                     if fallback is not None:

tests/unit/test_control_connection.py

@@ -341,6 +341,17 @@ def test_wait_for_schema_agreement_retries_control_connection_after_mismatch_the
         session.wait_for_schema_agreement.assert_not_called()
         assert self.connection.wait_for_responses.call_count == 3
 
+    def test_wait_for_schema_agreement_raises_connection_error_after_mismatch(self):
+        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")]
+
+        with self.assertRaises(ConnectionShutdown):
+            self.control_connection.wait_for_schema_agreement()
+
     def test_wait_for_schema_agreement_does_not_exceed_configured_wait_with_session_fallback(self):
         session = Mock(is_shutdown=False)