Merge branch 'main' into dapr-state-store-clickhouse

Author: middt

.github/workflows/certification.yml renamed to .github/workflows/certification.yml.disabled

@@ -247,12 +247,6 @@ func (aeh *AzureEventHubs) Subscribe(subscribeCtx context.Context, config Subscr
 	}
 	topic := config.Topic
 
-	// Get the processor client
-	processor, err := aeh.getProcessorForTopic(subscribeCtx, topic)
-	if err != nil {
-		return fmt.Errorf("error trying to establish a connection: %w", err)
-	}
-
 	// This component has built-in retries because Event Hubs doesn't support N/ACK for messages
 	retryHandler := func(ctx context.Context, events []*azeventhubs.ReceivedEventData) ([]HandlerResponseItem, error) {
 		b := aeh.backOffConfig.NewBackOffWithContext(ctx)
@@ -282,51 +276,58 @@ func (aeh *AzureEventHubs) Subscribe(subscribeCtx context.Context, config Subscr
 
 	subscriptionLoopFinished := make(chan bool, 1)
 
-	// Process all partition clients as they come in
-	subscriberLoop := func() {
-		for {
-			// This will block until a new partition client is available
-			// It returns nil if processor.Run terminates or if the context is canceled
-			partitionClient := processor.NextPartitionClient(subscribeCtx)
-			if partitionClient == nil {
-				subscriptionLoopFinished <- true
-				return
-			}
-			aeh.logger.Debugf("Received client for partition %s", partitionClient.PartitionID())
-
-			// Once we get a partition client, process the events in a separate goroutine
-			go func() {
-				processErr := aeh.processEvents(subscribeCtx, partitionClient, retryConfig)
-				// Do not log context.Canceled which happens at shutdown
-				if processErr != nil && !errors.Is(processErr, context.Canceled) {
-					aeh.logger.Errorf("Error processing events from partition client: %v", processErr)
-				}
-			}()
-		}
-	}
-
-	// Start the processor
+	// Start the subscribe + processor loop
 	go func() {
 		for {
-			go subscriberLoop()
-			// This is a blocking call that runs until the context is canceled
-			err = processor.Run(subscribeCtx)
-			// Exit if the context is canceled
-			if err != nil && errors.Is(err, context.Canceled) {
-				return
-			}
+			// Get the processor client
+			processor, err := aeh.getProcessorForTopic(subscribeCtx, topic)
 			if err != nil {
-				aeh.logger.Errorf("Error from event processor: %v", err)
+				aeh.logger.Errorf("error trying to establish a connection: %w", err)
 			} else {
-				aeh.logger.Debugf("Event processor terminated without error")
-			}
-			// wait for subscription loop finished signal
-			select {
-			case <-subscribeCtx.Done():
-				return
-			case <-subscriptionLoopFinished:
-				// noop
+				// Process all partition clients as they come in
+				subscriberLoop := func() {
+					for {
+						// This will block until a new partition client is available
+						// It returns nil if processor.Run terminates or if the context is canceled
+						partitionClient := processor.NextPartitionClient(subscribeCtx)
+						if partitionClient == nil {
+							subscriptionLoopFinished <- true
+							return
+						}
+						aeh.logger.Debugf("Received client for partition %s", partitionClient.PartitionID())
+
+						// Once we get a partition client, process the events in a separate goroutine
+						go func() {
+							processErr := aeh.processEvents(subscribeCtx, partitionClient, retryConfig)
+							// Do not log context.Canceled which happens at shutdown
+							if processErr != nil && !errors.Is(processErr, context.Canceled) {
+								aeh.logger.Errorf("Error processing events from partition client: %v", processErr)
+							}
+						}()
+					}
+				}
+
+				go subscriberLoop()
+				// This is a blocking call that runs until the context is canceled or a non-recoverable error is returned.
+				err = processor.Run(subscribeCtx)
+				// Exit if the context is canceled
+				if err != nil && errors.Is(err, context.Canceled) {
+					return
+				}
+				if err != nil {
+					aeh.logger.Errorf("Error from event processor: %v", err)
+				} else {
+					aeh.logger.Debugf("Event processor terminated without error")
+				}
+				// wait for subscription loop finished signal
+				select {
+				case <-subscribeCtx.Done():
+					return
+				case <-subscriptionLoopFinished:
+					// noop
+				}
 			}
+
 			// Waiting here is not strictly necessary, however, we will wait for a short time to increase the likelihood of transient errors having disappeared
 			select {
 			case <-subscribeCtx.Done():

common/component/azure/eventhubs/eventhubs.go

@@ -214,21 +214,21 @@ func ParseClientFromProperties(properties map[string]string, componentType metad
 	// start the token refresh goroutine
 
 	if settings.UseEntraID {
-		StartEntraIDTokenRefreshBackgroundRoutine(c, settings.Username, *tokenExpires, tokenCredential, ctx, logger)
+		StartEntraIDTokenRefreshBackgroundRoutine(c, settings.Username, *tokenExpires, tokenCredential, logger)
 	}
 	return c, &settings, nil
 }
 
-func StartEntraIDTokenRefreshBackgroundRoutine(client RedisClient, username string, nextExpiration time.Time, cred *azcore.TokenCredential, parentCtx context.Context, logger *kitlogger.Logger) {
+func StartEntraIDTokenRefreshBackgroundRoutine(client RedisClient, username string, nextExpiration time.Time, cred *azcore.TokenCredential, logger *kitlogger.Logger) {
 	go func(cred *azcore.TokenCredential, username string, logger *kitlogger.Logger) {
-		ctx, cancel := context.WithCancel(parentCtx)
+		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
 		backoffConfig := kitretry.DefaultConfig()
 		backoffConfig.MaxRetries = 3
 		backoffConfig.Policy = kitretry.PolicyExponential
 
 		var backoffManager backoff.BackOff
-		const refreshGracePeriod = 2 * time.Minute
+		const refreshGracePeriod = 5 * time.Minute
 		tokenRefreshDuration := time.Until(nextExpiration.Add(-refreshGracePeriod))
 
 		(*logger).Debugf("redis client: starting entraID token refresh loop")

common/component/redis/redis.go

@@ -15,12 +15,12 @@ package workflows
 
 import (
 	"context"
-	"encoding/json"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/wrapperspb"
 
 	"github.com/dapr/kit/logger"
 
@@ -61,14 +61,12 @@ func ConformanceTests(t *testing.T, props map[string]string, workflowItem workfl
 	t.Run("start", func(t *testing.T) {
 		testLogger.Info("Start test running...")
 
-		inputBytes, _ := json.Marshal(10) // Time that the activity within the workflow runs for
-
 		testInstanceID := "TestID"
 		t.Run("start", func(t *testing.T) {
 			req := &workflows.StartRequest{
-				InstanceID:    testInstanceID,
+				InstanceID:    &testInstanceID,
 				WorkflowName:  "TestWorkflow",
-				WorkflowInput: inputBytes,
+				WorkflowInput: wrapperspb.String("10"),
 				Options: map[string]string{
 					"task_queue": "TestTaskQueue",
 				},

tests/conformance/workflows/workflows.go

@@ -1,11 +1,13 @@
 package workflows
 
+import "google.golang.org/protobuf/types/known/wrapperspb"
+
 // StartRequest is the struct describing a start workflow request.
 type StartRequest struct {
-	InstanceID    string            `json:"instanceID"`
-	Options       map[string]string `json:"options"`
-	WorkflowName  string            `json:"workflowName"`
-	WorkflowInput []byte            `json:"workflowInput"`
+	InstanceID    *string                 `json:"instanceID"`
+	Options       map[string]string       `json:"options"`
+	WorkflowName  string                  `json:"workflowName"`
+	WorkflowInput *wrapperspb.StringValue `json:"workflowInput"`
 }
 
 // GetRequest is the struct describing a get workflow state request.
@@ -16,14 +18,14 @@ type GetRequest struct {
 // TerminateRequest is the struct describing a terminate workflow request.
 type TerminateRequest struct {
 	InstanceID string `json:"instanceID"`
-	Recursive  bool   `json:"recursive"`
+	Recursive  *bool  `json:"recursive"`
 }
 
 // RaiseEventRequest is the struct describing a raise workflow event request.
 type RaiseEventRequest struct {
-	InstanceID string `json:"instanceID"`
-	EventName  string `json:"name"`
-	EventData  []byte `json:"data"`
+	InstanceID string                  `json:"instanceID"`
+	EventName  string                  `json:"name"`
+	EventData  *wrapperspb.StringValue `json:"data"`
 }
 
 // PauseRequest is the struct describing a pause workflow request.
@@ -39,5 +41,5 @@ type ResumeRequest struct {
 // PurgeRequest is the object describing a Purge request.
 type PurgeRequest struct {
 	InstanceID string `json:"instanceID"`
-	Recursive  bool   `json:"recursive"`
+	Recursive  *bool  `json:"recursive"`
 }