Repository: xataio/pgstream (go)

Issue:
[Batch processor] Graceful shutdown
The batch sender should stop gracefully when the `Close` method is called. It currently hangs if the close is called without cancelling the context.

Pull request:
[Batch sender] Graceful shutdown
This PR updates the batch sender implementation to cancel the send when the close is called, it otherwise hangs waiting for the send thread to finish until the context is cancelled or an error occurs.

The initialisation of the send goroutine is moved to the batch sender, since it's an implementation detail and should be owned and controlled by the batch sender rather than by the users (postgres writer, search indexer, kafka writer processors). It helps keeping the complexity of the batching mechanism contained within the batch sender, while keeping the processors implementation simple.

Closes #176 

Patch:
diff --git a/pkg/wal/processor/batch/mocks/mock_batch_sender.go b/pkg/wal/processor/batch/mocks/mock_batch_sender.go
index efbc999..3c09304 100644
--- a/pkg/wal/processor/batch/mocks/mock_batch_sender.go
+++ b/pkg/wal/processor/batch/mocks/mock_batch_sender.go
@@ -9,9 +9,9 @@ import (
 )
 
 type BatchSender[T batch.Message] struct {
-	AddToBatchFn func(context.Context, *batch.WALMessage[T]) error
-	SendFn       func(context.Context) error
-	msgChan      chan *batch.WALMessage[T]
+	SendMessageFn func(context.Context, *batch.WALMessage[T]) error
+	CloseFn       func()
+	msgChan       chan *batch.WALMessage[T]
 }
 
 func NewBatchSender[T batch.Message]() *BatchSender[T] {
@@ -20,9 +20,9 @@ func NewBatchSender[T batch.Message]() *BatchSender[T] {
 	}
 }
 
