diff --git a/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/azure_vmssclient.go b/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/azure_vmssclient.go
index 76922336b91..145c92bb2cf 100644
--- a/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/azure_vmssclient.go
+++ b/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/azure_vmssclient.go
@@ -434,6 +434,53 @@ func (c *Client) DeleteInstances(ctx context.Context, resourceGroupName string,
 	return nil
 }
 
+// DeleteInstancesAsync sends the delete request to ARM client and DOEST NOT wait on the future
+func (c *Client) DeleteInstancesAsync(ctx context.Context, resourceGroupName string, vmScaleSetName string, vmInstanceIDs compute.VirtualMachineScaleSetVMInstanceRequiredIDs) (*azure.Future, *retry.Error) {
+	mc := metrics.NewMetricContext("vmss", "delete_instances_async", resourceGroupName, c.subscriptionID, "")
+
+	// Report errors if the client is rate limited.
+	if !c.rateLimiterWriter.TryAccept() {
+		mc.RateLimitedCount()
+		return nil, retry.GetRateLimitError(true, "VMSSDeleteInstancesAsync")
+	}
+
+	// Report errors if the client is throttled.
+	if c.RetryAfterWriter.After(time.Now()) {
+		mc.ThrottledCount()
+		rerr := retry.GetThrottlingError("VMSSDeleteInstancesAsync", "client throttled", c.RetryAfterWriter)
+		return nil, rerr
+	}
+
+	resourceID := armclient.GetResourceID(
+		c.subscriptionID,
+		resourceGroupName,
+		"Microsoft.Compute/virtualMachineScaleSets",
+		vmScaleSetName,
+	)
+
+	response, rerr := c.armClient.PostResource(ctx, resourceID, "delete", vmInstanceIDs)
+	defer c.armClient.CloseResponse(ctx, response)
+
+	if rerr != nil {
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmss.deletevms.request", resourceID, rerr.Error())
+		return nil, rerr
+	}
+
+	err := autorest.Respond(response, azure.WithErrorUnlessStatusCode(http.StatusOK, http.StatusAccepted))
+	if err != nil {
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmss.deletevms.respond", resourceID, rerr.Error())
+		return nil, retry.GetError(response, err)
+	}
+
+	future, err := azure.NewFutureFromResponse(response)
+	if err != nil {
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmss.deletevms.future", resourceID, rerr.Error())
+		return nil, retry.NewError(false, err)
+	}
+
+	return &future, nil
+}
+
 // deleteVMSSInstances deletes the instances for a VirtualMachineScaleSet.
 func (c *Client) deleteVMSSInstances(ctx context.Context, resourceGroupName string, vmScaleSetName string, vmInstanceIDs compute.VirtualMachineScaleSetVMInstanceRequiredIDs) *retry.Error {
 	resourceID := armclient.GetResourceID(
diff --git a/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/azure_vmssclient_test.go b/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/azure_vmssclient_test.go
index cc6c7aeb4b3..4605d9dfd3d 100644
--- a/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/azure_vmssclient_test.go
+++ b/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/azure_vmssclient_test.go
@@ -755,6 +755,35 @@ func TestDeleteInstancesWaitError(t *testing.T) {
 	assert.NotNil(t, rerr)
 	assert.Equal(t, vmssDeleteInstancesErr, rerr)
 }
+func TestDeleteInstancesAsync(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	vmss := getTestVMSS("vmss1")
+	vmInstanceIDs := compute.VirtualMachineScaleSetVMInstanceRequiredIDs{
+		InstanceIds: &[]string{"0", "1", "2"},
+	}
+	response := &http.Response{
+		StatusCode: http.StatusOK,
+		Request:    &http.Request{Method: "POST"},
+		Body:       ioutil.NopCloser(bytes.NewReader([]byte("{}"))),
+	}
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	armClient.EXPECT().PostResource(gomock.Any(), to.String(vmss.ID), "delete", vmInstanceIDs).Return(response, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	vmssClient := getTestVMSSClient(armClient)
+	future, rerr := vmssClient.DeleteInstancesAsync(context.TODO(), "rg", "vmss1", vmInstanceIDs)
+	assert.Nil(t, rerr)
+	assert.Equal(t, future.Status(), "Succeeded")
+
+	// on error
+	retryErr := &retry.Error{RawError: fmt.Errorf("error")}
+	armClient.EXPECT().PostResource(gomock.Any(), to.String(vmss.ID), "delete", vmInstanceIDs).Return(&http.Response{StatusCode: http.StatusBadRequest}, retryErr).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+	_, rerr = vmssClient.DeleteInstancesAsync(context.TODO(), "rg", "vmss1", vmInstanceIDs)
+	assert.Equal(t, retryErr, rerr)
+}
 
 func getTestVMSS(name string) compute.VirtualMachineScaleSet {
 	return compute.VirtualMachineScaleSet{
diff --git a/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/interface.go b/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/interface.go
index 16aeb41eb42..2c971eb0ccd 100644
--- a/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/interface.go
+++ b/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/interface.go
@@ -54,4 +54,7 @@ type Interface interface {
 
 	// DeleteInstances deletes the instances for a VirtualMachineScaleSet.
 	DeleteInstances(ctx context.Context, resourceGroupName string, vmScaleSetName string, vmInstanceIDs compute.VirtualMachineScaleSetVMInstanceRequiredIDs) *retry.Error
+
+	// DeleteInstancesAsync sends the delete request to the ARM client and DOEST NOT wait on the future
+	DeleteInstancesAsync(ctx context.Context, resourceGroupName string, vmScaleSetName string, vmInstanceIDs compute.VirtualMachineScaleSetVMInstanceRequiredIDs) (*azure.Future, *retry.Error)
 }
diff --git a/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/mockvmssclient/interface.go b/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/mockvmssclient/interface.go
index b5143764064..232cdd569c3 100644
--- a/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/mockvmssclient/interface.go
+++ b/staging/src/k8s.io/legacy-cloud-providers/azure/clients/vmssclient/mockvmssclient/interface.go
@@ -139,3 +139,18 @@ func (mr *MockInterfaceMockRecorder) DeleteInstances(ctx, resourceGroupName, vmS
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteInstances", reflect.TypeOf((*MockInterface)(nil).DeleteInstances), ctx, resourceGroupName, vmScaleSetName, vmInstanceIDs)
 }
+
+// DeleteInstancesAsync mocks base method
+func (m *MockInterface) DeleteInstancesAsync(ctx context.Context, resourceGroupName, VMScaleSetName string, vmInstanceIDs compute.VirtualMachineScaleSetVMInstanceRequiredIDs) (*azure.Future, *retry.Error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "DeleteInstancesAsync", ctx, resourceGroupName, VMScaleSetName, vmInstanceIDs)
+	ret0, _ := ret[0].(*azure.Future)
+	ret1, _ := ret[1].(*retry.Error)
+	return ret0, ret1
+}
+
+// DeleteInstancesAsync indicates an expected call of DeleteInstancesAsync
+func (mr *MockInterfaceMockRecorder) DeleteInstancesAsync(ctx, resourceGroupName, VMScaleSetName, vmInstanceIDs interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteInstancesAsync", reflect.TypeOf((*MockInterface)(nil).DeleteInstancesAsync), ctx, resourceGroupName, VMScaleSetName, vmInstanceIDs)
+}