diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index 27b2c35af1..348ce85ba7 100644
--- a/pkg/controller/controller.go
+++ b/pkg/controller/controller.go
@@ -43,9 +43,12 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	storagelistersv1 "k8s.io/client-go/listers/storage/v1"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/klog"
 	"sigs.k8s.io/sig-storage-lib-external-provisioner/v5/controller"
 	"sigs.k8s.io/sig-storage-lib-external-provisioner/v5/util"
@@ -222,6 +225,7 @@ type csiProvisioner struct {
 	claimLister                           corelisters.PersistentVolumeClaimLister
 	vaLister                              storagelistersv1.VolumeAttachmentLister
 	extraCreateMetadata                   bool
+	eventRecorder                         record.EventRecorder
 }
 
 var _ controller.Provisioner = &csiProvisioner{}
@@ -289,6 +293,10 @@ func NewCSIProvisioner(client kubernetes.Interface,
 	vaLister storagelistersv1.VolumeAttachmentLister,
 	extraCreateMetadata bool,
 ) controller.Provisioner {
+	broadcaster := record.NewBroadcaster()
+	broadcaster.StartLogging(klog.Infof)
+	broadcaster.StartRecordingToSink(&corev1.EventSinkImpl{Interface: client.CoreV1().Events(v1.NamespaceAll)})
+	eventRecorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: fmt.Sprintf("external-provisioner")})
 
 	csiClient := csi.NewControllerClient(grpcClient)
 	provisioner := &csiProvisioner{
@@ -312,6 +320,7 @@ func NewCSIProvisioner(client kubernetes.Interface,
 		claimLister:                           claimLister,
 		vaLister:                              vaLister,
 		extraCreateMetadata:                   extraCreateMetadata,
+		eventRecorder:                         eventRecorder,
 	}
 	return provisioner
 }
@@ -475,7 +484,11 @@ func (p *csiProvisioner) ProvisionExt(options controller.ProvisionOptions) (*v1.
 		case pvcKind:
 			rc.clone = true
 		default:
-			klog.Infof("Unsupported DataSource specified (%s), the provisioner won't act on this request", options.PVC.Spec.DataSource.Kind)
+			klog.Infof("DataSource specified (%s) is not supported by the provisioner, waiting for an external data populator to create the volume", options.PVC.Spec.DataSource.Kind)
+			// DataSource is not VolumeSnapshot and PVC
+			// Wait for an external data populator to create the volume
+			p.eventRecorder.Event(options.PVC, v1.EventTypeNormal, "Provisioning", fmt.Sprintf("Waiting for a volume to be created by an external data populator"))
+			return nil, controller.ProvisioningFinished, nil
 		}
 	}
 	if err := p.checkDriverCapabilities(rc); err != nil {
diff --git a/pkg/controller/controller_test.go b/pkg/controller/controller_test.go
index 835406fd79..38b6a590e9 100644
--- a/pkg/controller/controller_test.go
+++ b/pkg/controller/controller_test.go
@@ -407,8 +407,11 @@ func TestCreateDriverReturnsInvalidCapacityDuringProvision(t *testing.T) {
 	defer mockController.Finish()
 	defer driver.Stop()
 
+	var clientSetObjects []runtime.Object
+	clientSet := fakeclientset.NewSimpleClientset(clientSetObjects...)
+
 	pluginCaps, controllerCaps := provisionCapabilities()
-	csiProvisioner := NewCSIProvisioner(nil, 5*time.Second, "test-provisioner", "test",
+	csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test",
 		5, csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, nil, nil, false)
 
 	// Requested PVC with requestedBytes storage
@@ -808,6 +811,7 @@ type provisioningTestcase struct {
 	expectState       controller.ProvisioningState
 	expectCreateVolDo interface{}
 	withExtraMetadata bool
+	skipCreateVolume  bool
 }
 
 type pvSpec struct {
@@ -864,6 +868,7 @@ func getDefaultSecretObjects() []runtime.Object {
 func TestProvision(t *testing.T) {
 	var requestedBytes int64 = 100
 	deletePolicy := v1.PersistentVolumeReclaimDelete
+	apiGrp := "my.example.io"
 	testcases := map[string]provisioningTestcase{
 		"normal provision": {
 			volOpts: controller.ProvisionOptions{
@@ -1623,6 +1628,37 @@ func TestProvision(t *testing.T) {
 				},
 			},
 		},
+		"provision with any volume data source": {
+			volOpts: controller.ProvisionOptions{
+				StorageClass: &storagev1.StorageClass{
+					ReclaimPolicy: &deletePolicy,
+					Parameters:    map[string]string{},
+					Provisioner:   "test-driver",
+				},
+				PVC: &v1.PersistentVolumeClaim{
+					ObjectMeta: metav1.ObjectMeta{
+						UID:         "testid",
+						Annotations: driverNameAnnotation,
+					},
+					Spec: v1.PersistentVolumeClaimSpec{
+						Resources: v1.ResourceRequirements{
+							Requests: v1.ResourceList{
+								v1.ResourceName(v1.ResourceStorage): resource.MustParse(strconv.FormatInt(requestedBytes, 10)),
+							},
+						},
+						AccessModes: []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
+						DataSource: &v1.TypedLocalObjectReference{
+							Name:     "testPopulator",
+							Kind:     "MyPopulator",
+							APIGroup: &apiGrp,
+						},
+					},
+				},
+			},
+			expectState:      controller.ProvisioningFinished,
+			expectErr:        false,
+			skipCreateVolume: true,
+		},
 	}
 
 	for k, tc := range testcases {
@@ -1703,7 +1739,7 @@ func runProvisionTest(t *testing.T, k string, tc provisioningTestcase, requested
 		controllerServer.EXPECT().CreateVolume(gomock.Any(), gomock.Any()).Do(tc.expectCreateVolDo).Return(out, tc.createVolumeError).Times(1)
 	} else {
 		// Setup regular mock call expectations.
-		if !tc.expectErr {
+		if !tc.expectErr && !tc.skipCreateVolume {
 			controllerServer.EXPECT().CreateVolume(gomock.Any(), gomock.Any()).Return(out, tc.createVolumeError).Times(1)
 		}
 	}