Initializing userInfo.AnyCommits

Author: nshestakov

ydb/core/persqueue/partition.cpp

@@ -3263,9 +3263,7 @@ void TPartition::EmulatePostProcessUserAct(const TEvPQ::TEvSetClientInfo& act,
         );
 
         userInfo.Offset = offset;
-        if (userInfo.Offset <= (i64)StartOffset) {
-            userInfo.AnyCommits = false;
-        }
+        userInfo.AnyCommits = userInfo.Offset > (i64)StartOffset;
 
         if (LastOffsetHasBeenCommited(userInfo)) {
             SendReadingFinished(user);

ydb/core/persqueue/partition_init.cpp

@@ -433,12 +433,12 @@ void TInitInfoRangeStep::Handle(TEvKeyValue::TEvResponse::TPtr &ev, const TActor
                 Y_ABORT_UNLESS(key);
                 RequestInfoRange(ctx, Partition()->Tablet, PartitionId(), *key);
             } else {
-                Done(ctx);
+                PostProcessing(ctx);
             }
             break;
         }
         case NKikimrProto::NODATA:
-            Done(ctx);
+            PostProcessing(ctx);
             break;
         case NKikimrProto::ERROR:
             PQ_LOG_ERROR("read topic error");
@@ -450,6 +450,16 @@ void TInitInfoRangeStep::Handle(TEvKeyValue::TEvResponse::TPtr &ev, const TActor
     };
 }
 
+void TInitInfoRangeStep::PostProcessing(const TActorContext& ctx) {
+    auto& usersInfoStorage = Partition()->UsersInfoStorage;
+    for (auto& [_, userInfo] : usersInfoStorage->GetAll()) {
+        userInfo.AnyCommits = userInfo.Offset > (i64)Partition()->StartOffset;
+    }
+
+    Done(ctx);
+}
+
+
 
 //
 // TInitDataRangeStep

ydb/core/persqueue/partition_init.h

@@ -140,6 +140,8 @@ class TInitInfoRangeStep: public TBaseKVStep {
 
     void Execute(const TActorContext& ctx) override;
     void Handle(TEvKeyValue::TEvResponse::TPtr& ev, const TActorContext& ctx) override;
+
+    void PostProcessing(const TActorContext& ctx);
 };
 
 class TInitDataRangeStep: public TBaseKVStep {