Merge pull request #1649 from SammyVimes/fix-describe-consumer Fix describe consumer not parsing partition consumer stats

Author: rekby

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Fixed DescribeConsumer ignoring PartitionConsumerStats
 * Added virtualtimestamps field to cdc description
 
 ## v3.99.10

internal/grpcwrapper/rawoptional/rawoptional.go

@@ -48,7 +48,7 @@ func (v *Duration) MustFromProto(proto *durationpb.Duration) {
 }
 
 func (v *Duration) ToDuration() *time.Duration {
-	if v.HasValue {
+	if !v.HasValue {
 		return nil
 	}
 
@@ -96,7 +96,7 @@ func (v *Time) MustFromProto(proto *timestamppb.Timestamp) {
 }
 
 func (v *Time) ToTime() *time.Time {
-	if v.HasValue {
+	if !v.HasValue {
 		return nil
 	}
 

internal/grpcwrapper/rawtopic/describe_consumer.go

@@ -147,5 +147,9 @@ func (pi *DescribeConsumerResultPartitionInfo) FromProto(proto *Ydb_Topic.Descri
 	pi.ChildPartitionIDs = clone.Int64Slice(proto.GetChildPartitionIds())
 	pi.ParentPartitionIDs = clone.Int64Slice(proto.GetParentPartitionIds())
 
+	if err := pi.PartitionConsumerStats.FromProto(proto.GetPartitionConsumerStats()); err != nil {
+		return err
+	}
+
 	return pi.PartitionStats.FromProto(proto.GetPartitionStats())
 }

tests/integration/topic_client_test.go

@@ -247,6 +247,26 @@ func TestDescribeTopicConsumer(t *testing.T) {
 
 	requireAndCleanSubset(&consumer.Consumer.Attributes, &expectedConsumerDesc.Consumer.Attributes)
 
+	for i := range consumer.Partitions {
+		// Fields that are checked here are dynamic and they change with time, so we need to set them to expected values
+		// to make the comparison possible.
+		p := &consumer.Partitions[i]
+
+		require.NotNil(t, p.PartitionStats.LastWriteTime)
+
+		p.PartitionStats.LastWriteTime = expectedConsumerDesc.Partitions[i].PartitionStats.LastWriteTime
+		p.PartitionStats.MaxWriteTimeLag = expectedConsumerDesc.Partitions[i].PartitionStats.MaxWriteTimeLag
+
+		require.NotNil(t, p.PartitionConsumerStats.LastReadTime)
+		require.NotNil(t, p.PartitionConsumerStats.MaxReadTimeLag)
+		require.NotNil(t, p.PartitionConsumerStats.MaxWriteTimeLag)
+
+		p.PartitionConsumerStats.PartitionReadSessionCreateTime = expectedConsumerDesc.Partitions[i].PartitionConsumerStats.PartitionReadSessionCreateTime
+		p.PartitionConsumerStats.LastReadTime = expectedConsumerDesc.Partitions[i].PartitionConsumerStats.LastReadTime
+		p.PartitionConsumerStats.MaxReadTimeLag = expectedConsumerDesc.Partitions[i].PartitionConsumerStats.MaxReadTimeLag
+		p.PartitionConsumerStats.MaxWriteTimeLag = expectedConsumerDesc.Partitions[i].PartitionConsumerStats.MaxWriteTimeLag
+	}
+
 	require.Equal(t, expectedConsumerDesc, consumer)
 }
 

topic/topictypes/topictypes_test.go

@@ -325,7 +325,7 @@ func TestTopicConsumerDescriptionFromRaw(t *testing.T) {
 			v.testName, func(t *testing.T) {
 				d := TopicConsumerDescription{}
 				d.FromRaw(v.rawConsumerDescription)
-				if !reflect.DeepEqual(d.Consumer, v.expectedDescription.Consumer) {
+				if !reflect.DeepEqual(d, v.expectedDescription) {
 					t.Errorf("got\n%+v\nexpected\n%+v", d, v.expectedDescription)
 				}
 			},