fixed test, replace semaphore with mutex

internal/pkg/bulk/bulk_test.go

@@ -314,7 +314,8 @@ func TestCancelCtxChildBulker(t *testing.T) {
 func TestCancelCtxChildBulkerReplaced(t *testing.T) {
 	bulker := NewBulker(nil, nil)
 
-	ctx, _ := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
 	ctx = testlog.SetLogger(t).WithContext(ctx)
 
 	logger := testlog.SetLogger(t)
@@ -345,13 +346,12 @@ func TestCancelCtxChildBulkerReplaced(t *testing.T) {
 	go func() {
 		defer wg.Done()
 
-		_, err := childBulker.APIKeyAuth(ctx, apikey.APIKey{})
+		err := childBulker.APIKeyUpdate(ctx, "", "", make([]byte, 0))
 
 		t.Log(err)
-		// TODO not context canceled error
-		// if !errors.Is(err, context.Canceled) {
-		// 	t.Error("Expected context cancel err: ", err)
-		// }
+		if !errors.Is(err, context.Canceled) {
+			t.Error("Expected context cancel err: ", err)
+		}
 	}()
 
 	wg.Wait()

internal/pkg/bulk/engine.go

@@ -88,7 +88,7 @@ type Bulker struct {
 	remoteOutputConfigMap map[string]map[string]interface{}
 	bulkerMap             map[string]Bulk
 	cancelFn              context.CancelFunc
-	remoteOutputLimit     *semaphore.Weighted
+	remoteOutputMutex   sync.RWMutex
 }
 
 const (
@@ -119,7 +119,6 @@ func NewBulker(es esapi.Transport, tracer *apm.Tracer, opts ...BulkOpt) *Bulker
 		remoteOutputConfigMap: make(map[string]map[string]interface{}),
 		// remote ES bulkers
 		bulkerMap:         make(map[string]Bulk),
-		remoteOutputLimit: semaphore.NewWeighted(1),
 	}
 }
 
@@ -137,10 +136,8 @@ func (b *Bulker) CancelFn() context.CancelFunc {
 
 func (b *Bulker) updateBulkerMap(ctx context.Context, outputName string, newBulker *Bulker) error {
 	// concurrency control of updating map
-	if err := b.remoteOutputLimit.Acquire(ctx, 1); err != nil {
-		return err
-	}
-	defer b.remoteOutputLimit.Release(1)
+	b.remoteOutputMutex.Lock()
+	defer b.remoteOutputMutex.Unlock()
 
 	b.bulkerMap[outputName] = newBulker
 	return nil