Report PQ SDK Errors to user

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

@@ -491,8 +491,9 @@ class TDqPqReadActor : public NActors::TActor<TDqPqReadActor>, public IDqCompute
         }
 
         void operator()(NYdb::NTopic::TSessionClosedEvent& ev) {
-            ythrow yexception() << "SessionId: " << Self.GetSessionId() << " Read session to topic \"" << Self.SourceParams.GetTopicPath()
-                << "\" was closed: " << ev.DebugString();
+            const auto& LogPrefix = Self.LogPrefix;
+            SRC_LOG_E("SessionId: " << Self.GetSessionId() << " Read session to topic \"" << Self.SourceParams.GetTopicPath() << "\" was closed: " << ev.DebugString());
+            Self.Send(Self.ComputeActorId, new TEvAsyncInputError(Self.InputIndex, ev.GetIssues(), NYql::NDqProto::StatusIds::BAD_REQUEST));
         }
 
         void operator()(NYdb::NTopic::TReadSessionEvent::TCommitOffsetAcknowledgementEvent&) { }

ydb/tests/fq/yds/test_invalid_consumer.py

@@ -0,0 +1,36 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import logging
+import os
+import pytest
+import time
+
+from ydb.tests.tools.datastreams_helpers.test_yds_base import TestYdsBase
+
+import ydb.public.api.protos.draft.fq_pb2 as fq
+from ydb.tests.tools.fq_runner.kikimr_utils import yq_v1
+
+
+class TestConsumer(TestYdsBase):
+    @yq_v1
+    def test_invalid(self, kikimr, client):
+        self.init_topics("invalid_consumer", create_output=False)
+
+        sql = R'''
+            PRAGMA pq.Consumer="InvalidConsumerName";
+            SELECT * FROM yds.`{input_topic}` LIMIT 1;
+            ''' \
+            .format(
+            input_topic=self.input_topic,
+        )
+
+        client.create_yds_connection("yds", os.getenv("YDB_DATABASE"), os.getenv("YDB_ENDPOINT"))
+        query_id = client.create_query("simple", sql, type=fq.QueryContent.QueryType.STREAMING).result.query_id
+        client.wait_query_status(query_id, fq.QueryMeta.FAILED)
+
+        describe_result = client.describe_query(query_id).result
+        logging.debug("Describe result: {}".format(describe_result))
+        describe_string = "{}".format(describe_result)
+        assert "Query failed with code BAD_REQUEST" in describe_string
+        assert "no read rule provided for consumer" in describe_string

ydb/tests/fq/yds/ya.make

@@ -28,6 +28,7 @@ TEST_SRCS(
     test_cpu_quota.py
     test_delete_read_rules_after_abort_by_system.py
     test_eval.py
+    test_invalid_consumer.py
     test_kill_pq_bill.py
     test_mem_alloc.py
     test_metrics_cleanup.py