Merge pull request #1424 Added write topic writer ack

Author: rekby

CHANGELOG.md

@@ -1,3 +1,5 @@
+* Added log topic writer ack
+
 ## v3.77.0
 * Changed log message about send topic message
 * Added experimental support for executing scripts over query service client (`query.Client.ExecuteScript` and `query.CLient.FetchScriptResults`)

internal/grpcwrapper/rawtopic/rawtopicwriter/messages.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"strconv"
 	"time"
 
 	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Topic"
@@ -221,6 +222,48 @@ func (r *WriteResult) fromProto(response *Ydb_Topic.StreamWriteMessage_WriteResp
 	return r.WriteStatistics.fromProto(response.GetWriteStatistics())
 }
 
+// GetAcks implemtnts trace.TopicWriterResultMessagesInfoAcks interface
+func (r *WriteResult) GetAcks() (res traceAck) {
+	res.AcksCount = len(r.Acks)
+	if res.AcksCount > 0 {
+		res.SeqNoMin = r.Acks[0].SeqNo
+		res.WrittenOffsetMin = r.Acks[0].MessageWriteStatus.WrittenOffset
+	}
+	for i := range r.Acks {
+		ack := &r.Acks[i]
+		if ack.MessageWriteStatus.Type == WriteStatusTypeWritten {
+			res.WrittenCount++
+		}
+		if ack.MessageWriteStatus.Type == WriteStatusTypeSkipped {
+			res.SkipCount++
+		}
+
+		if ack.SeqNo < res.SeqNoMin {
+			res.SeqNoMin = ack.SeqNo
+		} else if ack.SeqNo > res.SeqNoMax {
+			res.SeqNoMax = ack.SeqNo
+		}
+
+		if ack.MessageWriteStatus.WrittenOffset < res.SeqNoMin {
+			res.WrittenOffsetMin = ack.MessageWriteStatus.WrittenOffset
+		} else if ack.MessageWriteStatus.WrittenOffset > res.WrittenOffsetMax {
+			res.WrittenOffsetMax = ack.MessageWriteStatus.WrittenOffset
+		}
+	}
+
+	return res
+}
+
+type traceAck = struct {
+	AcksCount        int
+	SeqNoMin         int64
+	SeqNoMax         int64
+	WrittenOffsetMin int64
+	WrittenOffsetMax int64
+	WrittenCount     int
+	SkipCount        int
+}
+
 type WriteAck struct {
 	SeqNo              int64
 	MessageWriteStatus MessageWriteStatus
@@ -269,6 +312,19 @@ const (
 	WriteStatusTypeSkipped
 )
 
+func (t WriteStatusType) String() string {
+	switch t {
+	case WriteStatusTypeUnknown:
+		return "Unknown"
+	case WriteStatusTypeSkipped:
+		return "Skipped"
+	case WriteStatusTypeWritten:
+		return "Written"
+	default:
+		return strconv.Itoa(int(t))
+	}
+}
+
 type WriteStatusSkipReason int
 
 const (

log/topic.go

@@ -820,6 +820,25 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
 			}
 		}
 	}
+	t.OnWriterReceiveResult = func(info trace.TopicWriterResultMessagesInfo) {
+		if d.Details()&trace.TopicWriterStreamEvents == 0 {
+			return
+		}
+		acks := info.Acks.GetAcks()
+		ctx := with(context.Background(), DEBUG, "ydb", "topic", "writer", "receive", "result")
+		l.Log(ctx, "topic writer receive result from server",
+			String("writer_instance_id", info.WriterInstanceID),
+			String("session_id", info.SessionID),
+			Int("acks_count", acks.AcksCount),
+			Int64("seq_no_min", acks.SeqNoMin),
+			Int64("seq_no_max", acks.SeqNoMax),
+			Int64("written_offset_min", acks.WrittenOffsetMin),
+			Int64("written_offset_max", acks.WrittenOffsetMax),
+			Int("written_offset_count", acks.WrittenCount),
+			Int("skip_count", acks.SkipCount),
+			versionField(),
+		)
+	}
 	t.OnWriterReadUnknownGrpcMessage = func(info trace.TopicOnWriterReadUnknownGrpcMessageInfo) {
 		if d.Details()&trace.TopicWriterStreamEvents == 0 {
 			return

trace/topic.go

@@ -121,6 +121,8 @@ type (
 		// Internals: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#internals
 		OnWriterSendMessages func(TopicWriterSendMessagesStartInfo) func(TopicWriterSendMessagesDoneInfo)
 		// Internals: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#internals
+		OnWriterReceiveResult func(TopicWriterResultMessagesInfo)
+		// Internals: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#internals
 		OnWriterReadUnknownGrpcMessage func(TopicOnWriterReadUnknownGrpcMessageInfo)
 	}
 
@@ -470,6 +472,27 @@ type (
 		Error error
 	}
 
+	// Internals: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#internals
+	TopicWriterResultMessagesInfo struct {
+		WriterInstanceID string
+		SessionID        string
+		PartitionID      int64
+		Acks             TopicWriterResultMessagesInfoAcks
+	}
+
+	// Internals: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#internals
+	TopicWriterResultMessagesInfoAcks interface {
+		GetAcks() struct {
+			AcksCount        int
+			SeqNoMin         int64
+			SeqNoMax         int64
+			WrittenOffsetMin int64
+			WrittenOffsetMax int64
+			WrittenCount     int
+			SkipCount        int
+		}
+	}
+
 	// Internals: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#internals
 	TopicOnWriterReadUnknownGrpcMessageInfo struct {
 		WriterInstanceID string