Merge 3adeb87 into b765769

ydb-platform · Feb 2, 2024 · 6cb6dd2 · 6cb6dd2
2 parents b765769 + 3adeb87
commit 6cb6dd2
Show file tree

Hide file tree

Showing 6 changed files with 74 additions and 9 deletions.
diff --git a/ydb/core/fq/libs/checkpoint_storage/ydb_checkpoint_storage.cpp b/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);
+            }
         });
 }
 

diff --git a/ydb/core/fq/libs/checkpoint_storage/ydb_state_storage.cpp b/ydb/core/fq/libs/checkpoint_storage/ydb_state_storage.cpp
@@ -286,6 +286,7 @@ TFuture<IStateStorage::TGetStateResult> TStateStorage::GetState(
         return MakeFuture<IStateStorage::TGetStateResult>(result);
     }
 
+
     auto context = MakeIntrusive<TContext>(
         YdbConnection->TablePathPrefix,
         taskIds,
@@ -366,11 +367,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() {

diff --git a/ydb/core/fq/libs/ydb/ut/ya.make b/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()
+
diff --git a/ydb/core/fq/libs/ydb/ut/ydb_ut.cpp b/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
diff --git a/ydb/core/fq/libs/ydb/ya.make b/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
+)
diff --git a/ydb/core/fq/libs/ydb/ydb.cpp b/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;
+            }
     });
 }
-Original file line number
+Diff line change
@@ Expand Up / @@ -24,3 +24,7 @@ GENERATE_ENUM_SERIALIZATION(ydb.h) @@
     YQL_LAST_ABI_VERSION()
     END()
+    RECURSE_FOR_TESTS(
+        ut
+    )