try to fix flaky tests

Author: kardymonds

ydb/library/yql/providers/pq/async_io/dq_pq_read_actor_base.cpp

@@ -36,6 +36,7 @@ void TDqPqReadActorBase::SaveState(const NDqProto::TCheckpoint& /*checkpoint*/,
         partitionState->SetCluster(cluster);
         partitionState->SetPartition(partition);
         partitionState->SetOffset(offset);
+        SRC_LOG_D("SessionId: " << GetSessionId() << " SaveState: partition " << partition << ", offset: " << offset);
     }
 
     stateProto.SetStartingMessageTimestampMs(StartingMessageTimestamp.MilliSeconds());

ydb/tests/fq/yds/test_recovery.py

@@ -14,6 +14,7 @@
 import ydb.tests.library.common.yatest_common as yatest_common
 from ydb.tests.tools.fq_runner.kikimr_runner import StreamingOverKikimr
 from ydb.tests.tools.fq_runner.kikimr_runner import StreamingOverKikimrConfig
+from ydb.tests.tools.fq_runner.kikimr_runner import TenantConfig
 from ydb.tests.tools.fq_runner.kikimr_utils import yq_v1
 from ydb.tests.tools.datastreams_helpers.test_yds_base import TestYdsBase
 from ydb.tests.tools.datastreams_helpers.control_plane import create_stream
@@ -33,7 +34,7 @@ def run_with_sleep(args):
 
 @pytest.fixture
 def kikimr():
-    kikimr_conf = StreamingOverKikimrConfig(node_count=8, cloud_mode=True)
+    kikimr_conf = StreamingOverKikimrConfig(node_count={"/cp": TenantConfig(1), "/compute": TenantConfig(2)}, cloud_mode=True)
     kikimr = StreamingOverKikimr(kikimr_conf)
     kikimr.start_mvp_mock_server()
     kikimr.start()
@@ -50,8 +51,8 @@ def setup_class(cls):
 
     @retry.retry_intrusive
     def get_graph_master_node_id(self, query_id):
-        for node_index in self.kikimr.control_plane.kikimr_cluster.nodes:
-            if self.kikimr.control_plane.get_task_count(node_index, query_id) > 0:
+        for node_index in self.kikimr.compute_plane.kikimr_cluster.nodes:
+            if self.kikimr.compute_plane.get_task_count(node_index, query_id) > 0:
                 return node_index
         assert False, "No active graphs found"
 
