Added max_committed_time_lag for DescribeConsumer (#16857)

Author: nshestakov

include/ydb-cpp-sdk/client/topic/control_plane.h

@@ -111,6 +111,7 @@ class TPartitionConsumerStats {
     const TInstant& GetLastReadTime() const;
     const TDuration& GetMaxReadTimeLag() const;
     const TDuration& GetMaxWriteTimeLag() const;
+    const TDuration& GetMaxCommittedTimeLag() const;
 
 private:
     uint64_t CommittedOffset_;
@@ -120,6 +121,7 @@ class TPartitionConsumerStats {
     TInstant LastReadTime_;
     TDuration MaxReadTimeLag_;
     TDuration MaxWriteTimeLag_;
+    TDuration MaxCommittedTimeLag_;
 };
 
 // Topic partition location

src/api/protos/ydb_topic.proto

@@ -810,6 +810,8 @@ message Consumer {
         google.protobuf.Duration max_read_time_lag = 2;
         // Maximum of differences between write timestamp and create timestamp for all messages, read during last minute.
         google.protobuf.Duration max_write_time_lag = 3;
+        // The difference between the write timestamp of the last commited message and the current time.
+        google.protobuf.Duration max_committed_time_lag = 5;
         // Bytes read statistics.
         MultipleWindowsStat bytes_read = 4;
     }
@@ -1214,6 +1216,8 @@ message DescribeConsumerResult {
         google.protobuf.Duration max_read_time_lag = 6;
         // Maximum of differences between write timestamp and create timestamp for all messages, read during last minute.
         google.protobuf.Duration max_write_time_lag = 7;
+        // The difference between the write timestamp of the last commited message and the current time.
+        google.protobuf.Duration max_committed_time_lag = 13;
 
         // How much bytes were read during several windows statistics from this partition.
         MultipleWindowsStat bytes_read = 8;

src/client/topic/impl/topic.cpp

@@ -376,6 +376,7 @@ TPartitionConsumerStats::TPartitionConsumerStats(const Ydb::Topic::DescribeConsu
     , LastReadTime_(TInstant::Seconds(partitionStats.last_read_time().seconds()))
     , MaxReadTimeLag_(TDuration::Seconds(partitionStats.max_read_time_lag().seconds()))
     , MaxWriteTimeLag_(TDuration::Seconds(partitionStats.max_write_time_lag().seconds()))
+    , MaxCommittedTimeLag_(TDuration::Seconds(partitionStats.max_committed_time_lag().seconds()))
 {}
 
 uint64_t TPartitionConsumerStats::GetCommittedOffset() const {
@@ -406,6 +407,10 @@ const TDuration& TPartitionConsumerStats::GetMaxWriteTimeLag() const {
     return MaxWriteTimeLag_;
 }
 
+const TDuration& TPartitionConsumerStats::GetMaxCommittedTimeLag() const {
+    return MaxCommittedTimeLag_;
+}
+
 TPartitionLocation::TPartitionLocation(const Ydb::Topic::PartitionLocation& partitionLocation)
     : NodeId_(partitionLocation.node_id())
     , Generation_(partitionLocation.generation())

src/client/topic/ut/ut_utils/topic_sdk_test_setup.cpp

@@ -66,6 +66,27 @@ TTopicDescription TTopicSdkTestSetup::DescribeTopic(const TString& path)
     return status.GetTopicDescription();
 }
 
+TConsumerDescription TTopicSdkTestSetup::DescribeConsumer(const TString& path, const TString& consumer)
+{
+    TTopicClient client(MakeDriver());
+
+    TDescribeConsumerSettings settings;
+    settings.IncludeStats(true);
+    settings.IncludeLocation(true);
+
+    auto status = client.DescribeConsumer(path, consumer, settings).GetValueSync();
+    UNIT_ASSERT(status.IsSuccess());
+
+    return status.GetConsumerDescription();
+}
+
+TStatus TTopicSdkTestSetup::Commit(const TString& path, const TString& consumerName, size_t partitionId, size_t offset) {
+    TTopicClient client(MakeDriver());
+
+    return client.CommitOffset(path, partitionId, consumerName, offset).GetValueSync();
+}
+
+
 TString TTopicSdkTestSetup::GetEndpoint() const {
     return "localhost:" + ToString(Server.GrpcPort);
 }

src/client/topic/ut/ut_utils/topic_sdk_test_setup.h

@@ -22,6 +22,9 @@ class TTopicSdkTestSetup {
                      size_t maxPartitionCount = 100);
 
     TTopicDescription DescribeTopic(const TString& path = TString{TEST_TOPIC});
+    TConsumerDescription DescribeConsumer(const TString& path = TString{TEST_TOPIC}, const TString& consumer = TString{TEST_CONSUMER});
+
+    TStatus Commit(const TString& path, const TString& consumerName, size_t partitionId, size_t offset);
 
     TString GetEndpoint() const;
     TString GetTopicPath(const TString& name = TString{TEST_TOPIC}) const;