Fix ThrottlingException raising in YMQ when doing GetQueueUrl with drained queue list budget (#3600)

Author: siarheivesialou

ydb/core/ymq/actor/action.h

@@ -630,22 +630,23 @@ class TActionActor
             }
         }
 
-        if (ev->Get()->Throttled) {
-            MakeError(MutableErrorDesc(), NErrors::THROTTLING_EXCEPTION, "Too many requests for nonexistent queue.");
-            SendReplyAndDie();
-            return;
-        }
-
         if (ev->Get()->Fail) {
             MakeError(MutableErrorDesc(), NErrors::INTERNAL_FAILURE, "Failed to get configuration.");
             SendReplyAndDie();
             return;
         }
 
-        if (TDerived::NeedExistingQueue() && !ev->Get()->QueueExists) {
-            MakeError(MutableErrorDesc(), NErrors::NON_EXISTENT_QUEUE);
-            SendReplyAndDie();
-            return;
+        if (TDerived::NeedExistingQueue()) {
+            if (ev->Get()->Throttled) {
+                MakeError(MutableErrorDesc(), NErrors::THROTTLING_EXCEPTION, "Too many requests for nonexistent queue.");
+                SendReplyAndDie();
+                return;
+            }
+            if (!ev->Get()->QueueExists) {
+                MakeError(MutableErrorDesc(), NErrors::NON_EXISTENT_QUEUE);
+                SendReplyAndDie();
+                return;
+            }
         }
 
         bool isACLProtectedAccount = Cfg().GetForceAccessControl();

ydb/tests/functional/sqs/cloud/test_yandex_cloud_mode.py

@@ -838,3 +838,50 @@ def test_cloud_double_create_queue(self, is_fifo, tables_format):
         time.sleep(1)
         queue_url2 = self._sqs_api.create_queue(self.queue_name, is_fifo=is_fifo)
         assert queue_url1 == queue_url2, f'{queue_url1} vs {queue_url2}'
+
+    @pytest.mark.parametrize(**TABLES_FORMAT_PARAMS)
+    @pytest.mark.parametrize(**IS_FIFO_PARAMS)
+    def test_not_throttling_with_custom_queue_name(self, is_fifo, tables_format):
+        self._init_with_params(is_fifo, tables_format)
+
+        self._sqs_api = self._create_api_for_user(
+            user_name='ignored',
+            raise_on_error=True,
+            force_private=True,
+            iam_token=self.iam_token,
+            folder_id=self.folder_id,
+        )
+
+        custom_queue_name = 'MyCustomQueue'
+        queue_url = self._sqs_api.create_queue(
+            queue_name=self.queue_name,
+            private_api=True,
+            custom_name=custom_queue_name,
+        )
+
+        nonexistent_queue_url = queue_url.replace(self.queue_name, self.queue_name + '_nonex')
+
+        def get_attributes_of_nonexistent_queue():
+            self._sqs_api.get_queue_attributes(nonexistent_queue_url)
+
+        # Draining budget
+        for _ in range(16):
+            try:
+                get_attributes_of_nonexistent_queue()
+            except Exception:
+                pass
+
+        # Check that there is no more budget
+        assert_that(
+            get_attributes_of_nonexistent_queue,
+            raises(
+                RuntimeError,
+                pattern=".*<Code>ThrottlingException</Code>.*"
+            )
+        )
+
+        # Check that getting queue url with custom name still works
+        assert_that(
+            lambda: self._sqs_api.get_queue_url(custom_queue_name),
+            not_(raises(RuntimeError))
+        )