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

Author: dapr-bot

pubsub/aws/snssqs/subscription_mgmt.go

@@ -118,7 +118,11 @@ func (sm *SubscriptionManager) queueConsumerController(queueConsumerCbk func(con
 
 			sm.lock.Unlock()
 		case <-sm.closeCh:
-			return
+			if sm.topicsHandlers.Size() == 0 {
+				return
+			} else {
+				sm.logger.Info("Shutdown initiated, waiting for all subscriptions to be cleaned up")
+			}
 		}
 	}
 }

tests/certification/bindings/zeebe/command/create_instance_test.go

@@ -290,6 +290,7 @@ func TestCreateInstanceOperation(t *testing.T) {
 	flow.New(t, "Test create instance operation (async)").
 		Step(dockercompose.Run("zeebe", zeebe_test.DockerComposeYaml)).
 		Step("Waiting for Zeebe Readiness...", retry.Do(time.Second*3, 10, zeebe_test.CheckZeebeConnection)).
+		Step(app.Run("workerApp", fmt.Sprintf(":%d", appPort), workers(0))).
 		Step(sidecar.Run(zeebe_test.SidecarName,
 			append(componentRuntimeOptions(),
 				embedded.WithAppProtocol(protocol.HTTPProtocol, strconv.Itoa(appPort)),
@@ -298,7 +299,6 @@ func TestCreateInstanceOperation(t *testing.T) {
 				embedded.WithResourcesPath("components/standard"),
 			)...,
 		)).
-		Step(app.Run("workerApp", fmt.Sprintf(":%d", appPort), workers(0))).
 		Step("Waiting for the component to start", flow.Sleep(10*time.Second)).
 		Step("Deploy process in version 1", deployVersion1).
 		Step("Deploy process in version 2", deployVersion2).
@@ -314,6 +314,7 @@ func TestCreateInstanceOperation(t *testing.T) {
 	flow.New(t, "Test create instance operation (sync)").
 		Step(dockercompose.Run("zeebe", zeebe_test.DockerComposeYaml)).
 		Step("Waiting for Zeebe Readiness...", retry.Do(time.Second*3, 10, zeebe_test.CheckZeebeConnection)).
+		Step(app.Run("workerApp", fmt.Sprintf(":%d", appPort), workers(20*time.Second))).
 		Step(sidecar.Run(zeebe_test.SidecarName,
 			append(componentRuntimeOptions(),
 				embedded.WithAppProtocol(protocol.HTTPProtocol, strconv.Itoa(appPort)),
@@ -322,7 +323,6 @@ func TestCreateInstanceOperation(t *testing.T) {
 				embedded.WithResourcesPath("components/syncProcessCreation"),
 			)...,
 		)).
-		Step(app.Run("workerApp", fmt.Sprintf(":%d", appPort), workers(20*time.Second))).
 		Step("Waiting for the component to start", flow.Sleep(10*time.Second)).
 		Step("Deploy process in version 1", deployVersion1).
 		Step("Deploy process in version 2", deployVersion2).

tests/certification/flow/sidecar/sidecar.go

@@ -18,6 +18,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/cenkalti/backoff"
 	"github.com/dapr/dapr/pkg/runtime"
 	"github.com/dapr/dapr/pkg/runtime/registry"
 	"github.com/dapr/kit/logger"
@@ -133,6 +134,21 @@ func (s Sidecar) Start(ctx flow.Context) error {
 		options.clientCallback(&client)
 	}
 
+	// Wait for the sidecar to be healthy
+	var bo backoff.BackOff = backoff.NewConstantBackOff(100 * time.Millisecond)
+	bo = backoff.WithMaxRetries(bo, 200) // 20 seconds
+	bo = backoff.WithContext(bo, ctx)
+	retryErr := backoff.Retry(func() error {
+		if !rtConf.Healthz.IsReady() {
+			return fmt.Errorf("sidecar is not ready")
+		}
+		return nil
+	}, bo)
+
+	if retryErr != nil {
+		return retryErr
+	}
+
 	return nil
 }