Updated the waiting to be done at the DAG level

Author: varundeepsaini

bundle/deploy/prepare.go

@@ -8,6 +8,7 @@ import (
 	"github.com/databricks/cli/bundle/deployplan"
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/databricks-sdk-go"
+	"github.com/databricks/databricks-sdk-go/apierr"
 	"github.com/databricks/databricks-sdk-go/retries"
 	"github.com/databricks/databricks-sdk-go/service/apps"
 )
@@ -29,6 +30,12 @@ func (r *ResourceApp) DoRead(ctx context.Context, id string) (*apps.App, error)
 }
 
 func (r *ResourceApp) DoCreate(ctx context.Context, config *apps.App) (string, *apps.App, error) {
+	// Wait for any existing app with the same name to be deleted.
+	err := r.waitForDeletion(ctx, config.Name)
+	if err != nil {
+		return "", nil, err
+	}
+
 	request := apps.CreateAppRequest{
 		App:             *config,
 		NoCompute:       true,
@@ -74,6 +81,30 @@ func (r *ResourceApp) WaitAfterCreate(ctx context.Context, config *apps.App) (*a
 	return r.waitForApp(ctx, r.client, config.Name)
 }
 
+// waitForDeletion waits for an app to be deleted if it exists and is in DELETING state.
+func (r *ResourceApp) waitForDeletion(ctx context.Context, name string) error {
+	retrier := retries.New[struct{}](retries.WithTimeout(-1), retries.WithRetryFunc(shouldRetry))
+	_, err := retrier.Run(ctx, func(ctx context.Context) (*struct{}, error) {
+		app, err := r.client.Apps.GetByName(ctx, name)
+		if err != nil {
+			if apierr.IsMissing(err) {
+				return nil, nil
+			}
+			return nil, retries.Halt(err)
+		}
+
+		if app.ComputeStatus.State == apps.ComputeStateDeleting {
+			log.Infof(ctx, "App %s is in DELETING state, waiting for it to be deleted...", name)
+			return nil, retries.Continues("app is deleting")
+		}
+
+		// App exists but is not in DELETING state. This shouldn't happen in normal
+		// flow since we only call DoCreate for new resources. Let Create API handle it.
+		return nil, nil
+	})
+	return err
+}
+
 // waitForApp waits for the app to reach the target state. The target state is either ACTIVE or STOPPED.
 // Apps with no_compute set to true will reach the STOPPED state, otherwise they will reach the ACTIVE state.
 // We can't use the default waiter from SDK because it only waits on ACTIVE state but we need also STOPPED state.

bundle/deploy/prepare_test.go

@@ -173,7 +173,6 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand
 		permissions.ApplyWorkspaceRootPermissions(),
 		metrics.TrackUsedCompute(),
 		deploy.ResourcePathMkdir(),
-		deploy.PrepareEnvironment(),
 	)
 
 	if logdiag.HasError(ctx) {