Fixed topic tests

Author: Gazizonoki

src/client/query/client.cpp

@@ -858,8 +858,8 @@ class TTransaction::TImpl : public std::enable_shared_from_this<TImpl> {
 
 private:
     bool ChangesAreAccepted = true; // haven't called Commit or Rollback yet
-    std::vector<TPrecommitTransactionCallback> PrecommitCallbacks;
-    std::vector<TOnFailureTransactionCallback> OnFailureCallbacks;
+    mutable std::vector<TPrecommitTransactionCallback> PrecommitCallbacks;
+    mutable std::vector<TOnFailureTransactionCallback> OnFailureCallbacks;
 
     std::mutex PrecommitCallbacksMutex;
     std::mutex OnFailureCallbacksMutex;

src/client/topic/impl/read_session_impl.ipp

@@ -2183,6 +2183,9 @@ void TSingleClusterReadSessionImpl<UseMigrationProtocol>::TrySubscribeOnTransact
             return;
         }
 
+        txInfo->IsActive = true;
+        txInfo->Subscribed = true;
+
         auto callback = [cbContext = this->SelfContext, txId, txInfo, consumer = Settings.ConsumerName_, client]() {
             std::vector<TTopicOffsets> offsets;
 
@@ -2205,9 +2208,6 @@ void TSingleClusterReadSessionImpl<UseMigrationProtocol>::TrySubscribeOnTransact
         };
 
         tx.AddPrecommitCallback(std::move(callback));
-
-        txInfo->IsActive = true;
-        txInfo->Subscribed = true;
     }
 }
 

src/client/topic/impl/write_session_impl.cpp

@@ -551,34 +551,34 @@ void TWriteSessionImpl::TrySubscribeOnTransactionCommit(TTransactionBase* tx)
             return;
         }
 
+        txInfo->AllAcksReceived = NThreading::NewPromise<TStatus>();
         txInfo->IsActive = true;
         txInfo->Subscribed = true;
-        txInfo->AllAcksReceived = NThreading::NewPromise<TStatus>();
-    }
-
-    auto callback = [cbContext = this->SelfContext, txId, txInfo]() {
-        with_lock(txInfo->Lock) {
-            Y_ABORT_UNLESS(!txInfo->CommitCalled);
 
-            txInfo->CommitCalled = true;
-
-            if (txInfo->WriteCount == txInfo->AckCount) {
-                txInfo->AllAcksReceived.SetValue(MakeCommitTransactionSuccess());
-                if (auto self = cbContext->LockShared()) {
-                    self->DeleteTx(txId);
+        auto callback = [cbContext = this->SelfContext, txId, txInfo]() {
+            with_lock(txInfo->Lock) {
+                Y_ABORT_UNLESS(!txInfo->CommitCalled);
+    
+                txInfo->CommitCalled = true;
+    
+                if (txInfo->WriteCount == txInfo->AckCount) {
+                    txInfo->AllAcksReceived.SetValue(MakeCommitTransactionSuccess());
+                    if (auto self = cbContext->LockShared()) {
+                        self->DeleteTx(txId);
+                    }
+                    return txInfo->AllAcksReceived.GetFuture();
+                }
+    
+                if (txInfo->IsActive) {
+                    return txInfo->AllAcksReceived.GetFuture();
                 }
-                return txInfo->AllAcksReceived.GetFuture();
-            }
-
-            if (txInfo->IsActive) {
-                return txInfo->AllAcksReceived.GetFuture();
             }
-        }
-
-        return NThreading::MakeFuture(MakeSessionExpiredError());
-    };
-
-    tx->AddPrecommitCallback(std::move(callback));
+    
+            return NThreading::MakeFuture(MakeSessionExpiredError());
+        };
+    
+        tx->AddPrecommitCallback(std::move(callback));
+    }
 }
 
 void TWriteSessionImpl::TrySignalAllAcksReceived(ui64 seqNo)

tests/integration/topic/topic_to_table.cpp

@@ -1335,9 +1335,6 @@ void TxUsage::TestWriteToTopic10()
 
 void TxUsage::TestWriteToTopic26()
 {
-    // TODO(brgayazov): fix test
-    GTEST_SKIP() << "Test is flaky";
-
     //
     // the test verifies a transaction in which data is read from a partition of one topic and written to
     // another partition of this topic
@@ -1723,9 +1720,6 @@ TEST_F(TxUsageQuery, TEST_NAME(WriteToTopic_Demo_17))
 
 void TxUsage::TestWriteToTopic25()
 {
-    // TODO(brgayazov): fix test
-    GTEST_SKIP() << "Test is flaky";
-
     //
     // the test verifies a transaction in which data is read from one topic and written to another
     //