bug: reader race condition

ShatteredRealms · Dec 11, 2024 · a6b0b97 · a6b0b97
1 parent 7d8b6c5
commit a6b0b97
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/pkg/bus/bus_kafka_reader.go b/pkg/bus/bus_kafka_reader.go
@@ -44,6 +44,7 @@ func (k *kafkaBusReader[T]) FetchMessage(ctx context.Context) (*T, error) {
 	}
 	k.currentMessage = new(kafka.Message)
 
+	k.mu.Lock()
 	if k.Reader == nil {
 		k.Reader = kafka.NewReader(kafka.ReaderConfig{
 			Brokers:  k.brokers.Addresses(),
@@ -56,6 +57,7 @@ func (k *kafkaBusReader[T]) FetchMessage(ctx context.Context) (*T, error) {
 	}
 	var err error
 	*k.currentMessage, err = k.Reader.FetchMessage(ctx)
+	k.mu.Unlock()
 	if k.isResetting {
 		log.Logger.WithContext(ctx).Info("Reset started, skipping message")
 		return k.FetchMessage(ctx)