@@ -61,9 +62,9 @@ def dump_workers(self, worker_count, ca_count, wait_time=yatest_common.plain_or_
             wcs = 0
             ccs = 0
             list = []
-            for node_index in self.kikimr.control_plane.kikimr_cluster.nodes:
-                wc = self.kikimr.control_plane.get_worker_count(node_index)
-                cc = self.kikimr.control_plane.get_ca_count(node_index)
+            for node_index in self.kikimr.compute_plane.kikimr_cluster.nodes:
+                wc = self.kikimr.compute_plane.get_worker_count(node_index)
+                cc = self.kikimr.compute_plane.get_ca_count(node_index)
                 wcs += wc
                 ccs += cc
                 list.append([node_index, wc, cc])
@@ -81,8 +82,8 @@ def dump_workers(self, worker_count, ca_count, wait_time=yatest_common.plain_or_
 
     @yq_v1
     def test_delete(self, client, kikimr):
-        for node_index in kikimr.control_plane.kikimr_cluster.nodes:
-            kikimr.control_plane.wait_bootstrap(node_index)
+        kikimr.control_plane.wait_bootstrap()
+        kikimr.compute_plane.wait_bootstrap()
 
         self.kikimr = kikimr
         self.init_topics("recovery", partitions_count=2)
@@ -117,10 +118,9 @@ def test_program_state_recovery(self, client, kikimr):
         #             [  Bucket2  )           |(emited)
         #                   .<------------------------------------- restart
         #                         [  Bucket3  )           |(emited)
-        for node_index in kikimr.control_plane.kikimr_cluster.nodes:
-            kikimr.control_plane.wait_bootstrap(node_index)
-        for node_index in kikimr.control_plane.kikimr_cluster.nodes:
-            kikimr.control_plane.wait_discovery(node_index)
+        kikimr.control_plane.wait_bootstrap()
+        kikimr.compute_plane.wait_bootstrap()
+        kikimr.compute_plane.wait_discovery()
 
         self.kikimr = kikimr
         self.init_topics("program_state_recovery", partitions_count=1)
@@ -162,18 +162,18 @@ def test_program_state_recovery(self, client, kikimr):
 
         # restart node with CA
         node_to_restart = None
-        for node_index in kikimr.control_plane.kikimr_cluster.nodes:
-            wc = kikimr.control_plane.get_worker_count(node_index)
+        for node_index in kikimr.compute_plane.kikimr_cluster.nodes:
+            wc = kikimr.compute_plane.get_worker_count(node_index)
             if wc is not None:
                 if wc > 0 and node_index != master_node_index and node_to_restart is None:
                     node_to_restart = node_index
         assert node_to_restart is not None, "Can't find any task on non master node"
 
         logging.debug("Restart non-master node {}".format(node_to_restart))
 
-        kikimr.control_plane.kikimr_cluster.nodes[node_to_restart].stop()
-        kikimr.control_plane.kikimr_cluster.nodes[node_to_restart].start()
-        kikimr.control_plane.wait_bootstrap(node_to_restart)
+        kikimr.compute_plane.kikimr_cluster.nodes[node_to_restart].stop()
+        kikimr.compute_plane.kikimr_cluster.nodes[node_to_restart].start()
+        kikimr.compute_plane.wait_bootstrap(node_to_restart)
 
         self.write_stream([f'{{"time" = {i};}}' for i in range(116, 144, 2)])
 
@@ -198,10 +198,9 @@ def test_program_state_recovery(self, client, kikimr):
     #     ids=["not_master", "master"]
     # )
     def test_recovery(self, client, kikimr):
-        for node_index in kikimr.control_plane.kikimr_cluster.nodes:
-            kikimr.control_plane.wait_bootstrap(node_index)
-        for node_index in kikimr.control_plane.kikimr_cluster.nodes:
-            kikimr.control_plane.wait_discovery(node_index)
+        kikimr.control_plane.wait_bootstrap()
+        kikimr.compute_plane.wait_bootstrap()
+        kikimr.compute_plane.wait_discovery()
 
         self.init_topics("recovery", partitions_count=2)
 
@@ -242,18 +241,18 @@ def test_recovery(self, client, kikimr):
         self.dump_workers(2, 4)
 
         node_to_restart = None
-        for node_index in kikimr.control_plane.kikimr_cluster.nodes:
-            wc = kikimr.control_plane.get_worker_count(node_index)
+        for node_index in kikimr.compute_plane.kikimr_cluster.nodes:
+            wc = kikimr.compute_plane.get_worker_count(node_index)
             if wc is not None:
                 if wc > 0 and node_index != master_node_index and node_to_restart is None:
                     node_to_restart = node_index
         assert node_to_restart is not None, "Can't find any task on non master node"
 
         logging.debug("Restart non-master node {}".format(node_to_restart))
 
-        kikimr.control_plane.kikimr_cluster.nodes[node_to_restart].stop()
-        kikimr.control_plane.kikimr_cluster.nodes[node_to_restart].start()
-        kikimr.control_plane.wait_bootstrap(node_to_restart)
+        kikimr.compute_plane.kikimr_cluster.nodes[node_to_restart].stop()
+        kikimr.compute_plane.kikimr_cluster.nodes[node_to_restart].start()
+        kikimr.compute_plane.wait_bootstrap(node_to_restart)
 
         self.dump_workers(2, 4)
 
@@ -273,9 +272,9 @@ def test_recovery(self, client, kikimr):
 
         logging.debug("Restart Master node {}".format(master_node_index))
 
-        kikimr.control_plane.kikimr_cluster.nodes[master_node_index].stop()
-        kikimr.control_plane.kikimr_cluster.nodes[master_node_index].start()
-        kikimr.control_plane.wait_bootstrap(master_node_index)
+        kikimr.compute_plane.kikimr_cluster.nodes[master_node_index].stop()
+        kikimr.compute_plane.kikimr_cluster.nodes[master_node_index].start()
+        kikimr.compute_plane.wait_bootstrap(master_node_index)
         master_node_index = self.get_graph_master_node_id(query_id)
 
         logging.debug("New master node {}".format(master_node_index))
@@ -295,10 +294,10 @@ def test_recovery(self, client, kikimr):
                 d[n] = 1
         assert len(d) == 30
 
-        zero_checkpoints_metric = kikimr.control_plane.get_checkpoint_coordinator_metric(
+        zero_checkpoints_metric = kikimr.compute_plane.get_checkpoint_coordinator_metric(
             query_id, "StartedFromEmptyCheckpoint"
         )
-        restored_metric = kikimr.control_plane.get_checkpoint_coordinator_metric(
+        restored_metric = kikimr.compute_plane.get_checkpoint_coordinator_metric(
             query_id, "RestoredFromSavedCheckpoint"
         )
         assert restored_metric >= 1, "RestoredFromSavedCheckpoint: {}, StartedFromEmptyCheckpoint: {}".format(
@@ -420,10 +419,9 @@ def test_ic_disconnection(self, client):
 
     @yq_v1
     def test_program_state_recovery_error_if_no_states(self, client, kikimr):
-        for node_index in kikimr.control_plane.kikimr_cluster.nodes:
-            kikimr.control_plane.wait_bootstrap(node_index)
-        for node_index in kikimr.control_plane.kikimr_cluster.nodes:
-            kikimr.control_plane.wait_discovery(node_index)
+        kikimr.control_plane.wait_bootstrap()
+        kikimr.compute_plane.wait_bootstrap()
+        kikimr.compute_plane.wait_discovery()
         self.init_topics("error_if_no_states", partitions_count=1)
 
         sql = R'''
@@ -443,17 +441,16 @@ def test_program_state_recovery_error_if_no_states(self, client, kikimr):
             query_id, kikimr.compute_plane.get_completed_checkpoints(query_id) + 1
         )
 
-        for node_index in kikimr.control_plane.kikimr_cluster.nodes:
-            kikimr.control_plane.kikimr_cluster.nodes[node_index].stop()
+        for node_index in kikimr.compute_plane.kikimr_cluster.nodes:
+            kikimr.compute_plane.kikimr_cluster.nodes[node_index].stop()
 
         session = kikimr.driver.table_client.session().create()
         checkpoint_table_prefix = "/local/CheckpointCoordinatorStorage_" + kikimr.uuid + '/states'
         session.transaction().execute(f"DELETE FROM `{checkpoint_table_prefix}`", commit_tx=True)
 
-        for node_index in kikimr.control_plane.kikimr_cluster.nodes:
-            kikimr.control_plane.kikimr_cluster.nodes[node_index].start()
-        for node_index in kikimr.control_plane.kikimr_cluster.nodes:
-            kikimr.control_plane.wait_bootstrap(node_index)
+        for node_index in kikimr.compute_plane.kikimr_cluster.nodes:
+            kikimr.compute_plane.kikimr_cluster.nodes[node_index].start()
+        kikimr.compute_plane.wait_bootstrap()
 
         client.wait_query_status(query_id, fq.QueryMeta.FAILED)
         describe_result = client.describe_query(query_id).result

ydb/tests/fq/yds/test_recovery_mz.py

@@ -22,7 +22,7 @@
 @pytest.fixture
 def kikimr():
     kikimr_conf = StreamingOverKikimrConfig(
-        cloud_mode=True, node_count={"/cp": TenantConfig(1), "/compute": TenantConfig(8)}
+        cloud_mode=True, node_count={"/cp": TenantConfig(1), "/compute": TenantConfig(2)}
     )
     kikimr = StreamingOverKikimr(kikimr_conf)
     # control

ydb/tests/fq/yds/test_row_dispatcher.py

@@ -544,7 +544,7 @@ def test_stop_start_with_filter(self, kikimr, client):
         self.write_stream(data)
 
         kikimr.compute_plane.wait_completed_checkpoints(
-            query_id, kikimr.compute_plane.get_completed_checkpoints(query_id) + 2
+            query_id, kikimr.compute_plane.get_completed_checkpoints(query_id) + 10     # long sleep to send status from topic_session to read_actor
         )
         stop_yds_query(client, query_id)
         wait_actor_count(kikimr, "FQ_ROW_DISPATCHER_SESSION", 0)

ydb/tests/tools/datastreams_helpers/data_plane.py

@@ -14,7 +14,7 @@
 from ydb.public.api.protos.ydb_status_codes_pb2 import StatusIds
 
 
-READ_TOOL_TIMEOUT = yatest_common.plain_or_under_sanitizer(20, 300)
+READ_TOOL_TIMEOUT = yatest_common.plain_or_under_sanitizer(30, 300)
 
 
 def write_stream(path, data, partition_key=None):