Merge pull request #359 from ydb-platform/kikimr-15560-fix-nil-stream

fix panic on nil stream in connect failed

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,5 @@
+* Fixed nil pointer exception in topic reader if reconnect failed
+
 ## v3.37.1
 * Refactored the `xsql.badconn.Error`
 

internal/topic/topicreaderinternal/stream_reconnector.go

@@ -242,7 +242,7 @@ func (r *readerReconnector) reconnect(ctx context.Context, oldReader batchedStre
 	}
 
 	r.m.WithLock(func() {
-		r.streamErr = nil
+		r.streamErr = err
 		if err == nil {
 			r.streamVal = newStream
 		}

internal/topic/topicreaderinternal/stream_reconnector_test.go

@@ -341,6 +341,22 @@ func TestTopicReaderReconnectorFireReconnectOnRetryableError(t *testing.T) {
 	})
 }
 
+func TestTopicReaderReconnectorReconnectWithError(t *testing.T) {
+	ctx := context.Background()
+	testErr := errors.New("test")
+	reconnector := &readerReconnector{
+		connectTimeout: infiniteTimeout,
+		readerConnect: func(ctx context.Context) (batchedStreamReader, error) {
+			return nil, testErr
+		},
+		streamErr: errors.New("start-error"),
+	}
+	reconnector.initChannelsAndClock()
+	err := reconnector.reconnect(ctx, nil)
+	require.ErrorIs(t, err, testErr)
+	require.ErrorIs(t, reconnector.streamErr, testErr)
+}
+
 type readerConnectFuncAnswer struct {
 	callback readerConnectFunc
 	stream   batchedStreamReader

topic/reader_e2e_test.go

@@ -27,6 +27,10 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
+const (
+	consumerName = "test-consumer"
+)
+
 func TestReadMessages(t *testing.T) {
 	ctx := testCtx(t)
 
@@ -256,21 +260,21 @@ func createCDCFeed(ctx context.Context, t *testing.T, db ydb.Connection) {
 	})
 	require.NoError(t, err)
 
-	topicPath := db.Name() + "/test/feed"
+	topicPath := testCDCFeedName(db)
 
 	require.NoError(t, err)
 
 	err = db.Topic().Alter(
 		ctx,
 		topicPath,
-		topicoptions.AlterWithAddConsumers(topictypes.Consumer{Name: "test"}),
+		topicoptions.AlterWithAddConsumers(topictypes.Consumer{Name: consumerName}),
 	)
 	require.NoError(t, err)
 }
 
 func createFeedReader(t *testing.T, db ydb.Connection, opts ...topicoptions.ReaderOption) *topicreader.Reader {
-	topicPath := db.Name() + "/test/feed"
-	reader, err := db.Topic().StartReader("test", []topicoptions.ReadSelector{
+	topicPath := testCDCFeedName(db)
+	reader, err := db.Topic().StartReader(consumerName, []topicoptions.ReadSelector{
 		{
 			Path: topicPath,
 		},
@@ -316,3 +320,7 @@ func testCtx(t testing.TB) context.Context {
 	pprof.SetGoroutineLabels(pprof.WithLabels(ctx, pprof.Labels("test", t.Name())))
 	return ctx
 }
+
+func testCDCFeedName(db ydb.Connection) string {
+	return db.Name() + "/test/feed"
+}