Federation mode with tx counters test (#9523)

Author: FloatingCrowbar

ydb/core/persqueue/ut/partition_ut.cpp

@@ -352,7 +352,7 @@ TPartition* TPartitionFixture::CreatePartitionActor(const TPartitionId& id,
                         config.MeteringMode);
     Config.SetLocalDC(true);
 
-    NPersQueue::TTopicNamesConverterFactory factory(true, "/Root/PQ", "dc1");
+    NPersQueue::TTopicNamesConverterFactory factory(Ctx->Runtime->GetAppData(0).PQConfig.GetTopicsAreFirstClassCitizen(), "/Root/PQ", "dc1");
     TopicConverter = factory.MakeTopicConverter(Config);
     TActorId quoterId;
     if (Ctx->Runtime->GetAppData(0).PQConfig.GetQuotingConfig().GetEnableQuoting()) {
@@ -1323,6 +1323,8 @@ class TPartitionTxTestHelper : public TPartitionFixture {
     void ExpectNoBatchCompletion();
     void WaitBatchCompletion(ui64 userActsCount);
     void ResetBatchCompletion();
+
+    void NonConflictingActsBatchOkTest();
 };
 
 ui64 TPartitionTxTestHelper::MakeAndSendNormalOffsetCommit(ui64 client, ui64 offset) {
@@ -1369,6 +1371,11 @@ void TPartitionTxTestHelper::SendWriteInfoResponseImpl(const TActorId& supportiv
     auto iter = this->WriteInfoData.find(supportiveId);
     Y_ABORT_UNLESS(!iter.IsEnd());
     reply->SrcIdInfo = iter->second;
+    reply->BytesWrittenTotal = 1;
+    reply->BytesWrittenGrpc = 1;
+    reply->BytesWrittenUncompressed = 1;
+    reply->MessagesWrittenTotal = 1;
+    reply->MessagesWrittenGrpc = 1;
     SendEvent(reply, supportiveId, partitionId);
 }
 
@@ -2548,7 +2555,7 @@ Y_UNIT_TEST_F(DataTxCalcPredicateOrder, TPartitionTxTestHelper)
     WaitCommitDone(tx2);
 }
 
-Y_UNIT_TEST_F(NonConflictingActsBatchOk, TPartitionTxTestHelper) {
+void TPartitionTxTestHelper::NonConflictingActsBatchOkTest() {
     TTxBatchingTestParams params {.WriterSessions{"src3", "src4"}};
     Init(std::move(params));
     ResetBatchCompletion();
@@ -2591,6 +2598,14 @@ Y_UNIT_TEST_F(NonConflictingActsBatchOk, TPartitionTxTestHelper) {
     WaitImmediateTxComplete(immTx2, true);
     WaitCommitDone(tx3);
 }
+Y_UNIT_TEST_F(TestNonConflictingActsBatchOk, TPartitionTxTestHelper) {
+    NonConflictingActsBatchOkTest();
+}
+
+Y_UNIT_TEST_F(TestTxBatchInFederation, TPartitionTxTestHelper) {
+    Ctx->Runtime->GetAppData(0).PQConfig.SetTopicsAreFirstClassCitizen(false);
+    NonConflictingActsBatchOkTest();
+}
 
 Y_UNIT_TEST_F(ConflictingActsInSeveralBatches, TPartitionTxTestHelper) {
     TTxBatchingTestParams params {.WriterSessions{"src1", "src4"},.EndOffset=1};
@@ -3124,6 +3139,8 @@ Y_UNIT_TEST_F(TestBatchingWithProposeConfig, TPartitionTxTestHelper) {
     WaitImmediateTxComplete(immTx2, true);
 }
 
+
+
 Y_UNIT_TEST_F(GetUsedStorage, TPartitionFixture) {
     auto* actor = CreatePartition({
                     .Partition=TPartitionId{2, TWriteId{0, 10}, 100'001},