diff --git a/builder/digitalocean/builder.go b/builder/digitalocean/builder.go
index 0532410..f9ae20a 100644
--- a/builder/digitalocean/builder.go
+++ b/builder/digitalocean/builder.go
@@ -136,7 +136,8 @@ func (b *Builder) Run(ctx context.Context, ui packersdk.Ui, hook packersdk.Hook)
 		new(stepShutdown),
 		new(stepPowerOff),
 		&stepSnapshot{
-			snapshotTimeout: b.config.SnapshotTimeout,
+			snapshotTimeout:         b.config.SnapshotTimeout,
+			waitForSnapshotTransfer: *b.config.WaitSnapshotTransfer,
 		},
 	}
 
diff --git a/builder/digitalocean/builder_acc_test.go b/builder/digitalocean/builder_acc_test.go
index d610134..b779932 100644
--- a/builder/digitalocean/builder_acc_test.go
+++ b/builder/digitalocean/builder_acc_test.go
@@ -3,8 +3,10 @@ package digitalocean
 import (
 	"context"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
+	"regexp"
 	"testing"
 
 	"github.com/digitalocean/godo"
@@ -68,6 +70,43 @@ func TestBuilderAcc_multiRegion(t *testing.T) {
 	})
 }
 
+func TestBuilderAcc_multiRegionNoWait(t *testing.T) {
+	if skip := testAccPreCheck(t); skip == true {
+		return
+	}
+	acctest.TestPlugin(t, &acctest.PluginTestCase{
+		Name:     "test-digitalocean-builder-multi-region",
+		Template: fmt.Sprintf(testBuilderAccMultiRegionNoWait, "ubuntu-20-04-x64"),
+		Check: func(buildCommand *exec.Cmd, logfile string) error {
+			if buildCommand.ProcessState != nil {
+				if buildCommand.ProcessState.ExitCode() != 0 {
+					return fmt.Errorf("Bad exit code. Logfile: %s", logfile)
+				}
+			}
+
+			logs, err := os.Open(logfile)
+			if err != nil {
+				return fmt.Errorf("Unable find %s", logfile)
+			}
+			defer logs.Close()
+
+			logsBytes, err := io.ReadAll(logs)
+			if err != nil {
+				return fmt.Errorf("Unable to read %s", logfile)
+			}
+			logsString := string(logsBytes)
+
+			notExpected := regexp.MustCompile(`Transfer to .* is complete.`)
+			matches := notExpected.FindStringSubmatch(logsString)
+			if len(matches) > 0 {
+				return fmt.Errorf("logs contains unexpected value: %v", matches)
+			}
+
+			return nil
+		},
+	})
+}
+
 func testAccPreCheck(t *testing.T) bool {
 	if os.Getenv(acctest.TestEnvVar) == "" {
 		t.Skipf("Acceptance tests skipped unless env '%s' set", acctest.TestEnvVar)
@@ -127,9 +166,7 @@ const (
 		"region": "nyc2",
 		"size": "s-1vcpu-1gb",
 		"image": "%v",
-		"ssh_username": "root",
-		"user_data": "",
-		"user_data_file": ""
+		"ssh_username": "root"
 	}]
 }
 `
@@ -142,10 +179,22 @@ const (
 		"size": "s-1vcpu-1gb",
 		"image": "%v",
 		"ssh_username": "root",
-		"user_data": "",
-		"user_data_file": "",
 		"snapshot_regions": ["nyc1", "nyc2", "nyc3"]
 	}]
 }
+`
+
+	testBuilderAccMultiRegionNoWait = `
+{
+	"builders": [{
+		"type": "digitalocean",
+		"region": "nyc2",
+		"size": "s-1vcpu-1gb",
+		"image": "%v",
+		"ssh_username": "root",
+		"snapshot_regions": ["nyc2", "nyc3"],
+		"wait_snapshot_transfer": false
+	}]
+}
 `
 )
