Add tests for parent repair session cleanup

patch by Jaroslaw Grabowski and Berenguer Blasi; reviewed by Ekaterina Dimitrova and Andrés de la Peña for CASSANDRA-16446

Co-authored-by: jtgrabowski <jaroslaw.grabowski@datastax.com>
Co-authored-by: Bereng <berenguerblasi@gmail.com>

repair_tests/incremental_repair_test.py

@@ -16,7 +16,7 @@
 
 from dtest import Tester, create_ks, create_cf
 from tools.assertions import assert_almost_equal, assert_one
-from tools.data import insert_c1c2
+from tools.data import create_c1c2_table, insert_c1c2
 from tools.misc import new_node, ImmutableMapping
 from tools.jmxutils import make_mbean, JolokiaAgent
 
@@ -33,6 +33,16 @@ class ConsistentState(object):
     FAILED = 5
 
 
+def assert_parent_repair_session_count(nodes, expected):
+    for node in nodes:
+        with JolokiaAgent(node) as jmx:
+            result = jmx.execute_method("org.apache.cassandra.db:type=RepairService",
+                                        "parentRepairSessionsCount")
+            assert expected == result, "The number of cached ParentRepairSessions should be {} but was {}. " \
+                                       "This may mean that PRS objects are leaking on node {}. Check " \
+                                       "ActiveRepairService for PRS clean up code.".format(expected, result, node.name)
+
+
 class TestIncRepair(Tester):
 
     @pytest.fixture(autouse=True)
@@ -1070,6 +1080,22 @@ def test_repaired_tracking_with_mismatching_replicas(self):
                                                      expect_confirmed_inconsistencies=True,
                                                      expect_read_repair=False)
 
+    @since('4.0')
+    def test_parent_repair_session_cleanup(self):
+        """
+        Calls incremental repair on 3 node cluster and verifies if all ParentRepairSession objects are cleaned
+        @jira_ticket CASSANDRA-16446
+        """
+        self.cluster.populate(3).start()
+        session = self.patient_cql_connection(self.cluster.nodelist()[0])
+        create_ks(session, 'ks', 2)
+        create_c1c2_table(self, session)
+
+        for node in self.cluster.nodelist():
+            node.repair(options=['ks'])
+
+        assert_parent_repair_session_count(self.cluster.nodelist(), 0)
+
     def setup_for_repaired_data_tracking(self):
         self.fixture_dtest_setup.setup_overrides.cluster_options = ImmutableMapping({'hinted_handoff_enabled': 'false',
                                                                                      'num_tokens': 1,

repair_tests/preview_repair_test.py

@@ -4,7 +4,9 @@
 from cassandra import ConsistencyLevel
 from cassandra.query import SimpleStatement
 
-from dtest import Tester
+from dtest import Tester, create_ks
+from repair_tests.incremental_repair_test import assert_parent_repair_session_count
+from tools.data import create_c1c2_table
 
 since = pytest.mark.since
 
@@ -18,6 +20,22 @@ def assert_no_repair_history(self, session):
         rows = session.execute("select * from system_distributed.parent_repair_history")
         assert rows.current_rows == []
 
+    @since('4.0')
+    def test_parent_repair_session_cleanup(self):
+        """
+        Calls incremental repair preview on 3 node cluster and verifies if all ParentRepairSession objects are cleaned
+        @jira_ticket CASSANDRA-16446
+        """
+        self.cluster.populate(3).start()
+        session = self.patient_cql_connection(self.cluster.nodelist()[0])
+        create_ks(session, 'ks', 2)
+        create_c1c2_table(self, session)
+
+        for node in self.cluster.nodelist():
+            node.repair(options=['ks', '--preview'])
+
+        assert_parent_repair_session_count(self.cluster.nodelist(), 0)
+
     @pytest.mark.no_vnodes
     def test_preview(self):
         """ Test that preview correctly detects out of sync data """

repair_tests/repair_test.py

@@ -16,6 +16,7 @@
 from dtest import FlakyRetryPolicy, Tester, create_ks, create_cf
 from tools.data import insert_c1c2, query_c1c2
 from tools.jmxutils import JolokiaAgent, make_mbean
+from repair_tests.incremental_repair_test import assert_parent_repair_session_count
 
 since = pytest.mark.since
 logger = logging.getLogger(__name__)
@@ -160,6 +161,16 @@ def _repair_and_verify(self, sequential=True):
 
 class TestRepair(BaseRepairTest):
 
+    @since('4.0')
+    def test_parent_repair_session_cleanup(self):
+        """
+        Calls range_tombstone_digest with a sequential repair and verifies if
+        all ParentRepairSession objects are cleaned
+        @jira_ticket CASSANDRA-16446
+        """
+        self._range_tombstone_digest(sequential=True)
+        assert_parent_repair_session_count(self.cluster.nodes.values(), 0)
+
     @since('2.2.1', max_version='4')
     def test_no_anticompaction_after_dclocal_repair(self):
         """