diff --git a/controller/sync.go b/controller/sync.go
index 11139ac85bcb1..ebfe86ba3c043 100644
--- a/controller/sync.go
+++ b/controller/sync.go
@@ -484,9 +484,11 @@ func (sc *syncContext) getSyncTasks() (_ syncTasks, successful bool) {
 		serverRes, err := kube.ServerResourceForGroupVersionKind(sc.disco, task.groupVersionKind())
 		if err != nil {
 			// Special case for custom resources: if CRD is not yet known by the K8s API server,
-			// skip verification during `kubectl apply --dry-run` since we expect the CRD
+			// and the CRD is part of this sync or the resource is annotated with SkipDryRunOnMissingResource=true,
+			// then skip verification during `kubectl apply --dry-run` since we expect the CRD
 			// to be created during app synchronization.
-			if apierr.IsNotFound(err) && sc.hasCRDOfGroupKind(task.group(), task.kind()) {
+			skipDryRunOnMissingResource := resource.HasAnnotationOption(task.targetObj, common.AnnotationSyncOptions, "SkipDryRunOnMissingResource=true")
+			if apierr.IsNotFound(err) && (skipDryRunOnMissingResource || sc.hasCRDOfGroupKind(task.group(), task.kind())) {
 				sc.log.WithFields(log.Fields{"task": task}).Debug("skip dry-run for custom resource")
 				task.skipDryRun = true
 			} else {
diff --git a/controller/sync_test.go b/controller/sync_test.go
index ce866ea2b7b5c..5b6133c5b3dd2 100644
--- a/controller/sync_test.go
+++ b/controller/sync_test.go
@@ -478,6 +478,101 @@ func TestObjectsGetANamespace(t *testing.T) {
 	assert.Equal(t, "", pod.GetNamespace())
 }
 
+func TestSyncCustomResources(t *testing.T) {
+	type fields struct {
+		skipDryRunAnnotationPresent bool
+		crdAlreadyPresent           bool
+		crdInSameSync               bool
+	}
+
+	tests := []struct {
+		name        string
+		fields      fields
+		wantDryRun  bool
+		wantSuccess bool
+	}{
+
+		{"unknown crd", fields{
+			skipDryRunAnnotationPresent: false, crdAlreadyPresent: false, crdInSameSync: false,
+		}, true, false},
+		{"crd present in same sync", fields{
+			skipDryRunAnnotationPresent: false, crdAlreadyPresent: false, crdInSameSync: true,
+		}, false, true},
+		{"crd is already present in cluster", fields{
+			skipDryRunAnnotationPresent: false, crdAlreadyPresent: true, crdInSameSync: false,
+		}, true, true},
+		{"crd is already present in cluster, skip dry run annotated", fields{
+			skipDryRunAnnotationPresent: true, crdAlreadyPresent: true, crdInSameSync: false,
+		}, true, true},
+		{"unknown crd, skip dry run annotated", fields{
+			skipDryRunAnnotationPresent: true, crdAlreadyPresent: false, crdInSameSync: false,
+		}, false, true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+
+			knownCustomResourceTypes := []v1.APIResource{}
+			if tt.fields.crdAlreadyPresent {
+				knownCustomResourceTypes = append(knownCustomResourceTypes, v1.APIResource{Kind: "TestCrd", Group: "argoproj.io", Version: "v1", Namespaced: true})
+			}
+
+			syncCtx := newTestSyncCtx(
+				&v1.APIResourceList{
+					GroupVersion: "argoproj.io/v1",
+					APIResources: knownCustomResourceTypes,
+				},
+				&v1.APIResourceList{
+					GroupVersion: "apiextensions.k8s.io/v1beta1",
+					APIResources: []v1.APIResource{
+						{Kind: "CustomResourceDefinition", Group: "apiextensions.k8s.io", Version: "v1beta1", Namespaced: true},
+					},
+				},
+			)
+
+			cr := test.Unstructured(`
+{
+  "apiVersion": "argoproj.io/v1",
+  "kind": "TestCrd",
+  "metadata": {
+    "name": "my-resource"
+  }
+}
+`)
+
+			if tt.fields.skipDryRunAnnotationPresent {
+				cr.SetAnnotations(map[string]string{common.AnnotationSyncOptions: "SkipDryRunOnMissingResource=true"})
+			}
+
+			resources := []managedResource{{Target: cr}}
+			if tt.fields.crdInSameSync {
+				resources = append(resources, managedResource{Target: test.NewCRD()})
+			}
+
+			syncCtx.compareResult = &comparisonResult{managedResources: resources}
+
+			tasks, successful := syncCtx.getSyncTasks()
+
+			if successful != tt.wantSuccess {
+				t.Errorf("successful = %v, want: %v", successful, tt.wantSuccess)
+				return
+			}
+
+			skipDryRun := false
+			for _, task := range tasks {
+				if task.targetObj.GetKind() == cr.GetKind() {
+					skipDryRun = task.skipDryRun
+					break
+				}
+			}
+
+			if tt.wantDryRun != !skipDryRun {
+				t.Errorf("dryRun = %v, want: %v", !skipDryRun, tt.wantDryRun)
+			}
+		})
+	}
+
+}
+
 func TestPersistRevisionHistory(t *testing.T) {
 	app := newFakeApp()
 	app.Status.OperationState = nil
diff --git a/docs/user-guide/sync-options.md b/docs/user-guide/sync-options.md
index f11f227d02d30..93db6f7a49944 100644
--- a/docs/user-guide/sync-options.md
+++ b/docs/user-guide/sync-options.md
@@ -38,3 +38,22 @@ metadata:
 
 If you want to exclude a whole class of objects globally, consider setting `resource.customizations` in [system level configuation](../user-guide/diffing.md#system-level-configuration). 
     
+## Skip Dry Run for new custom resources types
+
+>v1.6
+
+When syncing a custom resource which is not yet known to the cluster, there are generally two options:
+
+1) The CRD manifest is part of the same sync. Then ArgoCD will automatically skip the dry run, the CRD will be applied and the resource can be created.
+2) In some cases the CRD is not part of the sync, but it could be created in another way, e.g. by a controller in the cluster. An example is [gatekeeper](https://github.com/open-policy-agent/gatekeeper),
+which creates CRDs in response to user defined `ConstraintTemplates`. ArgoCD cannot find the CRD in the sync and will fail with the error `the server could not find the requested resource`.
+
+To skip the dry run for missing resource types, use the following annotation:
+
+```yaml
+metadata:
+  annotations:
+    argocd.argoproj.io/sync-options: SkipDryRunOnMissingResource=true
+```
+
+The dry run will still be executed if the CRD is already present in the cluster.