Merge d92de94 into 52b2390

Author: Alek5andr-Kotov

ydb/core/persqueue/partition.cpp

@@ -2310,7 +2310,9 @@ void TPartition::CommitWriteOperations(TTransaction& t)
 
     if (!t.WriteInfo->BlobsFromHead.empty()) {
         auto& first = t.WriteInfo->BlobsFromHead.front();
-        NewHead.PartNo = first.GetPartNo();
+        // In one operation, a partition can write blocks of several transactions. Some of them can be broken down
+        // into parts. We need to take this division into account.
+        NewHead.PartNo += first.GetPartNo();
 
         Parameters->HeadCleared = Parameters->HeadCleared || !t.WriteInfo->BodyKeys.empty();
 

ydb/public/sdk/cpp/src/client/topic/ut/topic_to_table_ut.cpp

@@ -2910,6 +2910,60 @@ Y_UNIT_TEST_F(Sinks_Olap_WriteToTopicAndTable_3, TFixtureSinks)
 
     CheckTabletKeys("topic_A");
 }
+
+Y_UNIT_TEST_F(Write_Random_Sized_Messages_In_Wide_Transactions, TFixture)
+{
+    // The test verifies the simultaneous execution of several transactions. There is a topic
+    // with PARTITIONS_COUNT partitions. In each transaction, the test writes to all the partitions.
+    // The size of the messages is random. Such that both large blobs in the body and small ones in
+    // the head of the partition are obtained. Message sizes are multiples of 500 KB. This way we
+    // will make sure that when committing transactions, the division into blocks is taken into account.
+
+    const size_t PARTITIONS_COUNT = 20;
+    const size_t TXS_COUNT = 100;
+
+    CreateTopic("topic_A", TEST_CONSUMER, PARTITIONS_COUNT);
+
+    TVector<NTable::TSession> sessions;
+    TVector<NTable::TTransaction> transactions;
+
+    // We open TXS_COUNT transactions and write messages to the topic.
+    for (size_t i = 0; i < TXS_COUNT; ++i) {
+        sessions.push_back(CreateTableSession());
+        auto& session = sessions.back();
+
+        transactions.push_back(BeginTx(session));
+        auto& tx = transactions.back();
+
+        for (size_t j = 0; j < PARTITIONS_COUNT; ++j) {
+            TString sourceId = TEST_MESSAGE_GROUP_ID;
+            sourceId += "_";
+            sourceId += ToString(i);
+            sourceId += "_";
+            sourceId += ToString(j);
+
+            size_t count = RandomNumber<size_t>(20) + 3;
+            WriteToTopic("topic_A", sourceId, TString(512 * 1000 * count, 'x'), &tx, j);
+
+            WaitForAcks("topic_A", sourceId);
+        }
+    }
+
+    // We are doing an asynchronous commit of transactions. They will be executed simultaneously.
+    TVector<NTable::TAsyncCommitTransactionResult> futures;
+
+    for (size_t i = 0; i < TXS_COUNT; ++i) {
+        futures.push_back(transactions[i].Commit());
+    }
+
+    // All transactions must be completed successfully.
+    for (size_t i = 0; i < TXS_COUNT; ++i) {
+        futures[i].Wait();
+        const auto& result = futures[i].GetValueSync();
+        UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+    }
+}
+
 }
 
 }