fix default settings to writer in tests

Author: rekby

internal/topic/retriable_error.go

@@ -9,7 +9,7 @@ import (
 )
 
 const (
-	DefaultConnectionTimeout = time.Minute
+	DefaultStartTimeout = time.Minute
 )
 
 type RetrySettings struct {

internal/topic/topicclientinternal/client.go

@@ -201,7 +201,7 @@ func (c *Client) StartReader(
 		topicoptions.WithCommonConfig(c.cfg.Common),
 		topicreaderinternal.WithCredentials(c.cred),
 		topicreaderinternal.WithTrace(c.cfg.Trace),
-		topicoptions.WithReaderStartTimeout(topic.DefaultConnectionTimeout),
+		topicoptions.WithReaderStartTimeout(topic.DefaultStartTimeout),
 	}
 	opts = append(defaultOpts, opts...)
 

internal/topic/topicwriterinternal/writer_options.go

@@ -125,6 +125,12 @@ func WithSessionMeta(meta map[string]string) PublicWriterOption {
 	}
 }
 
+func WithStartTimeout(timeout time.Duration) PublicWriterOption {
+	return func(cfg *WriterReconnectorConfig) {
+		cfg.RetrySettings.StartTimeout = timeout
+	}
+}
+
 func WithWaitAckOnWrite(val bool) PublicWriterOption {
 	return func(cfg *WriterReconnectorConfig) {
 		cfg.WaitServerAck = val

internal/topic/topicwriterinternal/writer_reconnector.go

@@ -85,7 +85,7 @@ func newWriterReconnectorConfig(options ...PublicWriterOption) WriterReconnector
 		MaxMessageSize:     50 * 1024 * 1024,
 		MaxQueueLen:        1000,
 		RetrySettings: topic.RetrySettings{
-			StartTimeout: topic.DefaultConnectionTimeout,
+			StartTimeout: topic.DefaultStartTimeout,
 		},
 	}
 	if cfg.compressorCount == 0 {
@@ -141,6 +141,7 @@ func newWriterReconnectorStopped(cfg WriterReconnectorConfig) *WriterReconnector
 		firstInitResponseProcessedChan: make(empty.Chan),
 		encodersMap:                    NewEncoderMap(),
 		writerInstanceID:               writerInstanceID.String(),
+		retrySettings:                  cfg.RetrySettings,
 	}
 
 	res.queue.OnAckReceived = res.onAckReceived
@@ -356,7 +357,7 @@ func (w *WriterReconnector) connectionLoop(ctx context.Context) {
 		now := time.Now()
 		if topic.CheckResetReconnectionCounters(prevAttemptTime, now, w.cfg.connectTimeout) {
 			attempt = 0
-			startOfRetries = time.Now()
+			startOfRetries = w.clock.Now()
 		} else {
 			attempt++
 		}

internal/topic/topicwriterinternal/writer_reconnector_test.go

@@ -360,7 +360,7 @@ func TestWriterImpl_Reconnect(t *testing.T) {
 		require.ErrorIs(t, w.background.CloseReason(), testErr)
 	})
 
-	t.Run("ReconnectOnErrors", func(t *testing.T) {
+	xtest.TestManyTimesWithName(t, "ReconnectOnErrors", func(t testing.TB) {
 		ctx := xtest.Context(t)
 
 		w := newTestWriterStopped()
@@ -400,10 +400,6 @@ func TestWriterImpl_Reconnect(t *testing.T) {
 		}
 
 		strm2InitSent := make(empty.Chan)
-		go func() {
-			err := w.Write(ctx, newTestMessages(1))
-			require.NoError(t, err)
-		}()
 
 		strm2 := newStream("strm2", func() {
 			close(strm2InitSent)
@@ -446,7 +442,16 @@ func TestWriterImpl_Reconnect(t *testing.T) {
 			return res.stream, res.connectionError
 		}
 
-		w.connectionLoop(ctx)
+		connectionLoopStopped := make(empty.Chan)
+		go func() {
+			defer close(connectionLoopStopped)
+			w.connectionLoop(ctx)
+		}()
+
+		err := w.Write(ctx, newTestMessages(1))
+		require.NoError(t, err)
+
+		xtest.WaitChannelClosed(t, connectionLoopStopped)
 	})
 }
 

topic/topicoptions/topicoptions_writer.go

@@ -183,9 +183,7 @@ func WithWriterSetAutoCreatedAt(val bool) WriterOption {
 //
 // Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 func WithWriterStartTimeout(timeout time.Duration) WriterOption {
-	return func(cfg *topicwriterinternal.WriterReconnectorConfig) {
-		cfg.RetrySettings.StartTimeout = timeout
-	}
+	return topicwriterinternal.WithStartTimeout(timeout)
 }
 
 // WithWriterTrace