Merge pull request #836 remove a lot of expirements comments.

Author: rekby

CHANGELOG.md

@@ -1,3 +1,8 @@
+* Removed almost all experimental marks from topic api.
+* Rename some topic APIs (old names was deprecated and will be removed in one of next versions).
+* Deprecated topic options (the option will be removed): min size of read messages batch
+* Deprecated WithOnWriterFirstConnected callback, use Writer.WaitInitInfo instead.
+* Changed topic Codec base type from int to int32 (was experimental code)
 * Added `WaitInit` and `WaitInitInfo` method to the topic reader and writer
 
 ## v3.52.2

examples/topic/topicreader/topicreader_advanced.go

@@ -12,12 +12,10 @@ import (
 
 // ReadMessagesWithCustomBatching example of custom of readed message batch
 func ReadMessagesWithCustomBatching(ctx context.Context, db *ydb.Driver) {
-	reader, _ := db.Topic().StartReader("consumer", nil,
-		topicoptions.WithBatchReadMinCount(1000),
-	)
+	reader, _ := db.Topic().StartReader("consumer", nil)
 
 	for {
-		batch, _ := reader.ReadMessageBatch(ctx)
+		batch, _ := reader.ReadMessagesBatch(ctx)
 		processBatch(batch.Context(), batch)
 		_ = reader.Commit(batch.Context(), batch)
 	}

examples/topic/topicreader/topicreader_show.go

@@ -4,13 +4,12 @@ import (
 	"context"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3"
-	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicoptions"
 	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicreader"
 )
 
 // PartitionStopHandled is example of sdk handle server signal about stop partition
 func PartitionStopHandled(ctx context.Context, reader *topicreader.Reader) {
-	batch, _ := reader.ReadMessageBatch(ctx)
+	batch, _ := reader.ReadMessagesBatch(ctx)
 	if len(batch.Messages) == 0 {
 		return
 	}
@@ -21,12 +20,10 @@ func PartitionStopHandled(ctx context.Context, reader *topicreader.Reader) {
 
 // PartitionGracefulStopHandled is example of sdk handle server signal about graceful stop partition
 func PartitionGracefulStopHandled(ctx context.Context, db *ydb.Driver) {
-	reader, _ := db.Topic().StartReader("consumer", nil,
-		topicoptions.WithBatchReadMinCount(1000),
-	)
+	reader, _ := db.Topic().StartReader("consumer", nil)
 
 	for {
-		batch, _ := reader.ReadMessageBatch(ctx) // <- if partition soft stop batch can be less, then 1000
+		batch, _ := reader.ReadMessagesBatch(ctx) // <- if partition soft stop batch can be less, then 1000
 		processBatch(batch.Context(), batch)
 		_ = reader.Commit(batch.Context(), batch)
 	}

internal/grpcwrapper/rawtopic/rawtopiccommon/codec.go

@@ -5,7 +5,7 @@ import (
 )
 
 // Codec any int value, for example for custom codec
-type Codec int
+type Codec int32
 
 const (
 	CodecUNSPECIFIED Codec = iota

internal/topic/topicclientinternal/client.go

@@ -55,20 +55,12 @@ func newTopicConfig(opts ...topicoptions.TopicOption) topic.Config {
 	return c
 }
 
-// Close
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
+// Close the client
 func (c *Client) Close(_ context.Context) error {
 	return nil
 }
 
 // Alter topic options
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 func (c *Client) Alter(ctx context.Context, path string, opts ...topicoptions.AlterOption) error {
 	req := &rawtopic.AlterTopicRequest{}
 	req.OperationParams = c.defaultOperationParams
@@ -91,10 +83,6 @@ func (c *Client) Alter(ctx context.Context, path string, opts ...topicoptions.Al
 }
 
 // Create new topic
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 func (c *Client) Create(
 	ctx context.Context,
 	path string,
@@ -122,10 +110,6 @@ func (c *Client) Create(
 }
 
 // Describe topic
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 func (c *Client) Describe(
 	ctx context.Context,
 	path string,
@@ -166,10 +150,6 @@ func (c *Client) Describe(
 }
 
 // Drop topic
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 func (c *Client) Drop(ctx context.Context, path string, opts ...topicoptions.DropOption) error {
 	req := rawtopic.DropTopicRequest{}
 	req.OperationParams = c.defaultOperationParams
@@ -195,10 +175,6 @@ func (c *Client) Drop(ctx context.Context, path string, opts ...topicoptions.Dro
 
 // StartReader create new topic reader and start pull messages from server
 // it is fast non block call, connection will start in background
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 func (c *Client) StartReader(
 	consumer string,
 	readSelectors topicoptions.ReadSelectors,
@@ -224,10 +200,6 @@ func (c *Client) StartReader(
 }
 
 // StartWriter create new topic writer wrapper
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 func (c *Client) StartWriter(topicPath string, opts ...topicoptions.WriterOption) (*topicwriter.Writer, error) {
 	var connector topicwriterinternal.ConnectFunc = func(ctx context.Context) (
 		topicwriterinternal.RawTopicWriterStream,

internal/topic/topicreaderinternal/batch.go

@@ -15,10 +15,6 @@ var (
 )
 
 // PublicBatch is ordered group of message from one partition
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 type PublicBatch struct {
 	empty.DoNotCopy
 
@@ -100,28 +96,16 @@ func newBatchFromStream(
 
 // Context is cancelled when code should stop to process messages batch
 // for example - lost connection to server or receive stop partition signal without graceful flag
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 func (m *PublicBatch) Context() context.Context {
 	return m.commitRange.partitionSession.Context()
 }
 
 // Topic is path of source topic of the messages in the batch
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 func (m *PublicBatch) Topic() string {
 	return m.partitionSession().Topic
 }
 
 // PartitionID of messages in the batch
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 func (m *PublicBatch) PartitionID() int64 {
 	return m.partitionSession().PartitionID
 }

internal/topic/topicreaderinternal/commit_range.go

@@ -8,10 +8,6 @@ import (
 )
 
 // PublicCommitRangeGetter return data piece for commit messages range
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 type PublicCommitRangeGetter interface {
 	getCommitRange() PublicCommitRange
 }

internal/topic/topicreaderinternal/committer.go

@@ -23,11 +23,6 @@ var (
 
 type sendMessageToServerFunc func(msg rawtopicreader.ClientMessage) error
 
-// PublicCommitMode
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 type PublicCommitMode int
 
 const (

internal/topic/topicreaderinternal/decoders.go

@@ -39,9 +39,4 @@ func (m *decoderMap) Decode(codec rawtopiccommon.Codec, input io.Reader) (io.Rea
 	))
 }
 
-// PublicCreateDecoderFunc
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 type PublicCreateDecoderFunc func(input io.Reader) (io.Reader, error)

internal/topic/topicreaderinternal/message.go

@@ -15,17 +15,9 @@ import (
 var errMessageWasReadEarly = xerrors.Wrap(errors.New("ydb: message was read early"))
 
 // PublicErrUnexpectedCodec return when try to read message content with unknown codec
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 var PublicErrUnexpectedCodec = errors.New("unexpected codec") //nolint:revive,stylecheck
 
 // PublicMessage is representation of topic message
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 type PublicMessage struct {
 	empty.DoNotCopy
 
@@ -81,19 +73,10 @@ func (m *PublicMessage) Read(p []byte) (n int, err error) {
 }
 
 // PublicMessageContentUnmarshaler is interface for unmarshal message content
-//
-// # Experimental
-//
-// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 type PublicMessageContentUnmarshaler interface {
 	// UnmarshalYDBTopicMessage MUST NOT use data after return.
 	// If you need content after return from Consume - copy data content to
 	// own slice with copy(dst, data)
-	//
-	// Experimental
-	//
-	// Notice: This API is EXPERIMENTAL and may be changed or removed in a
-	// later release.
 	UnmarshalYDBTopicMessage(data []byte) error
 }