diff --git a/builder/digitalocean/config.go b/builder/digitalocean/config.go
index 1671d71..c3da58f 100644
--- a/builder/digitalocean/config.go
+++ b/builder/digitalocean/config.go
@@ -73,9 +73,12 @@ type Config struct {
 	// appear in your account. Defaults to `packer-{{timestamp}}` (see
 	// configuration templates for more info).
 	SnapshotName string `mapstructure:"snapshot_name" required:"false"`
-	// The regions of the resulting
-	// snapshot that will appear in your account.
+	// Additional regions that resulting snapshot should be distributed to.
 	SnapshotRegions []string `mapstructure:"snapshot_regions" required:"false"`
+	// When true, Packer will block until all snapshot transfers have been completed
+	// and report errors. When false, Packer will initiate the snapshot transfers
+	// and exit successfully without waiting for completion. Defaults to true.
+	WaitSnapshotTransfer *bool `mapstructure:"wait_snapshot_transfer" required:"false"`
 	// The time to wait, as a duration string, for a
 	// droplet to enter a desired state (such as "active") before timing out. The
 	// default state timeout is "6m".
@@ -212,6 +215,10 @@ func (c *Config) Prepare(raws ...interface{}) ([]string, error) {
 		c.SnapshotTimeout = 60 * time.Minute
 	}
 
+	if c.WaitSnapshotTransfer == nil {
+		c.WaitSnapshotTransfer = godo.PtrTo(true)
+	}
+
 	if es := c.Comm.Prepare(&c.ctx); len(es) > 0 {
 		errs = packersdk.MultiErrorAppend(errs, es...)
 	}
diff --git a/builder/digitalocean/config.hcl2spec.go b/builder/digitalocean/config.hcl2spec.go
index 28ebd98..41b47e5 100644
--- a/builder/digitalocean/config.hcl2spec.go
+++ b/builder/digitalocean/config.hcl2spec.go
@@ -81,6 +81,7 @@ type FlatConfig struct {
 	IPv6                      *bool             `mapstructure:"ipv6" required:"false" cty:"ipv6" hcl:"ipv6"`
 	SnapshotName              *string           `mapstructure:"snapshot_name" required:"false" cty:"snapshot_name" hcl:"snapshot_name"`
 	SnapshotRegions           []string          `mapstructure:"snapshot_regions" required:"false" cty:"snapshot_regions" hcl:"snapshot_regions"`
+	WaitSnapshotTransfer      *bool             `mapstructure:"wait_snapshot_transfer" required:"false" cty:"wait_snapshot_transfer" hcl:"wait_snapshot_transfer"`
 	StateTimeout              *string           `mapstructure:"state_timeout" required:"false" cty:"state_timeout" hcl:"state_timeout"`
 	SnapshotTimeout           *string           `mapstructure:"snapshot_timeout" required:"false" cty:"snapshot_timeout" hcl:"snapshot_timeout"`
 	DropletName               *string           `mapstructure:"droplet_name" required:"false" cty:"droplet_name" hcl:"droplet_name"`
@@ -175,6 +176,7 @@ func (*FlatConfig) HCL2Spec() map[string]hcldec.Spec {
 		"ipv6":                         &hcldec.AttrSpec{Name: "ipv6", Type: cty.Bool, Required: false},
 		"snapshot_name":                &hcldec.AttrSpec{Name: "snapshot_name", Type: cty.String, Required: false},
 		"snapshot_regions":             &hcldec.AttrSpec{Name: "snapshot_regions", Type: cty.List(cty.String), Required: false},
+		"wait_snapshot_transfer":       &hcldec.AttrSpec{Name: "wait_snapshot_transfer", Type: cty.Bool, Required: false},
 		"state_timeout":                &hcldec.AttrSpec{Name: "state_timeout", Type: cty.String, Required: false},
 		"snapshot_timeout":             &hcldec.AttrSpec{Name: "snapshot_timeout", Type: cty.String, Required: false},
 		"droplet_name":                 &hcldec.AttrSpec{Name: "droplet_name", Type: cty.String, Required: false},
diff --git a/builder/digitalocean/step_snapshot.go b/builder/digitalocean/step_snapshot.go
index 1c56c5c..87277c8 100644
--- a/builder/digitalocean/step_snapshot.go
+++ b/builder/digitalocean/step_snapshot.go
@@ -14,7 +14,8 @@ import (
 )
 
 type stepSnapshot struct {
-	snapshotTimeout time.Duration
+	snapshotTimeout         time.Duration
+	waitForSnapshotTransfer bool
 }
 
 func (s *stepSnapshot) Run(ctx context.Context, state multistep.StateBag) multistep.StepAction {
@@ -108,14 +109,16 @@ func (s *stepSnapshot) Run(ctx context.Context, state multistep.StateBag) multis
 					return fmt.Errorf("Error transferring snapshot: %s", err)
 				}
 
-				if err := WaitForImageState(
-					godo.ActionCompleted,
-					imageId,
-					imageTransfer.ID,
-					client, 20*time.Minute); err != nil {
-					return fmt.Errorf("Error waiting for snapshot transfer: %s", err)
+				if s.waitForSnapshotTransfer {
+					if err := WaitForImageState(
+						godo.ActionCompleted,
+						imageId,
+						imageTransfer.ID,
+						client, 20*time.Minute); err != nil {
+						return fmt.Errorf("Error waiting for snapshot transfer: %s", err)
+					}
+					ui.Say(fmt.Sprintf("Transfer to %s is complete.", region))
 				}
-				ui.Say(fmt.Sprintf("Transfer to %s is complete.", region))
 
 				return nil
 			})
diff --git a/docs-partials/builder/digitalocean/Config-not-required.mdx b/docs-partials/builder/digitalocean/Config-not-required.mdx
index 7494b5d..cd1f36b 100644
--- a/docs-partials/builder/digitalocean/Config-not-required.mdx
+++ b/docs-partials/builder/digitalocean/Config-not-required.mdx
@@ -30,8 +30,11 @@
   appear in your account. Defaults to `packer-{{timestamp}}` (see
   configuration templates for more info).
 
-- `snapshot_regions` ([]string) - The regions of the resulting
-  snapshot that will appear in your account.
+- `snapshot_regions` ([]string) - Additional regions that resulting snapshot should be distributed to.
+
+- `wait_snapshot_transfer` (\*bool) - When true, Packer will block until all snapshot transfers have been completed
+  and report errors. When false, Packer will initiate the snapshot transfers
+  and exit successfully without waiting for completion. Defaults to true.
 
 - `state_timeout` (duration string | ex: "1h5m2s") - The time to wait, as a duration string, for a
   droplet to enter a desired state (such as "active") before timing out. The