Merge a633036 into 1fa5b08

Author: kardymonds

ydb/core/fq/libs/checkpoint_storage/ydb_checkpoint_storage.cpp

@@ -805,10 +805,16 @@ TFuture<ICheckpointStorage::TCreateCheckpointResult> TCheckpointStorage::CreateC
 
     return future.Apply(
         [checkpointContext](const TFuture<NYdb::TStatus>& future) {
-            if (NYql::TIssues issues = StatusToIssues(future.GetValue())) {
-                return TCreateCheckpointResult(TString(), std::move(issues));
-            } else {
-                return TCreateCheckpointResult(checkpointContext->CheckpointGraphDescriptionContext->GraphDescId, NYql::TIssues());
+            try {    
+                if (NYql::TIssues issues = StatusToIssues(future.GetValue())) {
+                    return TCreateCheckpointResult(TString(), std::move(issues));
+                } else {
+                    return TCreateCheckpointResult(checkpointContext->CheckpointGraphDescriptionContext->GraphDescId, NYql::TIssues());
+                }
+            } catch (...) {
+                TIssues issues;
+                issues.AddIssue(CurrentExceptionMessage());
+                return TCreateCheckpointResult(TString(), issues);
             }
         });
 }
@@ -1099,7 +1105,13 @@ TFuture<ICheckpointStorage::TGetTotalCheckpointsStateSizeResult> TCheckpointStor
         });
     return future.Apply(
         [result](const TFuture<TStatus>& status) {
-          return std::make_pair(std::move(result->Size), std::move(status.GetValue().GetIssues()));
+           try {    
+               return std::make_pair(std::move(result->Size), std::move(status.GetValue().GetIssues()));
+            } catch (...) {
+                TIssues issues;
+                issues.AddIssue(CurrentExceptionMessage());
+                return ICheckpointStorage::TGetTotalCheckpointsStateSizeResult(0, issues);
+            }
         });
 }
 

ydb/core/fq/libs/checkpoint_storage/ydb_state_storage.cpp

@@ -115,6 +115,7 @@ TFuture<TStatus> ProcessState(
                 processed[taskIndex] = true;
             }
             LoadState(context->States[taskIndex], *parser.ColumnParser("blob").GetOptionalString());
+           // throw std::runtime_error("ddddd");
         }
     } else {
         errorMessage << "Not all states exist in database";
@@ -286,6 +287,7 @@ TFuture<IStateStorage::TGetStateResult> TStateStorage::GetState(
         return MakeFuture<IStateStorage::TGetStateResult>(result);
     }
 
+    
     auto context = MakeIntrusive<TContext>(
         YdbConnection->TablePathPrefix,
         taskIds,
@@ -366,11 +368,18 @@ TFuture<IStateStorage::TCountStatesResult> TStateStorage::CountStates(
         [context] (const TFuture<TStatus>& future) {
             TCountStatesResult countResult;
             countResult.first = context->Count;
-            const auto& status = future.GetValue();
-            if (!status.IsSuccess()) {
-                countResult.second = status.GetIssues();
+            try {    
+                const auto& status = future.GetValue();
+                if (!status.IsSuccess()) {
+                    countResult.second = status.GetIssues();
+                }
+            } catch (...) {
+                TIssues issues;
+                issues.AddIssue(CurrentExceptionMessage());
+                countResult.second = issues;
             }
             return countResult;
+           
         });
 }
 TExecDataQuerySettings TStateStorage::DefaultExecDataQuerySettings() {

ydb/core/fq/libs/ydb/ut/ya.make

@@ -0,0 +1,12 @@
+UNITTEST_FOR(ydb/core/fq/libs/ydb)
+
+SRCS(
+    ydb_ut.cpp
+)
+
+PEERDIR(
+    ydb/core/fq/libs/ydb
+)
+
+END()
+

ydb/core/fq/libs/ydb/ut/ydb_ut.cpp

@@ -0,0 +1,23 @@
+#include <ydb/core/fq/libs/ydb/ydb.h>
+
+#include <library/cpp/testing/unittest/registar.h>
+
+namespace NFq {
+
+Y_UNIT_TEST_SUITE(TFqYdbTest) {
+
+    Y_UNIT_TEST(ShouldStatusToIssuesProcessExceptions)
+    {
+        auto promise = NThreading::NewPromise<NYdb::TStatus>();
+        auto future = promise.GetFuture();
+        TString text("Test exception");
+        promise.SetException(text);
+        NThreading::TFuture<NYql::TIssues> future2 = NFq::StatusToIssues(future);
+
+        NYql::TIssues issues = future2.GetValueSync();
+        UNIT_ASSERT(issues.Size() == 1);
+        UNIT_ASSERT(issues.ToString().Contains(text));
+    }
+};
+
+} // namespace NFq

ydb/core/fq/libs/ydb/ya.make

@@ -24,3 +24,7 @@ GENERATE_ENUM_SERIALIZATION(ydb.h)
 YQL_LAST_ABI_VERSION()
 
 END()
+
+RECURSE_FOR_TESTS(
+    ut
+)

ydb/core/fq/libs/ydb/ydb.cpp

@@ -210,7 +210,13 @@ NYql::TIssues StatusToIssues(const NYdb::TStatus& status) {
 TFuture<TIssues> StatusToIssues(const TFuture<TStatus>& future) {
     return future.Apply(
         [] (const TFuture<TStatus>& future) {
-            return StatusToIssues(future.GetValue());
+            try {
+                return StatusToIssues(future.GetValue());
+            } catch (...) {
+                TIssues issues;
+                issues.AddIssue(CurrentExceptionMessage());
+                return issues;
+            }
     });
 }