-func (m *BatchSender[T]) AddToBatch(ctx context.Context, msg *batch.WALMessage[T]) error {
-	if m.AddToBatchFn != nil {
-		return m.AddToBatchFn(ctx, msg)
+func (m *BatchSender[T]) SendMessage(ctx context.Context, msg *batch.WALMessage[T]) error {
+	if m.SendMessageFn != nil {
+		return m.SendMessageFn(ctx, msg)
 	}
 
 	m.msgChan <- msg
@@ -31,15 +31,9 @@ func (m *BatchSender[T]) AddToBatch(ctx context.Context, msg *batch.WALMessage[T
 
 func (m *BatchSender[T]) Close() {
 	close(m.msgChan)
-}
-
-func (m *BatchSender[T]) Send(ctx context.Context) error {
-	if m.SendFn != nil {
-		return m.SendFn(ctx)
+	if m.CloseFn != nil {
+		m.CloseFn()
 	}
-
-	_ = m.GetWALMessages()
-	return nil
 }
 
 func (m *BatchSender[T]) GetWALMessages() []*batch.WALMessage[T] {
diff --git a/pkg/wal/processor/batch/wal_batch_sender.go b/pkg/wal/processor/batch/wal_batch_sender.go
index d1c888b..92d1180 100644
--- a/pkg/wal/processor/batch/wal_batch_sender.go
+++ b/pkg/wal/processor/batch/wal_batch_sender.go
@@ -29,7 +29,8 @@ type Sender[T Message] struct {
 	maxBatchSize      int64
 	batchSendInterval time.Duration
 
-	wg *sync.WaitGroup
+	wg       *sync.WaitGroup
+	cancelFn context.CancelFunc
 
 	sendBatchFn sendBatchFn[T]
 }
@@ -38,7 +39,7 @@ type sendBatchFn[T Message] func(context.Context, *Batch[T]) error
 
 var errSendStopped = errors.New("stop processing, sending has stopped")
 
-func NewSender[T Message](config *Config, sendfn sendBatchFn[T], logger loglib.Logger) (*Sender[T], error) {
+func NewSender[T Message](ctx context.Context, config *Config, sendfn sendBatchFn[T], logger loglib.Logger) (*Sender[T], error) {
 	s := &Sender[T]{
 		batchSendInterval: config.GetBatchTimeout(),
 		maxBatchBytes:     config.GetMaxBatchBytes(),
@@ -49,6 +50,7 @@ func NewSender[T Message](config *Config, sendfn sendBatchFn[T], logger loglib.L
 		logger:            logger,
 		sendBatchFn:       sendfn,
 		wg:                &sync.WaitGroup{},
+		cancelFn:          func() {},
 	}
 
 	maxQueueBytes, err := config.GetMaxQueueBytes()
@@ -57,10 +59,22 @@ func NewSender[T Message](config *Config, sendfn sendBatchFn[T], logger loglib.L
 	}
 	s.queueBytesSema = synclib.NewWeightedSemaphore(int64(maxQueueBytes))
 
+	// start the send process in the background
+	go func() {
+		ctx, s.cancelFn = context.WithCancel(ctx)
+		defer s.cancelFn()
+
+		if err := s.send(ctx); err != nil && !errors.Is(err, context.Canceled) {
+			s.logger.Error(err, "sending stopped")
+		}
+	}()
+
 	return s, nil
 }
 
-func (s *Sender[T]) AddToBatch(ctx context.Context, msg *WALMessage[T]) error {
+// SendMessage adds the message to the batch, which will be sent when the interval or the
+// max number of messages is reached by a background process.
+func (s *Sender[T]) SendMessage(ctx context.Context, msg *WALMessage[T]) error {
 	if msg == nil {
 		return nil
 	}
@@ -94,7 +108,7 @@ func (s *Sender[T]) AddToBatch(ctx context.Context, msg *WALMessage[T]) error {
 	return nil
 }
 
-func (s *Sender[T]) Send(ctx context.Context) error {
+func (s *Sender[T]) send(ctx context.Context) error {
 	// make sure we send on a separate go routine to isolate the IO operations,
 	// ensuring the goroutine is always sending, and minimise the wait time
 	// between batch sending
@@ -187,6 +201,8 @@ func (s *Sender[T]) Send(ctx context.Context) error {
 }
 
 func (s *Sender[T]) Close() {
+	s.cancelFn()
+	// wait for the send loop to finish
 	s.wg.Wait()
 	s.closeMsgChan()
 }
diff --git a/pkg/wal/processor/kafka/wal_kafka_batch_writer.go b/pkg/wal/processor/kafka/wal_kafka_batch_writer.go
index cc0c415..24229de 100644
--- a/pkg/wal/processor/kafka/wal_kafka_batch_writer.go
+++ b/pkg/wal/processor/kafka/wal_kafka_batch_writer.go
@@ -37,9 +37,8 @@ type BatchWriter struct {
 type Option func(*BatchWriter)
 
 type batchSender interface {
-	AddToBatch(context.Context, *batch.WALMessage[kafka.Message]) error
+	SendMessage(context.Context, *batch.WALMessage[kafka.Message]) error
 	Close()
-	Send(context.Context) error
 }
 
 var errRecordTooLarge = errors.New("record too large")
@@ -77,18 +76,11 @@ func NewBatchWriter(ctx context.Context, config *Config, opts ...Option) (*Batch
 		opt(w)
 	}
 
-	w.batchSender, err = batch.NewSender(&config.Batch, w.sendBatch, w.logger)
+	w.batchSender, err = batch.NewSender(ctx, &config.Batch, w.sendBatch, w.logger)
 	if err != nil {
 		return nil, err
 	}
 
-	// start the send process in the background
-	go func() {
-		if err := w.batchSender.Send(ctx); err != nil {
-			w.logger.Error(err, "sending stopped")
-		}
-	}()
-
 	return w, nil
 }
 
@@ -159,7 +151,7 @@ func (w *BatchWriter) ProcessWALEvent(ctx context.Context, walEvent *wal.Event)
 	}
 
 	msg := batch.NewWALMessage(kafkaMsg, walEvent.CommitPosition)
-	return w.batchSender.AddToBatch(ctx, msg)
+	return w.batchSender.SendMessage(ctx, msg)
 }
 
 func (w *BatchWriter) Name() string {
diff --git a/pkg/wal/processor/postgres/postgres_batch_writer.go b/pkg/wal/processor/postgres/postgres_batch_writer.go
index 2a6a56f..2e4742c 100644
--- a/pkg/wal/processor/postgres/postgres_batch_writer.go
+++ b/pkg/wal/processor/postgres/postgres_batch_writer.go
@@ -35,9 +35,8 @@ type BatchWriter struct {
 type Option func(*BatchWriter)
 
 type queryBatchSender interface {
-	AddToBatch(context.Context, *batch.WALMessage[*query]) error
+	SendMessage(context.Context, *batch.WALMessage[*query]) error
 	Close()
-	Send(context.Context) error
 }
 
 // NewBatchWriter returns a postgres processor that batches and writes data to
@@ -80,18 +79,11 @@ func NewBatchWriter(ctx context.Context, config *Config, opts ...Option) (*Batch
 		}
 	}
 
-	w.batchSender, err = batch.NewSender(&config.BatchConfig, w.sendBatch, w.logger)
+	w.batchSender, err = batch.NewSender(ctx, &config.BatchConfig, w.sendBatch, w.logger)
 	if err != nil {
 		return nil, err
 	}
 
-	// start the send process in the background
-	go func() {
-		if err := w.batchSender.Send(ctx); err != nil && !errors.Is(err, context.Canceled) {
-			w.logger.Error(err, "sending stopped")
-		}
-	}()
-
 	return w, nil
 }
 
@@ -132,7 +124,7 @@ func (w *BatchWriter) ProcessWALEvent(ctx context.Context, walEvent *wal.Event)
 	for _, q := range queries {
 		w.logger.Debug("batching query", loglib.Fields{"sql": q.getSQL(), "args": q.getArgs(), "commit_position": walEvent.CommitPosition})
 		msg := batch.NewWALMessage(q, walEvent.CommitPosition)
-		if err := w.batchSender.AddToBatch(ctx, msg); err != nil {
+		if err := w.batchSender.SendMessage(ctx, msg); err != nil {
 			return err
 		}
 	}
diff --git a/pkg/wal/processor/search/search_batch_indexer.go b/pkg/wal/processor/search/search_batch_indexer.go
index 871cd8d..3fad47e 100644
--- a/pkg/wal/processor/search/search_batch_indexer.go
+++ b/pkg/wal/processor/search/search_batch_indexer.go
@@ -34,9 +34,8 @@ type BatchIndexer struct {
 type Option func(*BatchIndexer)
 
 type batchSender interface {
-	AddToBatch(context.Context, *batch.WALMessage[*msg]) error
+	SendMessage(context.Context, *batch.WALMessage[*msg]) error
 	Close()
-	Send(context.Context) error
 }
 
 // NewBatchIndexer returns a processor of wal events that indexes data into the
@@ -55,18 +54,11 @@ func NewBatchIndexer(ctx context.Context, config IndexerConfig, store Store, lsn
 	}
 
 	var err error
-	indexer.batchSender, err = batch.NewSender(&config.Batch, indexer.sendBatch, indexer.logger)
+	indexer.batchSender, err = batch.NewSender(ctx, &config.Batch, indexer.sendBatch, indexer.logger)
 	if err != nil {
 		return nil, err
 	}
 
-	// start the send process in the background
-	go func() {
-		if err := indexer.batchSender.Send(ctx); err != nil {
-			indexer.logger.Error(err, "sending stopped")
-		}
-	}()
-
 	return indexer, nil
 }
 
@@ -116,7 +108,7 @@ func (i *BatchIndexer) ProcessWALEvent(ctx context.Context, event *wal.Event) (e
 		return nil
 	}
 
-	return i.batchSender.AddToBatch(ctx, batch.NewWALMessage(msg, event.CommitPosition))
+	return i.batchSender.SendMessage(ctx, batch.NewWALMessage(msg, event.CommitPosition))
 }
 
 func (i *BatchIndexer) Name() string {


Test patch:
diff --git a/pkg/wal/processor/batch/wal_batch_sender_test.go b/pkg/wal/processor/batch/wal_batch_sender_test.go
index 9aa34c0..35879c6 100644
--- a/pkg/wal/processor/batch/wal_batch_sender_test.go
+++ b/pkg/wal/processor/batch/wal_batch_sender_test.go
@@ -15,7 +15,7 @@ import (
 	"github.com/xataio/pgstream/pkg/wal"
 )
 
-func TestSender_AddToBatch(t *testing.T) {
+func TestSender_SendMessage(t *testing.T) {
 	t.Parallel()
 
 	noopSendFn := func(ctx context.Context, b *Batch[*mockMessage]) error { return nil }
@@ -117,12 +117,23 @@ func TestSender_AddToBatch(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			batchSender, err := NewSender(&Config{}, noopSendFn, log.NewNoopLogger())
-			require.NoError(t, err)
-			batchSender.queueBytesSema = tc.weightedSemaphore
+			ctx := context.Background()
+
+			batchSender := &Sender[*mockMessage]{
+				batchSendInterval: 100 * time.Millisecond,
+				maxBatchSize:      10,
+				msgChan:           make(chan *WALMessage[*mockMessage]),
+				queueBytesSema:    tc.weightedSemaphore,
+				sendDone:          make(chan error, 1),
+				once:              &sync.Once{},
+				logger:            log.NewNoopLogger(),
+				sendBatchFn:       noopSendFn,
+				wg:                &sync.WaitGroup{},
+				cancelFn:          func() {},
+			}
 			defer batchSender.Close()
 
-			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 			defer cancel()
 
 			if tc.sendDone {
@@ -137,7 +148,7 @@ func TestSender_AddToBatch(t *testing.T) {
 					batchSender.closeMsgChan()
 					wg.Done()
 				}()
-				err = batchSender.AddToBatch(ctx, tc.msg)
+				err := batchSender.SendMessage(ctx, tc.msg)
 				require.ErrorIs(t, err, tc.wantErr)
 			}()
 
@@ -159,7 +170,7 @@ func TestSender_AddToBatch(t *testing.T) {
 	}
 }
 
-func TestSender_Send(t *testing.T) {
+func TestSender_send(t *testing.T) {
 	t.Parallel()
 
 	testCommitPos := wal.CommitPosition("1")
@@ -264,26 +275,33 @@ func TestSender_Send(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
+			ctx := context.Background()
+
 			doneChan := make(chan struct{}, 1)
 			defer close(doneChan)
 
-			sender, err := NewSender(&Config{
-				BatchTimeout: 100 * time.Millisecond,
-				MaxBatchSize: 10,
-			}, tc.sendFn(doneChan), log.NewNoopLogger())
-			require.NoError(t, err)
+			sender := &Sender[*mockMessage]{
+				batchSendInterval: 100 * time.Millisecond,
+				maxBatchSize:      10,
+				msgChan:           make(chan *WALMessage[*mockMessage]),
+				queueBytesSema:    tc.semaphore,
+				sendDone:          make(chan error, 1),
+				once:              &sync.Once{},
+				logger:            log.NewNoopLogger(),
+				sendBatchFn:       tc.sendFn(doneChan),
+				wg:                &sync.WaitGroup{},
+				cancelFn:          func() {},
+			}
 			defer sender.Close()
 
-			sender.queueBytesSema = tc.semaphore
-
-			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 			defer cancel()
 
 			wg := sync.WaitGroup{}
 			wg.Add(1)
 			go func() {
 				defer wg.Done()
-				err := sender.Send(ctx)
+				err := sender.send(ctx)
 				require.ErrorIs(t, err, tc.wantErr)
 			}()
 
@@ -310,6 +328,7 @@ func TestSender_Send(t *testing.T) {
 
 	t.Run("graceful shutdown, drain in-flight batch", func(t *testing.T) {
 		t.Parallel()
+		ctx := context.Background()
 
 		doneChan := make(chan struct{}, 1)
 		defer close(doneChan)
@@ -327,28 +346,33 @@ func TestSender_Send(t *testing.T) {
 			}
 		}
 
-		sender, err := NewSender(&Config{
-			BatchTimeout: 1 * time.Minute,
-			MaxBatchSize: 10,
-		}, sendFn(doneChan), log.NewNoopLogger())
-		require.NoError(t, err)
-		defer sender.Close()
-
-		sender.queueBytesSema = &syncmocks.WeightedSemaphore{
-			ReleaseFn: func(i uint64, bytes int64) {
-				if i == 0 {
-					require.Equal(t, int64(1), bytes)
-				}
+		sender := &Sender[*mockMessage]{
+			batchSendInterval: 1 * time.Minute,
+			maxBatchSize:      10,
+			msgChan:           make(chan *WALMessage[*mockMessage]),
+			queueBytesSema: &syncmocks.WeightedSemaphore{
+				ReleaseFn: func(i uint64, bytes int64) {
+					if i == 0 {
+						require.Equal(t, int64(1), bytes)
+					}
+				},
 			},
+			sendDone:    make(chan error, 1),
+			once:        &sync.Once{},
+			logger:      log.NewNoopLogger(),
+			sendBatchFn: sendFn(doneChan),
+			wg:          &sync.WaitGroup{},
+			cancelFn:    func() {},
 		}
+		defer sender.Close()
 
-		ctx, cancel := context.WithCancel(context.Background())
+		ctx, cancel := context.WithCancel(ctx)
 
 		wg := sync.WaitGroup{}
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			err := sender.Send(ctx)
+			err := sender.send(ctx)
 			require.ErrorIs(t, err, context.Canceled)
 		}()
 
@@ -357,3 +381,61 @@ func TestSender_Send(t *testing.T) {
 		wg.Wait()
 	})
 }
+
+func TestSender(t *testing.T) {
+	ctx := context.Background()
+
+	errTest := errors.New("oh noes")
+	testCommitPos := wal.CommitPosition("1")
+
+	mockMsg := func(i uint) *mockMessage {
+		return &mockMessage{id: i}
+	}
+
+	testWALMsg := func(i uint) *WALMessage[*mockMessage] {
+		return NewWALMessage(mockMsg(i), testCommitPos)
+	}
+
+	doneChan := make(chan struct{}, 1)
+	defer close(doneChan)
+
+	sendFn := func(doneChan chan<- struct{}) sendBatchFn[*mockMessage] {
+		once := sync.Once{}
+		return func(ctx context.Context, b *Batch[*mockMessage]) error {
+			defer once.Do(func() { doneChan <- struct{}{} })
+
+			require.Len(t, b.messages, 1)
+			require.Len(t, b.positions, 1)
+			require.Equal(t, mockMsg(1), b.messages[0])
+			require.Equal(t, testCommitPos, b.positions[0])
+			return errTest
+		}
+	}
+
+	sender, err := NewSender(ctx, &Config{
+		BatchTimeout: 100 * time.Millisecond,
+		MaxBatchSize: 1,
+	}, sendFn(doneChan), log.NewNoopLogger())
+	require.NoError(t, err)
+	defer sender.Close()
+
+	err = sender.SendMessage(ctx, testWALMsg(1))
+	require.NoError(t, err)
+
+	timer := time.NewTimer(5 * time.Second)
+	defer timer.Stop()
+	for {
+		select {
+		case <-doneChan:
+			// give time for the error to cascade
+			time.Sleep(100 * time.Millisecond)
+			err = sender.SendMessage(ctx, testWALMsg(2))
+			require.ErrorIs(t, err, errSendStopped)
+			sender.Close()
+			return
+		case <-timer.C:
+			t.Error("test timeout")
+			return
+		}
+	}
+}
diff --git a/pkg/wal/processor/kafka/wal_kafka_batch_writer_test.go b/pkg/wal/processor/kafka/wal_kafka_batch_writer_test.go
index dce8e62..f07f665 100644
--- a/pkg/wal/processor/kafka/wal_kafka_batch_writer_test.go
+++ b/pkg/wal/processor/kafka/wal_kafka_batch_writer_test.go
@@ -7,11 +7,8 @@ import (
 	"errors"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/require"
-	"github.com/xataio/pgstream/internal/json"
-	"github.com/xataio/pgstream/internal/log/zerolog"
 	"github.com/xataio/pgstream/pkg/kafka"
 	kafkamocks "github.com/xataio/pgstream/pkg/kafka/mocks"
 	loglib "github.com/xataio/pgstream/pkg/log"
@@ -294,63 +291,3 @@ func TestBatchKafkaWriter_sendBatch(t *testing.T) {
 		})
 	}
 }
-
-func TestBatchKafkaWriter(t *testing.T) {
-	t.Parallel()
-
-	bw := BatchWriter{
-		logger: newTestLogger(),
-		writer: &kafkamocks.Writer{
-			WriteMessagesFn: func(ctx context.Context, i uint64, msgs ...kafka.Message) error {
-				time.Sleep(time.Second)
-				return errTest
-			},
-		},
-		maxBatchBytes: 10000,
-		serialiser:    json.Marshal,
-	}
-	defer bw.Close()
-
-	var err error
-	bw.batchSender, err = batch.NewSender(&batch.Config{
-		BatchTimeout:  time.Second,
-		MaxBatchSize:  10,
-		MaxBatchBytes: 10000,
-	}, bw.sendBatch, bw.logger)
-	require.NoError(t, err)
-
-	doneChan := make(chan struct{}, 1)
-	go func() {
-		err := bw.batchSender.Send(context.Background())
-		require.ErrorIs(t, err, errTest)
-		doneChan <- struct{}{}
-		close(doneChan)
-	}()
-
-	timer := time.NewTimer(5 * time.Second)
-	defer timer.Stop()
-	var processErr error
-	for {
-		select {
-		case <-doneChan:
-			require.ErrorIs(t, processErr, errTest)
-			return
-		case <-timer.C:
-			t.Error("test timeout")
-			return
-		default:
-			processErr = bw.ProcessWALEvent(context.Background(), &wal.Event{
-				CommitPosition: wal.CommitPosition("1"),
-				Data: &wal.Data{
-					Action: "I",
-				},
-			})
-		}
-	}
-}
-
-func newTestLogger() loglib.Logger {
-	return zerolog.NewStdLogger(zerolog.NewLogger(&zerolog.Config{
-		LogLevel: "trace",
-	}))
-}
diff --git a/pkg/wal/processor/postgres/postgres_batch_writer_test.go b/pkg/wal/processor/postgres/postgres_batch_writer_test.go
index 92ffe8b..c356ad4 100644
--- a/pkg/wal/processor/postgres/postgres_batch_writer_test.go
+++ b/pkg/wal/processor/postgres/postgres_batch_writer_test.go
@@ -91,7 +91,7 @@ func TestBatchWriter_ProcessWALEvent(t *testing.T) {
 			walEvent: testWalEvent,
 			batchSender: func() *batchmocks.BatchSender[*query] {
 				s := batchmocks.NewBatchSender[*query]()
-				s.AddToBatchFn = func(ctx context.Context, w *batch.WALMessage[*query]) error { return errTest }
+				s.SendMessageFn = func(ctx context.Context, w *batch.WALMessage[*query]) error { return errTest }
 				return s
 			}(),
 			adapter: &mockAdapter{
@@ -345,73 +345,3 @@ func TestBatchWriter_flushQueries(t *testing.T) {
 		})
 	}
 }
-
-func TestBatchWriter(t *testing.T) {
-	t.Parallel()
-
-	conn := &pgmocks.Querier{
-		ExecInTxFn: func(ctx context.Context, f func(tx pglib.Tx) error) error {
-			mockTx := pgmocks.Tx{
-				ExecFn: func(ctx context.Context, query string, args ...any) (pglib.CommandTag, error) {
-					time.Sleep(time.Second)
-					return pglib.CommandTag{}, errTest
-				},
-			}
-			return f(&mockTx)
-		},
-		CloseFn: func(ctx context.Context) error { return nil },
-	}
-
-	adapter := &mockAdapter{
-		walEventToQueriesFn: func(e *wal.Event) ([]*query, error) {
-			return []*query{{
-				sql:  "INSERT INTO test(id, name) VALUES($1, $2) ON CONFLICT (id) DO NOTHING",
-				args: []any{1, "alice"},
-			}}, nil
-		},
-	}
-
-	bw := BatchWriter{
-		logger:  loglib.NewNoopLogger(),
-		pgConn:  conn,
-		adapter: adapter,
-	}
-	defer bw.Close()
-
-	var err error
-	bw.batchSender, err = batch.NewSender(&batch.Config{
-		BatchTimeout:  time.Second,
-		MaxBatchSize:  10,
-		MaxBatchBytes: 10000,
-	}, bw.sendBatch, bw.logger)
-	require.NoError(t, err)
-
-	doneChan := make(chan struct{}, 1)
-	go func() {
-		err := bw.batchSender.Send(context.Background())
-		require.ErrorIs(t, err, errTest)
-		doneChan <- struct{}{}
-		close(doneChan)
-	}()
-
-	timer := time.NewTimer(5 * time.Second)
-	defer timer.Stop()
-	var processErr error
-	for {
-		select {
-		case <-doneChan:
-			require.ErrorIs(t, processErr, errTest)
-			return
-		case <-timer.C:
-			t.Error("test timeout")
-			return
-		default:
-			processErr = bw.ProcessWALEvent(context.Background(), &wal.Event{
-				CommitPosition: wal.CommitPosition("1"),
-				Data: &wal.Data{
-					Action: "I",
-				},
-			})
-		}
-	}
-}
diff --git a/pkg/wal/processor/search/search_batch_indexer_test.go b/pkg/wal/processor/search/search_batch_indexer_test.go
index 382eb89..1ceb64b 100644
--- a/pkg/wal/processor/search/search_batch_indexer_test.go
+++ b/pkg/wal/processor/search/search_batch_indexer_test.go
@@ -469,62 +469,3 @@ func TestBatchIndexer_sendBatch(t *testing.T) {
 		})
 	}
 }
-
-func TestBatchIndexer(t *testing.T) {
-	t.Parallel()
-
-	indexer := BatchIndexer{
-		logger: loglib.NewNoopLogger(),
-		store: &mockStore{
-			sendDocumentsFn: func(ctx context.Context, _ uint, docs []Document) ([]DocumentError, error) {
-				time.Sleep(time.Second)
-				return nil, errTest
-			},
-		},
-		adapter: &mockAdapter{
-			walEventToMsgFn: func(*wal.Event) (*msg, error) {
-				return &msg{
-					write: newTestDocument(),
-				}, nil
-			},
-		},
-		skipSchema: func(schemaName string) bool { return false },
-	}
-	defer indexer.Close()
-
-	var err error
-	indexer.batchSender, err = batch.NewSender(&batch.Config{
-		BatchTimeout: 100 * time.Millisecond,
-		MaxBatchSize: 10,
-	}, indexer.sendBatch, indexer.logger)
-	require.NoError(t, err)
-
-	doneChan := make(chan struct{}, 1)
-	go func() {
-		err := indexer.batchSender.Send(context.Background())
-		require.ErrorIs(t, err, errTest)
-		doneChan <- struct{}{}
-		close(doneChan)
-	}()
-
-	timer := time.NewTimer(5 * time.Second)
-	defer timer.Stop()
-	var processErr error
-	for {
-		select {
-		case <-doneChan:
-			require.ErrorIs(t, processErr, errTest)
-			return
-		case <-timer.C:
-			t.Error("test timeout")
-			return
-		default:
-			processErr = indexer.ProcessWALEvent(context.Background(), &wal.Event{
-				CommitPosition: wal.CommitPosition("1"),
-				Data: &wal.Data{
-					Action: "I",
-				},
-			})
-		}
-	}
-}