fix(validate_rows_per_partitions): use a new validation event

	validate_rows_per_partitions has assertions for the expected number of rows.
	These assertions are now caught and generates a corresponding new event:
	"PartitionRowsValidationEvent". This prevents failing the nemesis thread.
	Fixes: #8292

(cherry picked from commit b6391f9)

defaults/severities.yaml

@@ -137,3 +137,4 @@ ScyllaSysconfigSetupEvent: NORMAL
 TestStepEvent: ERROR
 CpuNotHighEnoughEvent: ERROR
 HWPerforanceEvent: CRITICAL
+PartitionRowsValidationEvent: CRITICAL

sdcm/sct_events/health.py

@@ -104,10 +104,20 @@ def msgfmt(self) -> str:
         return super().msgfmt
 
 
+class PartitionRowsValidationEvent(InformationalEvent):
+    def __init__(self, message: str, severity: Severity = Severity.NORMAL):
+        super().__init__(severity)
+        self.message = message
+
+    @property
+    def msgfmt(self) -> str:
+        return super().msgfmt + ": message={0.message}"
+
+
 DataValidatorEvent.add_subevent_type("DataValidator")
 DataValidatorEvent.add_subevent_type("ImmutableRowsValidator")
 DataValidatorEvent.add_subevent_type("UpdatedRowsValidator")
 DataValidatorEvent.add_subevent_type("DeletedRowsValidator")
 
 
-__all__ = ("ClusterHealthValidatorEvent", "DataValidatorEvent", )
+__all__ = ("ClusterHealthValidatorEvent", "DataValidatorEvent", "PartitionRowsValidationEvent")

sdcm/utils/database_query_utils.py

@@ -23,6 +23,7 @@
 from cassandra import ConsistencyLevel
 
 from sdcm.sct_events import Severity
+from sdcm.sct_events.health import PartitionRowsValidationEvent
 from sdcm.sct_events.system import TestFrameworkEvent
 from sdcm.utils.common import PageFetcher
 from sdcm.utils.decorators import retrying
@@ -153,12 +154,20 @@ def validate_rows_per_partitions(self, ignore_limit_rows_number: bool = False):
                 if not ignore_limit_rows_number and self.limit_rows_number:
                     missing_rows = {key: val for key, val in partitions_dict_after.items() if
                                     val < self.limit_rows_number}
-                    assert not missing_rows, f"Found missing rows for partitions: {missing_rows}"
-                else:
-                    self.tester.assertEqual(self.partitions_dict_before,
-                                            partitions_dict_after,
-                                            msg='Row amount in partitions is not same before and after running of nemesis: '
-                                                f' {partitions_dict_after}')
+                    if missing_rows:
+                        PartitionRowsValidationEvent(
+                            message=f"Found missing rows for partitions: {missing_rows}",
+                            severity=Severity.CRITICAL).publish()
+                        return
+                elif partitions_dict_after != self.partitions_dict_before:
+                    PartitionRowsValidationEvent(
+                        message=f"Row amount in partitions is not same before and after running of nemesis: {partitions_dict_after}",
+                        severity=Severity.CRITICAL).publish()
+                    return
+
+                PartitionRowsValidationEvent(
+                    message="Partition rows number is validated.",
+                    severity=Severity.NORMAL).publish()
 
 
 def get_table_clustering_order(ks_cf: str, ck_name: str, session) -> str: