fix for WriteTimeLag KIKIMR-14858

ref:afebc18e9af3957392063d2d752b11afc026047d

4 files changed, 55 insertions, 10 deletions

diff --git a/ydb/core/persqueue/partition.cpp b/ydb/core/persqueue/partition.cpp
index d1eecea1323..ae0e5b7a922 100644
--- a/ydb/core/persqueue/partition.cpp
+++ b/ydb/core/persqueue/partition.cpp
@@ -428,6 +428,9 @@ void TPartition::FillReadFromTimestamps(const NKikimrPQ::TPQTabletConfig& config
             userInfo.UserActs.push_back(event.Release());
             userInfo.Session = "";
             userInfo.Offset = 0;
+            if (userInfo.Important) {
+                userInfo.Offset = StartOffset;
+            }
             userInfo.Step = userInfo.Generation = 0;
         }
         hasReadRule.erase(consumer);
@@ -2893,7 +2896,7 @@ void TPartition::ReadTimestampForOffset(const TString& user, TUserInfo& userInfo
     );
 
     if (ReadingTimestamp) {
-        UpdateUserInfoTimestamp.push_back(user);
+        UpdateUserInfoTimestamp.push_back(std::make_pair(user, userInfo.ReadRuleGeneration));
         return;
     }
     if (userInfo.Offset < (i64)StartOffset) {
@@ -2927,7 +2930,7 @@ void TPartition::ReadTimestampForOffset(const TString& user, TUserInfo& userInfo
     ReadingForUserReadRuleGeneration = userInfo.ReadRuleGeneration;
 
     for (const auto& user : UpdateUserInfoTimestamp) {
-        Y_VERIFY(user != ReadingForUser);
+        Y_VERIFY(user.first != ReadingForUser || user.second != ReadingForUserReadRuleGeneration);
     }
 
     LOG_DEBUG_S(
@@ -2961,6 +2964,12 @@ void TPartition::Handle(TEvPQ::TEvProxyResponse::TPtr& ev, const TActorContext&
     ReadingTimestamp = false;
     auto userInfo = UsersInfoStorage.GetIfExists(ReadingForUser);
     if (!userInfo || userInfo->ReadRuleGeneration != ReadingForUserReadRuleGeneration) {
+        LOG_INFO_S(
+            ctx, NKikimrServices::PERSQUEUE,
+            "Topic '" << TopicConverter->GetClientsideName() << "' partition " << Partition
+                << " user " << ReadingForUser << " readTimeStamp for other generation or no client info at all"
+    );
+
         ProcessTimestampRead(ctx);
         return;
     }
@@ -2996,7 +3005,7 @@ void TPartition::Handle(TEvPQ::TEvProxyResponse::TPtr& ev, const TActorContext&
                 userInfo->ReadCreateTimestamp = userInfo->CreateTimestamp;
             }
         } else {
-            UpdateUserInfoTimestamp.push_back(ReadingForUser);
+            UpdateUserInfoTimestamp.push_back(std::make_pair(ReadingForUser, ReadingForUserReadRuleGeneration));
             userInfo->ReadScheduled = true;
         }
         Counters.Cumulative()[COUNTER_PQ_WRITE_TIMESTAMP_ERROR].Increment(1);
@@ -3010,10 +3019,11 @@ void TPartition::ProcessTimestampRead(const TActorContext& ctx) {
     ReadingForOffset = 0;
     ReadingForUserReadRuleGeneration = 0;
     while (!ReadingTimestamp && !UpdateUserInfoTimestamp.empty()) {
-        TString user = UpdateUserInfoTimestamp.front();
+        TString user = UpdateUserInfoTimestamp.front().first;
+        ui64 readRuleGeneration = UpdateUserInfoTimestamp.front().second;
         UpdateUserInfoTimestamp.pop_front();
         auto userInfo = UsersInfoStorage.GetIfExists(user);
-        if (!userInfo || !userInfo->ReadScheduled)
+        if (!userInfo || !userInfo->ReadScheduled || userInfo->ReadRuleGeneration != readRuleGeneration)
             continue;
         userInfo->ReadScheduled = false;
         if (userInfo->Offset == (i64)EndOffset)
@@ -3034,7 +3044,6 @@ void TPartition::Handle(TEvPQ::TEvError::TPtr& ev, const TActorContext& ctx)
         return;
     }
     Y_VERIFY(userInfo->ReadScheduled);
-    userInfo->ReadScheduled = false;
     Y_VERIFY(ReadingForUser != "");
 
     LOG_ERROR_S(
@@ -3043,8 +3052,7 @@ void TPartition::Handle(TEvPQ::TEvError::TPtr& ev, const TActorContext& ctx)
                 << " user " << ReadingForUser << " readTimeStamp error: " << ev->Get()->Error
     );
 
-    UpdateUserInfoTimestamp.push_back(ReadingForUser);
-    userInfo->ReadScheduled = true;
+    UpdateUserInfoTimestamp.push_back(std::make_pair(ReadingForUser, ReadingForUserReadRuleGeneration));
 
     ProcessTimestampRead(ctx);
 }
@@ -3540,6 +3548,10 @@ void TPartition::HandleSetOffsetResponse(NKikimrClient::TResponse& response, con
         userInfo->Session = "";
         userInfo->Generation = userInfo->Step = 0;
         userInfo->Offset = 0;
+        userInfo->ReadScheduled = false;
+        if (userInfo->Important) {
+            userInfo->Offset = StartOffset;
+        }
     } else {
         if (setSession || dropSession) {
             offset = userInfo->Offset;
diff --git a/ydb/core/persqueue/partition.h b/ydb/core/persqueue/partition.h
index 7c1ff5465ab..0b505416bad 100644
--- a/ydb/core/persqueue/partition.h
+++ b/ydb/core/persqueue/partition.h
@@ -570,7 +570,7 @@ private:
 
     TUsersInfoStorage UsersInfoStorage;
 
-    std::deque<TString> UpdateUserInfoTimestamp;
+    std::deque<std::pair<TString, ui64>> UpdateUserInfoTimestamp;
     bool ReadingTimestamp;
     TString ReadingForUser;
     ui64 ReadingForUserReadRuleGeneration;
diff --git a/ydb/core/persqueue/pq_impl.cpp b/ydb/core/persqueue/pq_impl.cpp
index 04ed308e423..299ab1afebc 100644
--- a/ydb/core/persqueue/pq_impl.cpp
+++ b/ydb/core/persqueue/pq_impl.cpp
@@ -312,7 +312,7 @@ public:
 
     bool HandleError(TEvPQ::TEvError *ev, const TActorContext& ctx)
     {
-        LOG_WARN_S(ctx, NKikimrServices::PERSQUEUE,
+        LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE,
                    "Answer error topic: '" << TopicName << "'" <<
                    " partition: " << Partition <<
                    " messageNo: " << MessageNo <<
diff --git a/ydb/core/persqueue/pq_ut.cpp b/ydb/core/persqueue/pq_ut.cpp
index 9704d5819e5..c419ac64800 100644
--- a/ydb/core/persqueue/pq_ut.cpp
+++ b/ydb/core/persqueue/pq_ut.cpp
@@ -2067,6 +2067,39 @@ Y_UNIT_TEST(TestWriteTimeStampEstimate) {
 
 }
 
+
+
+Y_UNIT_TEST(TestWriteTimeLag) {
+    TTestContext tc;
+    TFinalizer finalizer(tc);
+    tc.Prepare();
+
+    tc.Runtime->SetScheduledLimit(150);
+    tc.Runtime->SetDispatchTimeout(TDuration::Seconds(1));
+    tc.Runtime->SetLogPriority(NKikimrServices::PERSQUEUE, NLog::PRI_DEBUG);
+
+    PQTabletPrepare(20000000, 1_TB, 0, {{"aaa", false}}, tc);
+
+    TVector<std::pair<ui64, TString>> data{{1,TString(1024*1024, 'a')}};
+    for (ui32 i = 0; i < 20; ++i) {
+        CmdWrite(0, TStringBuilder() << "sourceid" << i, data, tc);
+    }
+
+    // After restart all caches are empty.
+    RestartTablet(tc);
+
+    PQTabletPrepare(20000000, 1_TB, 0, {{"aaa", false}, {"important", true}, {"another", true}}, tc);
+    PQTabletPrepare(20000000, 1_TB, 0, {{"aaa", false}, {"another1", true}, {"important", true}}, tc);
+    PQTabletPrepare(20000000, 1_TB, 0, {{"aaa", false}, {"another1", true}, {"important", true}, {"another", false}}, tc);
+
+    CmdGetOffset(0, "important", 12, tc, -1, 0);
+
+    CmdGetOffset(0, "another1", 12, tc, -1, 0);
+    CmdGetOffset(0, "another", 0, tc, -1, 0);
+    CmdGetOffset(0, "aaa", 0, tc, -1, 0);
+}
+
+
 void CheckEventSequence(TTestContext& tc, std::function<void()> scenario, std::deque<ui32> expectedEvents) {
     tc.Runtime->SetObserverFunc([&expectedEvents](TTestActorRuntimeBase&, TAutoPtr<IEventHandle>& ev) {
         if (!expectedEvents.empty() && ev->Type == expectedEvents.front()) {