KIKIMR-15009 Fix race on increment of counter in TMeteringSink

Fix race

ref:962a480c752f8b8e6ba4ec2f117dd86951518024

2 files changed, 7 insertions, 6 deletions

diff --git a/ydb/core/persqueue/metering_sink.cpp b/ydb/core/persqueue/metering_sink.cpp
index 0726ff94c7..8019405dfd 100644
--- a/ydb/core/persqueue/metering_sink.cpp
+++ b/ydb/core/persqueue/metering_sink.cpp
@@ -6,7 +6,7 @@
 
 namespace NKikimr::NPQ {
 
-ui32 TMeteringSink::MeteringCounter_{0};
+std::atomic<ui64> TMeteringSink::MeteringCounter_{0};
 
 bool TMeteringSink::Create(TInstant now, const TMeteringSink::TParameters& p,
                            const TSet<EMeteringJson>& whichToFlush,
@@ -68,6 +68,7 @@ TString TMeteringSink::GetMeteringJson(const TString& metricBillingId, const TSt
                                       const THashMap<TString, ui64>& tags,
                                       const TString& quantityUnit, ui64 quantity,
                                       TInstant start, TInstant end, TInstant now) {
+    MeteringCounter_.fetch_add(1);
     TStringStream output;
     NJson::TJsonWriter writer(&output, false);
 
@@ -80,7 +81,7 @@ TString TMeteringSink::GetMeteringJson(const TString& metricBillingId, const TSt
                  "-" << Parameters_.YdbDatabaseId <<
                  "-" << Parameters_.TabletId <<
                  "-" << start.MilliSeconds() <<
-                 "-" << (++MeteringCounter_));
+                 "-" << MeteringCounter_.load());
     writer.Write("schema", schemeName);
 
     writer.OpenMap("tags");
@@ -239,8 +240,8 @@ void TMeteringSink::Flush(TInstant now, bool force) {
     }
 }
 
-ui32 TMeteringSink::GetMeteringCounter() const {
-    return MeteringCounter_;
+ui64 TMeteringSink::GetMeteringCounter() const {
+    return MeteringCounter_.load();
 }
 
 bool TMeteringSink::IsTimeToFlush(TInstant now, TInstant last) const {
diff --git a/ydb/core/persqueue/metering_sink.h b/ydb/core/persqueue/metering_sink.h
index d836a50e8e..cdb04d3c59 100644
--- a/ydb/core/persqueue/metering_sink.h
+++ b/ydb/core/persqueue/metering_sink.h
@@ -40,7 +40,7 @@ public:
     TString GetMeteringJson(const TString& metricBillingId, const TString& schemeName,
                             const THashMap<TString, ui64>& tags, const TString& quantityUnit, ui64 quantity,
                             TInstant start, TInstant end, TInstant now);
-    ui32 GetMeteringCounter() const;
+    ui64 GetMeteringCounter() const;
 
 private:
     TParameters Parameters_{};
@@ -50,7 +50,7 @@ private:
     TMap<EMeteringJson, TInstant> LastFlush_;
     std::function<void(TString)> FlushFunction_;
 
-    static ui32 MeteringCounter_;
+    static std::atomic<ui64> MeteringCounter_;
 
 private:
     void Flush(TInstant now, bool force);