The consumer's generation number is not stored in the transaction (ydb-platform#9555)

Author: Alek5andr-Kotov

ydb/core/persqueue/pq_impl.cpp

@@ -715,6 +715,8 @@ void TPersQueue::ApplyNewConfig(const NKikimrPQ::TPQTabletConfig& newConfig,
 {
     Config = newConfig;
 
+    PQ_LOG_D("Apply new config " << Config.ShortDebugString());
+
     ui32 cacheSize = CACHE_SIZE;
     if (Config.HasCacheSize()) {
         cacheSize = Config.GetCacheSize();
@@ -1630,6 +1632,32 @@ void TPersQueue::CreateTopicConverter(const NKikimrPQ::TPQTabletConfig& config,
     Y_ABORT_UNLESS(topicConverter->IsValid(), "%s", topicConverter->GetReason().c_str());
 }
 
+void TPersQueue::UpdateReadRuleGenerations(NKikimrPQ::TPQTabletConfig& cfg) const
+{
+    Y_ABORT_UNLESS(cfg.HasVersion());
+    const int curConfigVersion = cfg.GetVersion();
+
+    // set rr generation for provided read rules
+    THashMap<TString, std::pair<ui64, ui64>> existed; // map name -> rrVersion, rrGeneration
+    for (const auto& c : Config.GetConsumers()) {
+        existed[c.GetName()] = std::make_pair(c.GetVersion(), c.GetGeneration());
+    }
+
+    for (auto& c : *cfg.MutableConsumers()) {
+        auto it = existed.find(c.GetName());
+        ui64 generation = 0;
+        if (it != existed.end() && it->second.first == c.GetVersion()) {
+            generation = it->second.second;
+        } else {
+            generation = curConfigVersion;
+        }
+        c.SetGeneration(generation);
+        if (ReadRuleCompatible()) {
+            cfg.AddReadRuleGenerations(generation);
+        }
+    }
+}
+
 void TPersQueue::ProcessUpdateConfigRequest(TAutoPtr<TEvPersQueue::TEvUpdateConfig> ev, const TActorId& sender, const TActorContext& ctx)
 {
     const auto& record = ev->GetRecord();
@@ -1642,7 +1670,7 @@ void TPersQueue::ProcessUpdateConfigRequest(TAutoPtr<TEvPersQueue::TEvUpdateConf
     NKikimrPQ::TPQTabletConfig cfg = record.GetTabletConfig();
 
     Y_ABORT_UNLESS(cfg.HasVersion());
-    int curConfigVersion = cfg.GetVersion();
+    const int curConfigVersion = cfg.GetVersion();
 
     if (curConfigVersion == oldConfigVersion) { //already applied
         LOG_INFO_S(ctx, NKikimrServices::PERSQUEUE, "Tablet " << TabletID()
@@ -1741,27 +1769,7 @@ void TPersQueue::ProcessUpdateConfigRequest(TAutoPtr<TEvPersQueue::TEvUpdateConf
 
     Migrate(cfg);
 
-    // set rr generation for provided read rules
-    {
-        THashMap<TString, std::pair<ui64, ui64>> existed; // map name -> rrVersion, rrGeneration
-        for (const auto& c : Config.GetConsumers()) {
-            existed[c.GetName()] = std::make_pair(c.GetVersion(), c.GetGeneration());
-        }
-
-        for (auto& c : *cfg.MutableConsumers()) {
-            auto it = existed.find(c.GetName());
-            ui64 generation = 0;
-            if (it != existed.end() && it->second.first == c.GetVersion()) {
-                generation = it->second.second;
-            } else {
-                generation = curConfigVersion;
-            }
-            c.SetGeneration(generation);
-            if (ReadRuleCompatible()) {
-                cfg.AddReadRuleGenerations(generation);
-            }
-        }
-    }
+    UpdateReadRuleGenerations(cfg);
 
     LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE, "Tablet " << TabletID()
                 << " Config update version " << cfg.GetVersion() << "(current " << Config.GetVersion() << ") received from actor " << sender
@@ -3727,6 +3735,10 @@ void TPersQueue::ProcessProposeTransactionQueue(const TActorContext& ctx)
             tx.OnProposeTransaction(event, GetAllowedStep(),
                                     TabletID());
 
+            if (tx.Kind == NKikimrPQ::TTransaction::KIND_CONFIG) {
+                UpdateReadRuleGenerations(tx.TabletConfig);
+            }
+
             if (tx.WriteId.Defined()) {
                 const TWriteId& writeId = *tx.WriteId;
                 Y_ABORT_UNLESS(TxWrites.contains(writeId),

ydb/core/persqueue/pq_impl.h

@@ -530,6 +530,8 @@ class TPersQueue : public NKeyValue::TKeyValueFlat {
 
     bool AllSupportivePartitionsHaveBeenDeleted(const TMaybe<TWriteId>& writeId) const;
     void DeleteWriteId(const TMaybe<TWriteId>& writeId);
+
+    void UpdateReadRuleGenerations(NKikimrPQ::TPQTabletConfig& cfg) const;
 };
 
 

ydb/public/sdk/cpp/client/ydb_topic/ut/topic_to_table_ut.cpp

@@ -39,6 +39,8 @@ class TFixture : public NUnitTest::TBaseFixture {
 
     void SetUp(NUnitTest::TTestContext&) override;
 
+    virtual bool GetEnablePQConfigTransactionsAtSchemeShard() const;
+
     NTable::TSession CreateTableSession();
     NTable::TTransaction BeginTx(NTable::TSession& session);
     void CommitTx(NTable::TTransaction& tx, EStatus status = EStatus::SUCCESS);
@@ -62,6 +64,8 @@ class TFixture : public NUnitTest::TBaseFixture {
                      std::optional<size_t> maxPartitionCount = std::nullopt);
     void DescribeTopic(const TString& path);
 
+    void AddConsumer(const TString& topic, const TVector<TString>& consumers);
+
     void WriteToTopicWithInvalidTxId(bool invalidTxId);
 
     TTopicWriteSessionPtr CreateTopicWriteSession(const TString& topicPath,
@@ -192,11 +196,18 @@ void TFixture::SetUp(NUnitTest::TTestContext&)
 {
     NKikimr::Tests::TServerSettings settings = TTopicSdkTestSetup::MakeServerSettings();
     settings.SetEnableTopicServiceTx(true);
+    settings.SetEnablePQConfigTransactionsAtSchemeShard(GetEnablePQConfigTransactionsAtSchemeShard());
+
     Setup = std::make_unique<TTopicSdkTestSetup>(TEST_CASE_NAME, settings);
 
     Driver = std::make_unique<TDriver>(Setup->MakeDriver());
 }
 
+bool TFixture::GetEnablePQConfigTransactionsAtSchemeShard() const
+{
+    return true;
+}
+
 NTable::TSession TFixture::CreateTableSession()
 {
     NTable::TTableClient client(GetDriver());
@@ -323,6 +334,20 @@ void TFixture::CreateTopic(const TString& path,
     Setup->CreateTopic(path, consumer, partitionCount, maxPartitionCount);
 }
 
+void TFixture::AddConsumer(const TString& path,
+                           const TVector<TString>& consumers)
+{
+    NTopic::TTopicClient client(GetDriver());
+    NTopic::TAlterTopicSettings settings;
+
+    for (const auto& consumer : consumers) {
+        settings.BeginAddConsumer(consumer);
+    }
+
+    auto result = client.AlterTopic(path, settings).GetValueSync();
+    UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
+}
+
 void TFixture::DescribeTopic(const TString& path)
 {
     Setup->DescribeTopic(path);