address most comments except for max poller settings

Author: shijiesheng

internal/internal_worker.go

@@ -1290,6 +1290,7 @@ func AugmentWorkerOptions(options WorkerOptions) WorkerOptions {
 }
 
 func augmentAutoScalerOptions(options AutoScalerOptions) AutoScalerOptions {
+	// need at least 2 for sticky task polling to work
 	if options.PollerMinCount <= 1 {
 		options.PollerMinCount = defaultMinPollerSize
 	}

internal/internal_worker_base.go

@@ -38,7 +38,6 @@ import (
 	"github.com/jonboulle/clockwork"
 	"github.com/uber-go/tally"
 	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
 	"golang.org/x/time/rate"
 
 	"go.uber.org/cadence/.gen/go/shared"
@@ -168,7 +167,7 @@ func createPollRetryPolicy() backoff.RetryPolicy {
 
 func newBaseWorker(options baseWorkerOptions, logger *zap.Logger, metricsScope tally.Scope, sessionTokenBucket *sessionTokenBucket) *baseWorker {
 	ctx, cancel := context.WithCancel(context.Background())
-	logger = logger.With(zapcore.Field{Key: tagWorkerType, Type: zapcore.StringType, String: options.workerType})
+	logger = logger.With(zap.String(tagWorkerType, options.workerType))
 	metricsScope = tagScope(metricsScope, tagWorkerType, options.workerType)
 
 	concurrency := &worker.ConcurrencyLimit{

internal/internal_worker_test.go

@@ -1551,10 +1551,12 @@ func TestGetTaskAutoConfigHint(t *testing.T) {
 		},
 	} {
 		t.Run(tt.name, func(t *testing.T) {
-			if task, ok := tt.task.(autoConfigHintAwareTask); ok {
-				assert.Equal(t, tt.want, task.getAutoConfigHint())
+			task, ok := tt.task.(autoConfigHintAwareTask)
+			if tt.want == nil {
+				assert.False(t, ok)
 			} else {
-				assert.Nil(t, tt.want)
+				assert.True(t, ok)
+				assert.Equal(t, tt.want, task.getAutoConfigHint())
 			}
 		})
 	}

internal/worker/concurrency_auto_scaler.go

@@ -48,7 +48,6 @@ const (
 	autoScalerEventStart                      autoScalerEvent = "auto-scaler-start"
 	autoScalerEventStop                       autoScalerEvent = "auto-scaler-stop"
 	autoScalerEventLogMsg                     string          = "concurrency auto scaler event"
-	testTimeFormat                            string          = "15:04:05"
 
 	metricsEnabled        = "enabled"
 	metricsDisabled       = "disabled"
@@ -136,12 +135,12 @@ func (c *ConcurrencyAutoScaler) Start() {
 	c.wg.Add(1)
 
 	go func() {
+		defer c.wg.Done()
 		defer func() {
 			if r := recover(); r != nil {
 				c.log.Error("panic in concurrency auto scaler, stopping the auto scaler", zap.Any("error", r))
 			}
 		}()
-		defer c.wg.Done()
 		ticker := c.clock.NewTicker(c.updateTick)
 		defer ticker.Stop()
 		for {
@@ -162,9 +161,11 @@ func (c *ConcurrencyAutoScaler) Stop() {
 	if c == nil {
 		return // no-op if auto scaler is not set
 	}
+	c.lock.Lock()
+	c.logEvent(autoScalerEventStop)
+	c.lock.Unlock()
 	close(c.shutdownChan)
 	c.wg.Wait()
-	c.logEvent(autoScalerEventStop)
 }
 
 // ProcessPollerHint reads the poller response hint and take actions in a transactional way

internal/worker/concurrency_auto_scaler_test.go

@@ -39,7 +39,8 @@ import (
 )
 
 const (
-	testTickTime = 1 * time.Second
+	testTickTime   = 1 * time.Second
+	testTimeFormat = time.TimeOnly
 )
 
 func createTestConcurrencyAutoScaler(t *testing.T, logger *zap.Logger, clock clockwork.Clock) *ConcurrencyAutoScaler {
@@ -303,11 +304,12 @@ func TestConcurrencyAutoScaler(t *testing.T) {
 			for _, event := range obs.FilterMessage(autoScalerEventLogMsg).All() {
 				if event.ContextMap()["event"] != string(autoScalerEventEmitMetrics) {
 					t.Log("event: ", event.ContextMap())
+
 					actualEvents = append(actualEvents, eventLog{
 						eventType:   autoScalerEvent(event.ContextMap()["event"].(string)),
 						enabled:     event.ContextMap()["enabled"].(bool),
 						pollerQuota: event.ContextMap()["poller_quota"].(int64),
-						time:        event.ContextMap()["time"].(time.Time).Format(testTimeFormat),
+						time:        event.Time.Format(testTimeFormat),
 					})
 				}
 			}