Merge pull request #1231 Fixed data race on close topic writer

Author: rekby

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Fixed race of stop internal processes on close topic writer
 * Fixed goroutines leak within topic reader on network problems
 
 ## v3.67.0

internal/background/worker.go

@@ -3,6 +3,7 @@ package background
 import (
 	"context"
 	"errors"
+	"fmt"
 	"runtime/pprof"
 	"sync"
 
@@ -75,7 +76,9 @@ func (b *Worker) Close(ctx context.Context, err error) error {
 	var resErr error
 	b.m.WithLock(func() {
 		if b.closed {
-			resErr = xerrors.WithStackTrace(ErrAlreadyClosed)
+			// The error of Close is second close, close reason added for describe previous close only, for better debug
+			//nolint:errorlint
+			resErr = xerrors.WithStackTrace(fmt.Errorf("%w with reason: %+v", ErrAlreadyClosed, b.closeReason))
 
 			return
 		}

internal/topic/topicwriterinternal/writer_reconnector.go

@@ -347,16 +347,17 @@ func (w *WriterReconnector) close(ctx context.Context, reason error) (resErr err
 		onDone(resErr)
 	}()
 
-	closeErr := w.queue.Close(reason)
-	if resErr == nil && closeErr != nil {
-		resErr = closeErr
-	}
-
+	// stop background work and single stream writer
 	bgErr := w.background.Close(ctx, reason)
 	if resErr == nil && bgErr != nil {
 		resErr = bgErr
 	}
 
+	closeErr := w.queue.Close(reason)
+	if resErr == nil && closeErr != nil {
+		resErr = closeErr
+	}
+
 	return resErr
 }
 

internal/topic/topicwriterinternal/writer_reconnector_test.go

@@ -555,6 +555,8 @@ func TestWriterImpl_CloseWithFlush(t *testing.T) {
 				},
 			},
 			Codec: rawtopiccommon.CodecRaw,
+		}).Do(func(_ *rawtopicwriter.WriteRequest) {
+			close(writeCompleted)
 		}).Return(nil)
 
 		flushCompleted := make(empty.Chan)
@@ -564,7 +566,6 @@ func TestWriterImpl_CloseWithFlush(t *testing.T) {
 				CreatedAt: messageTime,
 				Data:      bytes.NewReader(messageData),
 			}})
-			close(writeCompleted)
 			require.NoError(t, err)
 		}()
 
@@ -611,11 +612,11 @@ func TestWriterImpl_CloseWithFlush(t *testing.T) {
 		{
 			name: "flush",
 			flush: func(ctx context.Context, writer *WriterReconnector) error {
-				return writer.Close(ctx)
+				return writer.Flush(ctx)
 			},
 		},
 		{
-			name: "flush and close",
+			name: "flush_and_close",
 			flush: func(ctx context.Context, writer *WriterReconnector) error {
 				err := writer.Flush(ctx)
 				if err != nil {