Merge pull request #1330 Improve topic reader config validation

Author: rekby

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Improve config validation before start topic reader
 * Added metrics over `db.Table().Do()` and `db.Table().DoTx()`
 * Added method `ydb.ParamsBuilder().Param(name).Any(value)` to add custom `types.Value`
 * Upgraded dependencies:

internal/topic/topicreaderinternal/reader.go

@@ -22,9 +22,7 @@ var (
 		errors.New("ydb: first connection attempt not finished"),
 	))
 	errReaderClosed                 = xerrors.Wrap(errors.New("ydb: reader closed"))
-	errUnexpectedEmptyConsumername  = xerrors.Wrap(errors.New("ydb: create ydb reader with empty consumer name. Set one of: consumer name or option WithReaderWithoutConsumer")) //nolint:lll
 	errSetConsumerAndNoConsumer     = xerrors.Wrap(errors.New("ydb: reader has non empty consumer name and set option WithReaderWithoutConsumer. Only one of them must be set")) //nolint:lll
-	errCantCommitWithoutConsumer    = xerrors.Wrap(errors.New("ydb: reader can't commit messages without consumer"))
 	errCommitSessionFromOtherReader = xerrors.Wrap(errors.New("ydb: commit with session from other reader"))
 )
 
@@ -88,8 +86,11 @@ func NewReader(
 ) (Reader, error) {
 	cfg := convertNewParamsToStreamConfig(consumer, readSelectors, opts...)
 
-	if err := cfg.Validate(); err != nil {
-		return Reader{}, err
+	if errs := cfg.Validate(); len(errs) > 0 {
+		return Reader{}, xerrors.WithStackTrace(fmt.Errorf(
+			"ydb: failed to start topic reader, because is contains error in config: %w",
+			errors.Join(errs...),
+		))
 	}
 
 	readerID := nextReaderID()
@@ -234,20 +235,6 @@ type ReaderConfig struct {
 	topicStreamReaderConfig
 }
 
-func (c *ReaderConfig) Validate() error {
-	if c.Consumer != "" && c.ReadWithoutConsumer {
-		return xerrors.WithStackTrace(errSetConsumerAndNoConsumer)
-	}
-	if c.Consumer == "" && !c.ReadWithoutConsumer {
-		return xerrors.WithStackTrace(errUnexpectedEmptyConsumername)
-	}
-	if c.ReadWithoutConsumer && c.CommitMode != CommitModeNone {
-		return xerrors.WithStackTrace(errCantCommitWithoutConsumer)
-	}
-
-	return nil
-}
-
 type PublicReaderOption func(cfg *ReaderConfig)
 
 func WithCredentials(cred credentials.Credentials) PublicReaderOption {

internal/topic/topicreaderinternal/stream_reader_impl.go

@@ -28,6 +28,10 @@ var (
 	PublicErrCommitSessionToExpiredSession = xerrors.Wrap(errors.New("ydb: commit to expired session"))
 
 	errCommitWithNilPartitionSession = xerrors.Wrap(errors.New("ydb: commit with nil partition session"))
+	errUnexpectedEmptyConsumerName   = xerrors.Wrap(errors.New("ydb: create ydb reader with empty consumer name. Set one of: consumer name or option WithReaderWithoutConsumer")) //nolint:lll
+	errCantCommitWithoutConsumer     = xerrors.Wrap(errors.New("ydb: reader can't commit messages without consumer"))
+	errBufferSize                    = xerrors.Wrap(errors.New("ydb: buffer of topic reader must be greater than zero, see option topicoptions.WithReaderBufferSizeBytes")) //nolint:lll
+	errTopicSelectorsEmpty           = xerrors.Wrap(errors.New("ydb: topic selector for topic reader is empty, see arguments on topic starts"))                             //nolint:lll
 )
 
 type partitionSessionID = rawtopicreader.PartitionSessionID
@@ -86,6 +90,28 @@ func newTopicStreamReaderConfig() topicStreamReaderConfig {
 	}
 }
 
+func (cfg *topicStreamReaderConfig) Validate() []error {
+	var validateErrors []error
+
+	if cfg.Consumer != "" && cfg.ReadWithoutConsumer {
+		validateErrors = append(validateErrors, errSetConsumerAndNoConsumer)
+	}
+	if cfg.Consumer == "" && !cfg.ReadWithoutConsumer {
+		validateErrors = append(validateErrors, errUnexpectedEmptyConsumerName)
+	}
+	if cfg.ReadWithoutConsumer && cfg.CommitMode != CommitModeNone {
+		validateErrors = append(validateErrors, errCantCommitWithoutConsumer)
+	}
+	if cfg.BufferSizeProtoBytes <= 0 {
+		validateErrors = append(validateErrors, errBufferSize)
+	}
+	if len(cfg.ReadSelectors) == 0 {
+		validateErrors = append(validateErrors, errTopicSelectorsEmpty)
+	}
+
+	return validateErrors
+}
+
 func (cfg *topicStreamReaderConfig) initMessage() *rawtopicreader.InitRequest {
 	res := &rawtopicreader.InitRequest{
 		Consumer: cfg.Consumer,