Repository: kubernetes-sigs/gcp-compute-persistent-disk-csi-driver
Pull request #988 (2022-05-17 02:13:45)

Simplify node backoff logic for controller publish/unpublish op
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Follow the instructions for writing a release note: https://git.k8s.io/community/contributors/guide/release-notes.md
3. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

**What type of PR is this?**
> Uncomment only one ` /kind <>` line, hit enter to put that in a new line, and remove leading whitespaces from that line:
>
> /kind api-change
/kind bug
> /kind cleanup
> /kind design
> /kind documentation
> /kind failing-test
> /kind feature
> /kind flake

**What this PR does / why we need it**:
This PR simplifies the node backoff logic and removes the backoff queue which processes requests asynchronously.
Instead in this PR, the node is marked with a backoff condition on any new error. Similarly any successful attach or detach disk op would clear the backoff condition.  The backoff behavior scenarios are captured in this [doc](https://docs.google.com/document/d/138UQqyIbuWQ3xc_3fCXaI7MzKX_veNJe1bdWZ1ZZIcg/edit?resourcekey=0-f3uGOrMCZcmsd-v6meapFQ) for reference.

**Which issue(s) this PR fixes**:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #

**Special notes for your reviewer**:
The fix is inspired by [PR](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/pull/960), with some minor tweaks to handle all errors instead of only quota errors.

**Does this PR introduce a user-facing change?**:
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".
-->
```release-note
Simplify node backoff logic for controller publish/unpublish op
```


k8s-ci-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/pull/988#" title="Author self-approved">saikat-royc</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fgcp-compute-persistent-disk-csi-driver).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/blob/master/OWNERS)~~ [saikat-royc]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
k8s-ci-robot: @saikat-royc: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-sanity | ebf2ad5fd4a3d2b8f352358c61ea91aa20a2a836 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-sanity/1526385454662815744/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-sanity`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=988). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @saikat-royc: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-sanity | ebf2ad5fd4a3d2b8f352358c61ea91aa20a2a836 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-sanity/1526385454662815744/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-sanity`
pull-gcp-compute-persistent-disk-csi-driver-e2e | ebf2ad5fd4a3d2b8f352358c61ea91aa20a2a836 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-e2e/1526385454637649920/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=988). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
saikat-royc: /retest
k8s-ci-robot: @saikat-royc: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-e2e | ebf2ad5fd4a3d2b8f352358c61ea91aa20a2a836 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-e2e/1526385454637649920/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`
pull-gcp-compute-persistent-disk-csi-driver-sanity | ebf2ad5fd4a3d2b8f352358c61ea91aa20a2a836 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-sanity/1526391593139965952/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-sanity`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=988). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @saikat-royc: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-sanity | ebf2ad5fd4a3d2b8f352358c61ea91aa20a2a836 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-sanity/1526391593139965952/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-sanity`
pull-gcp-compute-persistent-disk-csi-driver-e2e | ebf2ad5fd4a3d2b8f352358c61ea91aa20a2a836 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-e2e/1526391593106411520/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=988). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @saikat-royc: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-sanity | ebf2ad5fd4a3d2b8f352358c61ea91aa20a2a836 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-sanity/1526391593139965952/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-sanity`
pull-gcp-compute-persistent-disk-csi-driver-e2e | ebf2ad5fd4a3d2b8f352358c61ea91aa20a2a836 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-e2e/1526391593106411520/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`
pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration | ebf2ad5fd4a3d2b8f352358c61ea91aa20a2a836 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration/1526385454750896128/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=988). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @saikat-royc: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-e2e | ebf2ad5fd4a3d2b8f352358c61ea91aa20a2a836 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-e2e/1526391593106411520/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`
pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration | ebf2ad5fd4a3d2b8f352358c61ea91aa20a2a836 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration/1526385454750896128/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration`
pull-gcp-compute-persistent-disk-csi-driver-sanity | 2a1953ce856fff056e15fd0eae00dd06cf46c1b9 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-sanity/1526688796463075328/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-sanity`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=988). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @saikat-royc: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-sanity | 2a1953ce856fff056e15fd0eae00dd06cf46c1b9 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-sanity/1526688796463075328/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-sanity`
pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration | 2a1953ce856fff056e15fd0eae00dd06cf46c1b9 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/988/pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration/1526688796542767104/) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=988). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
saikat-royc: [Issue](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/issues/990) created to track sanity test failure
amacaskill: I just had one comment on adding a comment explaining the backoff. But otherwise, /lgtm 

Patch:
diff --git a/pkg/gce-cloud-provider/compute/fake-gce.go b/pkg/gce-cloud-provider/compute/fake-gce.go
index 962079f4..309b0b39 100644
--- a/pkg/gce-cloud-provider/compute/fake-gce.go
+++ b/pkg/gce-cloud-provider/compute/fake-gce.go
@@ -504,7 +504,7 @@ func (cloud *FakeCloudProvider) UpdateDiskStatus(s string) {
 
 type FakeBlockingCloudProvider struct {
 	*FakeCloudProvider
-	ReadyToExecute chan chan struct{}
+	ReadyToExecute chan chan Signal
 }
 
 // FakeBlockingCloudProvider's method adds functionality to finely control the order of execution of CreateSnapshot calls.
@@ -512,19 +512,39 @@ type FakeBlockingCloudProvider struct {
 // The test calling this function can block on readyToExecute to ensure that the operation has started and
 // allowed the CreateSnapshot to continue by passing a struct into executeCreateSnapshot.
 func (cloud *FakeBlockingCloudProvider) CreateSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string, snapshotParams common.SnapshotParameters) (*computev1.Snapshot, error) {
-	executeCreateSnapshot := make(chan struct{})
+	executeCreateSnapshot := make(chan Signal)
 	cloud.ReadyToExecute <- executeCreateSnapshot
 	<-executeCreateSnapshot
 	return cloud.FakeCloudProvider.CreateSnapshot(ctx, project, volKey, snapshotName, snapshotParams)
 }
 
 func (cloud *FakeBlockingCloudProvider) CreateImage(ctx context.Context, project string, volKey *meta.Key, imageName string, snapshotParams common.SnapshotParameters) (*computev1.Image, error) {
-	executeCreateSnapshot := make(chan struct{})
+	executeCreateSnapshot := make(chan Signal)
 	cloud.ReadyToExecute <- executeCreateSnapshot
 	<-executeCreateSnapshot
 	return cloud.FakeCloudProvider.CreateImage(ctx, project, volKey, imageName, snapshotParams)
 }
 
+func (cloud *FakeBlockingCloudProvider) DetachDisk(ctx context.Context, project, deviceName, instanceZone, instanceName string) error {
+	execute := make(chan Signal)
+	cloud.ReadyToExecute <- execute
+	val := <-execute
+	if val.ReportError {
+		return fmt.Errorf("force mock error for DetachDisk device %s", deviceName)
+	}
+	return cloud.FakeCloudProvider.DetachDisk(ctx, project, deviceName, instanceZone, instanceName)
+}
+
+func (cloud *FakeBlockingCloudProvider) AttachDisk(ctx context.Context, project string, volKey *meta.Key, readWrite, diskType, instanceZone, instanceName string) error {
+	execute := make(chan Signal)
+	cloud.ReadyToExecute <- execute
+	val := <-execute
+	if val.ReportError {
+		return fmt.Errorf("force mock error for AttachDisk: volkey %s", volKey)
+	}
+	return cloud.FakeCloudProvider.AttachDisk(ctx, project, volKey, readWrite, diskType, instanceZone, instanceName)
+}
+
 func notFoundError() *googleapi.Error {
 	return &googleapi.Error{
 		Errors: []googleapi.ErrorItem{
@@ -544,3 +564,8 @@ func invalidError() *googleapi.Error {
 		},
 	}
 }
+
+type Signal struct {
+	ReportError   bool
+	ReportRunning bool
+}
diff --git a/pkg/gce-pd-csi-driver/controller.go b/pkg/gce-pd-csi-driver/controller.go
index 632a1a31..c65a9cec 100644
--- a/pkg/gce-pd-csi-driver/controller.go
+++ b/pkg/gce-pd-csi-driver/controller.go
@@ -31,14 +31,18 @@ import (
 	"google.golang.org/grpc/status"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/uuid"
-	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/client-go/util/workqueue"
+	"k8s.io/client-go/util/flowcontrol"
 	"k8s.io/klog"
 
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
 	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
 )
 
+const (
+	nodeBackoffInitialDuration = 200 * time.Millisecond
+	nodeBackoffMaxDuration     = 5 * time.Minute
+)
+
 type GCEControllerServer struct {
 	Driver        *GCEDriver
 	CloudProvider gce.GCECompute
@@ -54,15 +58,9 @@ type GCEControllerServer struct {
 	// Aborted error
 	volumeLocks *common.VolumeLocks
 
-	// queue is a rate limited work queue for Controller Publish/Unpublish
-	// Volume calls
-	queue workqueue.RateLimitingInterface
-
-	// publishErrorsSeenOnNode is a list of nodes with attach/detach
-	// operation failures so those nodes shall be rate limited for all
-	// the attach/detach operations until there is an attach / detach
-	// operation succeeds
-	publishErrorsSeenOnNode map[string]bool
+	// When the attacher sidecar issues controller publish/unpublish for multiple disks for a given node, the per-instance operation queue in GCE fills up causing attach/detach disk requests to immediately return with an error until the queue drains. nodeBackoff keeps track of any active backoff condition on a given node, and the time when retry of controller publish/unpublish is permissible. A node is marked with backoff when any error is encountered by the driver during controller publish/unpublish calls.
+	// If the controller eventually allows controller publish/publish requests for volumes (because the backoff time expired), and those requests fail, the next backoff retry time will be updated on every failure and capped at 'nodeBackoffMaxDuration'. Also, any successful controller publish/unpublish call will clear the backoff condition for the node.
+	nodeBackoff *flowcontrol.Backoff
 }
 
 type workItem struct {
@@ -364,73 +362,26 @@ func (gceCS *GCEControllerServer) DeleteVolume(ctx context.Context, req *csi.Del
 	return &csi.DeleteVolumeResponse{}, nil
 }
 
-// Run starts the GCEControllerServer.
-func (gceCS *GCEControllerServer) Run() {
-	go wait.Until(gceCS.worker, 1*time.Second, wait.NeverStop)
-}
-
-func (gceCS *GCEControllerServer) worker() {
-	// Runs until workqueue is shut down
-	for gceCS.processNextWorkItem() {
-	}
-}
-
-func (gceCS *GCEControllerServer) processNextWorkItem() bool {
-	item, quit := gceCS.queue.Get()
-	if quit {
-		return false
-	}
-	defer gceCS.queue.Done(item)
-
-	workItem, ok := item.(*workItem)
-	if !ok {
-		gceCS.queue.AddRateLimited(item)
-		return true
-	}
-
-	if workItem.publishReq != nil {
-		_, err := gceCS.executeControllerPublishVolume(workItem.ctx, workItem.publishReq)
-
-		if err != nil {
-			klog.Errorf("ControllerPublishVolume failed with error: %v", err)
-		}
-	}
-
-	if workItem.unpublishReq != nil {
-		_, err := gceCS.executeControllerUnpublishVolume(workItem.ctx, workItem.unpublishReq)
-
-		if err != nil {
-			klog.Errorf("ControllerUnpublishVolume failed with error: %v", err)
-		}
-	}
-
-	gceCS.queue.Forget(item)
-	return true
-}
-
 func (gceCS *GCEControllerServer) ControllerPublishVolume(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (*csi.ControllerPublishVolumeResponse, error) {
-	// Only valid requests will be queued
+	// Only valid requests will be accepted
 	_, _, err := gceCS.validateControllerPublishVolumeRequest(ctx, req)
-
 	if err != nil {
 		return nil, err
 	}
 
-	// If the node is not marked, proceed the request
-	if _, found := gceCS.publishErrorsSeenOnNode[req.NodeId]; !found {
-		return gceCS.executeControllerPublishVolume(ctx, req)
+	if gceCS.nodeBackoff.IsInBackOffSinceUpdate(req.NodeId, gceCS.nodeBackoff.Clock.Now()) {
+		return nil, status.Errorf(codes.Unavailable, "ControllerPublish not permitted on node %q due to backoff condition", req.NodeId)
 	}
 
-	// Node is marked so queue up the request. Note the original gRPC context may get canceled,
-	// so a new one is created here.
-	//
-	// Note that the original context probably has a timeout (see csiAttach in external-attacher),
-	// which is ignored.
-	gceCS.queue.AddRateLimited(&workItem{
-		ctx:        context.Background(),
-		publishReq: req,
-	})
-	return nil, status.Error(codes.Unavailable, "Request queued due to error condition on node")
+	resp, err := gceCS.executeControllerPublishVolume(ctx, req)
+	if err != nil {
+		klog.Infof("For node %s adding backoff due to error for volume %s", req.NodeId, req.VolumeId)
+		gceCS.nodeBackoff.Next(req.NodeId, gceCS.nodeBackoff.Clock.Now())
+	} else {
+		klog.Infof("For node %s clear backoff due to successful publish of volume %v", req.NodeId, req.VolumeId)
+		gceCS.nodeBackoff.Reset(req.NodeId)
+	}
+	return resp, err
 }
 
 func (gceCS *GCEControllerServer) validateControllerPublishVolumeRequest(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (string, *meta.Key, error) {
@@ -542,15 +493,8 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 
 	err = gceCS.CloudProvider.WaitForAttach(ctx, project, volKey, instanceZone, instanceName)
 	if err != nil {
-		// Mark the node and rate limit all the following attach/detach
-		// operations for this node
-		gceCS.publishErrorsSeenOnNode[nodeID] = true
 		return nil, status.Error(codes.Internal, fmt.Sprintf("unknown WaitForAttach error: %v", err))
 	}
-
-	// Attach succeeds so unmark the node
-	delete(gceCS.publishErrorsSeenOnNode, nodeID)
-
 	klog.V(4).Infof("ControllerPublishVolume succeeded for disk %v to instance %v", volKey, nodeID)
 	return pubVolResp, nil
 }
@@ -558,23 +502,23 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 func (gceCS *GCEControllerServer) ControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error) {
 	// Only valid requests will be queued
 	_, _, err := gceCS.validateControllerUnpublishVolumeRequest(ctx, req)
-
 	if err != nil {
 		return nil, err
 	}
 
-	// If the node is not marked, proceed the request
-	if _, found := gceCS.publishErrorsSeenOnNode[req.NodeId]; !found {
-		return gceCS.executeControllerUnpublishVolume(ctx, req)
+	if gceCS.nodeBackoff.IsInBackOffSinceUpdate(req.NodeId, gceCS.nodeBackoff.Clock.Now()) {
+		return nil, status.Errorf(codes.Unavailable, "ControllerUnpublish not permitted on node %q due to backoff condition", req.NodeId)
 	}
 
-	// Node is marked so queue up the request
-	gceCS.queue.AddRateLimited(&workItem{
-		ctx:          context.Background(),
-		unpublishReq: req,
-	})
-
-	return nil, status.Error(codes.Unavailable, "Request queued due to error condition on node")
+	resp, err := gceCS.executeControllerUnpublishVolume(ctx, req)
+	if err != nil {
+		klog.Infof("For node %s adding backoff due to error for volume %s", req.NodeId, req.VolumeId)
+		gceCS.nodeBackoff.Next(req.NodeId, gceCS.nodeBackoff.Clock.Now())
+	} else {
+		klog.Infof("For node %s clear backoff due to successful unpublish of volume %v", req.NodeId, req.VolumeId)
+		gceCS.nodeBackoff.Reset(req.NodeId)
+	}
+	return resp, err
 }
 
 func (gceCS *GCEControllerServer) validateControllerUnpublishVolumeRequest(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (string, *meta.Key, error) {
@@ -650,15 +594,9 @@ func (gceCS *GCEControllerServer) executeControllerUnpublishVolume(ctx context.C
 
 	err = gceCS.CloudProvider.DetachDisk(ctx, project, deviceName, instanceZone, instanceName)
 	if err != nil {
-		// Mark the node and rate limit all the following attach/detach
-		// operations for this node
-		gceCS.publishErrorsSeenOnNode[nodeID] = true
 		return nil, status.Error(codes.Internal, fmt.Sprintf("unknown detach error: %v", err))
 	}
 
-	// Detach succeeds so unmark the node
-	delete(gceCS.publishErrorsSeenOnNode, nodeID)
-
 	klog.V(4).Infof("ControllerUnpublishVolume succeeded for disk %v from node %v", volKey, nodeID)
 	return &csi.ControllerUnpublishVolumeResponse{}, nil
 }
diff --git a/pkg/gce-pd-csi-driver/gce-pd-driver.go b/pkg/gce-pd-csi-driver/gce-pd-driver.go
index 17d644ef..692c86ac 100644
--- a/pkg/gce-pd-csi-driver/gce-pd-driver.go
+++ b/pkg/gce-pd-csi-driver/gce-pd-driver.go
@@ -20,7 +20,7 @@ import (
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
-	"k8s.io/client-go/util/workqueue"
+	"k8s.io/client-go/util/flowcontrol"
 	"k8s.io/klog"
 	"k8s.io/mount-utils"
 	common "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
@@ -151,22 +151,17 @@ func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, devi
 
 func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute) *GCEControllerServer {
 	return &GCEControllerServer{
-		Driver:                  gceDriver,
-		CloudProvider:           cloudProvider,
-		seen:                    map[string]int{},
-		volumeLocks:             common.NewVolumeLocks(),
-		queue:                   workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "controllerserver"),
-		publishErrorsSeenOnNode: map[string]bool{},
+		Driver:        gceDriver,
+		CloudProvider: cloudProvider,
+		seen:          map[string]int{},
+		volumeLocks:   common.NewVolumeLocks(),
+		// flowcontrol uses an exponential backoff policy with a factor of 2
+		nodeBackoff: flowcontrol.NewBackOff(nodeBackoffInitialDuration, nodeBackoffMaxDuration),
 	}
 }
 
 func (gceDriver *GCEDriver) Run(endpoint string) {
 	klog.V(4).Infof("Driver: %v", gceDriver.name)
-
-	if gceDriver.cs != nil {
-		gceDriver.cs.Run()
-	}
-
 	//Start the nonblocking GRPC
 	s := NewNonBlockingGRPCServer()
 	// TODO(#34): Only start specific servers based on a flag.


Test patch:
diff --git a/pkg/gce-pd-csi-driver/controller_test.go b/pkg/gce-pd-csi-driver/controller_test.go
index 3574f611..649e3270 100644
--- a/pkg/gce-pd-csi-driver/controller_test.go
+++ b/pkg/gce-pd-csi-driver/controller_test.go
@@ -30,12 +30,14 @@ import (
 	compute "google.golang.org/api/compute/v1"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"k8s.io/apimachinery/pkg/util/clock"
 	"k8s.io/apimachinery/pkg/util/sets"
-	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/util/flowcontrol"
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
 	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
+	gcecloudprovider "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
 )
 
 const (
@@ -1895,7 +1897,7 @@ func TestPickRandAndConsecutive(t *testing.T) {
 }
 
 func TestVolumeOperationConcurrency(t *testing.T) {
-	readyToExecute := make(chan chan struct{}, 1)
+	readyToExecute := make(chan chan gce.Signal, 1)
 	gceDriver := initBlockingGCEDriver(t, []*gce.CloudDisk{
 		createZonalCloudDisk(name + "1"),
 		createZonalCloudDisk(name + "2"),
@@ -1953,13 +1955,13 @@ func TestVolumeOperationConcurrency(t *testing.T) {
 	// Start vol2CreateSnapshot and allow it to execute to completion. Then check for success.
 	vol2CreateSnapshotResp := runRequest(vol2CreateSnapshotReq)
 	execVol2CreateSnapshot := <-readyToExecute
-	execVol2CreateSnapshot <- struct{}{}
+	execVol2CreateSnapshot <- gce.Signal{}
 	if err := <-vol2CreateSnapshotResp; err != nil {
 		t.Errorf("Unexpected error: %v", err)
 	}
 
 	// To clean up, allow the vol1CreateSnapshotA to complete
-	execVol1CreateSnapshotA <- struct{}{}
+	execVol1CreateSnapshotA <- gce.Signal{}
 	if err := <-vol1CreateSnapshotAResp; err != nil {
 		t.Errorf("Unexpected error: %v", err)
 	}
@@ -2073,184 +2075,268 @@ func TestCreateVolumeDiskReady(t *testing.T) {
 	}
 }
 
-func TestControllerPublishUnpublishVolume(t *testing.T) {
-	testCases := []struct {
-		name              string
-		seedDisks         []*gce.CloudDisk
-		pubReq            *csi.ControllerPublishVolumeRequest
-		unpubReq          *csi.ControllerUnpublishVolumeRequest
-		errorSeenOnNode   bool
-		fakeCloudProvider bool
-	}{
-		{
-			name: "queue up publish requests if node has publish error",
-			seedDisks: []*gce.CloudDisk{
-				createZonalCloudDisk(name),
-			},
-			pubReq: &csi.ControllerPublishVolumeRequest{
-				VolumeId: testVolumeID,
-				NodeId:   testNodeID,
-				VolumeCapability: &csi.VolumeCapability{
-					AccessType: &csi.VolumeCapability_Mount{
-						Mount: &csi.VolumeCapability_MountVolume{},
-					},
-					AccessMode: &csi.VolumeCapability_AccessMode{
-						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
-					},
-				},
-			},
-			errorSeenOnNode:   true,
-			fakeCloudProvider: false,
-		},
-		{
-			name: "queue up and process publish requests if node has publish error",
-			seedDisks: []*gce.CloudDisk{
-				createZonalCloudDisk(name),
-			},
-			pubReq: &csi.ControllerPublishVolumeRequest{
-				VolumeId: testVolumeID,
-				NodeId:   testNodeID,
-				VolumeCapability: &csi.VolumeCapability{
-					AccessType: &csi.VolumeCapability_Mount{
-						Mount: &csi.VolumeCapability_MountVolume{},
-					},
-					AccessMode: &csi.VolumeCapability_AccessMode{
-						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
-					},
-				},
-			},
-			errorSeenOnNode:   true,
-			fakeCloudProvider: true,
-		},
-		{
-			name: "do not queue up publish requests if node doesn't have publish error",
-			seedDisks: []*gce.CloudDisk{
-				createZonalCloudDisk(name),
-			},
-			pubReq: &csi.ControllerPublishVolumeRequest{
-				VolumeId: testVolumeID,
-				NodeId:   testNodeID,
-				VolumeCapability: &csi.VolumeCapability{
-					AccessType: &csi.VolumeCapability_Mount{
-						Mount: &csi.VolumeCapability_MountVolume{},
-					},
-					AccessMode: &csi.VolumeCapability_AccessMode{
-						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
-					},
-				},
-			},
-			errorSeenOnNode:   false,
-			fakeCloudProvider: false,
-		},
-		{
-			name: "queue up unpublish requests if node has publish error",
-			seedDisks: []*gce.CloudDisk{
-				createZonalCloudDisk(name),
-			},
-			unpubReq: &csi.ControllerUnpublishVolumeRequest{
-				VolumeId: testVolumeID,
-				NodeId:   testNodeID,
-			},
-			errorSeenOnNode:   true,
-			fakeCloudProvider: false,
-		},
-		{
-			name: "queue up and process unpublish requests if node has publish error",
-			seedDisks: []*gce.CloudDisk{
-				createZonalCloudDisk(name),
-			},
-			unpubReq: &csi.ControllerUnpublishVolumeRequest{
-				VolumeId: testVolumeID,
-				NodeId:   testNodeID,
-			},
-			errorSeenOnNode:   true,
-			fakeCloudProvider: true,
-		},
-		{
-			name: "do not queue up unpublish requests if node doesn't have publish error",
-			seedDisks: []*gce.CloudDisk{
-				createZonalCloudDisk(name),
-			},
-			unpubReq: &csi.ControllerUnpublishVolumeRequest{
-				VolumeId: testVolumeID,
-				NodeId:   testNodeID,
-			},
-			errorSeenOnNode:   false,
-			fakeCloudProvider: false,
+type backoffTesterConfig struct {
+	mockMissingInstance bool // used by the backoff tester to mock a missing instance scenario
+}
+
+func TestControllerUnpublishBackoff(t *testing.T) {
+	backoffTesterForUnpublish(t, &backoffTesterConfig{
+		mockMissingInstance: true,
+	})
+	backoffTesterForUnpublish(t, &backoffTesterConfig{})
+}
+
+func backoffTesterForUnpublish(t *testing.T, config *backoffTesterConfig) {
+	readyToExecute := make(chan chan gce.Signal, 1)
+	disk1 := name + "1"
+	cloudDisks := []*gce.CloudDisk{
+		createZonalCloudDisk(disk1),
+	}
+	fcp, err := gce.CreateFakeCloudProvider(project, zone, cloudDisks)
+	if err != nil {
+		t.Fatalf("Failed to create fake cloud provider: %v", err)
+	}
+	fcpBlocking := &gce.FakeBlockingCloudProvider{
+		FakeCloudProvider: fcp,
+		ReadyToExecute:    readyToExecute,
+	}
+	instance := &compute.Instance{
+		Name: node,
+		Disks: []*compute.AttachedDisk{
+			{DeviceName: disk1}, // mock attached disks
 		},
 	}
-	for _, tc := range testCases {
-		t.Logf("test case: %s", tc.name)
+	if !config.mockMissingInstance {
+		fcp.InsertInstance(instance, zone, node)
+	}
 
-		var gceDriver *GCEDriver
+	driver := GetGCEDriver()
+	tc := clock.NewFakeClock(time.Now())
+	driver.cs = &GCEControllerServer{
+		Driver:        driver,
+		CloudProvider: fcpBlocking,
+		seen:          map[string]int{},
+		volumeLocks:   common.NewVolumeLocks(),
+		nodeBackoff:   flowcontrol.NewFakeBackOff(nodeBackoffInitialDuration, nodeBackoffMaxDuration, tc),
+	}
 
-		if tc.fakeCloudProvider {
-			fcp, err := gce.CreateFakeCloudProvider(project, zone, tc.seedDisks)
-			if err != nil {
-				t.Fatalf("Failed to create fake cloud provider: %v", err)
-			}
+	key := testNodeID
+	step := 1 * time.Millisecond
+	// Mock an active backoff condition on the node. This will setup a backoff duration of the 'nodeBackoffInitialDuration'.
+	driver.cs.nodeBackoff.Next(key, tc.Now())
+	unpubreq := &csi.ControllerUnpublishVolumeRequest{
+		VolumeId: testVolumeID + "1",
+		NodeId:   testNodeID,
+	}
+	// For the first 199 ms, the backoff condition is true. All controller publish request will be denied with 'Unavailable' error code.
+	for i := 0; i < 199; i++ {
+		tc.Step(step)
+		var err error
+		_, err = driver.cs.ControllerUnpublishVolume(context.Background(), unpubreq)
+		if !isUnavailableError(err) {
+			t.Errorf("unexpected error %v", err)
+		}
+	}
 
-			instance := &compute.Instance{
-				Name:  node,
-				Disks: []*compute.AttachedDisk{},
-			}
-			fcp.InsertInstance(instance, zone, node)
+	// Mock clock tick for the 200th millisecond. So backoff condition is no longer true.
+	tc.Step(step)
+	runUnpublishRequest := func(req *csi.ControllerUnpublishVolumeRequest) <-chan error {
+		response := make(chan error)
+		go func() {
+			_, err := driver.cs.ControllerUnpublishVolume(context.Background(), req)
+			response <- err
+		}()
+		return response
+	}
 
-			// Setup new driver each time so no interference
-			gceDriver = initGCEDriverWithCloudProvider(t, fcp)
-		} else {
-			gceDriver = initGCEDriver(t, tc.seedDisks)
+	// For a missing instance the driver should return a success code, and the node backoff condition should be cleared.
+	if config.mockMissingInstance {
+		_, err = driver.cs.ControllerUnpublishVolume(context.Background(), unpubreq)
+		// Driver is expected to remove the node key from the backoff map.
+		t1 := driver.cs.nodeBackoff.Get(key)
+		if t1 != 0 {
+			t.Error("unexpected delay")
 		}
+		return
+	}
+
+	// mock an error
+	var respUnpublish <-chan error
+	respUnpublish = runUnpublishRequest(unpubreq)
+	execute := <-readyToExecute
+	s1 := gcecloudprovider.Signal{ReportError: true}
+	execute <- s1
+	if err := <-respUnpublish; err == nil {
+		t.Errorf("expected error")
+	}
 
-		// mark the node in the map
-		if tc.errorSeenOnNode {
-			gceDriver.cs.publishErrorsSeenOnNode[testNodeID] = true
+	// The above failure should cause driver to call Backoff.Next() again and a backoff duration of 400 ms duration is set starting at the 200th millisecond.
+	// For the 200-599 ms, the backoff condition is true, and new controller publish requests will be deined.
+	for i := 0; i < 399; i++ {
+		tc.Step(step)
+		var err error
+		_, err = driver.cs.ControllerUnpublishVolume(context.Background(), unpubreq)
+		if !isUnavailableError(err) {
+			t.Errorf("unexpected error %v", err)
 		}
+	}
 
-		requestCount := 50
-		for i := 0; i < requestCount; i++ {
-			if tc.pubReq != nil {
-				gceDriver.cs.ControllerPublishVolume(context.Background(), tc.pubReq)
-			}
+	// Mock clock tick for the 600th millisecond. So backoff condition is no longer true.
+	tc.Step(step)
+	// Now mock a successful ControllerUnpublish request, where DetachDisk call succeeds.
+	respUnpublish = runUnpublishRequest(unpubreq)
+	execute = <-readyToExecute
+	s1 = gcecloudprovider.Signal{}
+	execute <- s1
+	if err := <-respUnpublish; err != nil {
+		t.Errorf("unexpected error")
+	}
 
-			if tc.unpubReq != nil {
-				gceDriver.cs.ControllerUnpublishVolume(context.Background(), tc.unpubReq)
-			}
-		}
+	// Driver is expected to remove the node key from the backoff map.
+	t1 := driver.cs.nodeBackoff.Get(key)
+	if t1 != 0 {
+		t.Error("unexpected delay")
+	}
+}
 
-		queued := false
+func TestControllerPublishBackoff(t *testing.T) {
+	backoffTesterForPublish(t, &backoffTesterConfig{
+		mockMissingInstance: true,
+	})
+	backoffTesterForPublish(t, &backoffTesterConfig{})
+}
 
-		if tc.errorSeenOnNode {
-			if err := wait.Poll(10*time.Nanosecond, 1*time.Second, func() (bool, error) {
-				if gceDriver.cs.queue.Len() > 0 {
-					queued = true
+func backoffTesterForPublish(t *testing.T, config *backoffTesterConfig) {
+	readyToExecute := make(chan chan gce.Signal, 1)
+	disk1 := name + "1"
+	cloudDisks := []*gce.CloudDisk{
+		createZonalCloudDisk(disk1),
+	}
+	fcp, err := gce.CreateFakeCloudProvider(project, zone, cloudDisks)
+	if err != nil {
+		t.Fatalf("Failed to create fake cloud provider: %v", err)
+	}
+	fcpBlocking := &gce.FakeBlockingCloudProvider{
+		FakeCloudProvider: fcp,
+		ReadyToExecute:    readyToExecute,
+	}
+	instance := &compute.Instance{
+		Name:  node,
+		Disks: []*compute.AttachedDisk{},
+	}
+	if !config.mockMissingInstance {
+		fcp.InsertInstance(instance, zone, node)
+	}
 
-					if tc.fakeCloudProvider {
-						gceDriver.cs.Run()
-					}
-				}
+	driver := GetGCEDriver()
+	tc := clock.NewFakeClock(time.Now())
+	driver.cs = &GCEControllerServer{
+		Driver:        driver,
+		CloudProvider: fcpBlocking,
+		seen:          map[string]int{},
+		volumeLocks:   common.NewVolumeLocks(),
+		nodeBackoff:   flowcontrol.NewFakeBackOff(nodeBackoffInitialDuration, nodeBackoffMaxDuration, tc),
+	}
 
-				// Items are queued up and eventually all processed
-				if tc.fakeCloudProvider {
-					return queued && gceDriver.cs.queue.Len() == 0, nil
-				}
+	key := testNodeID
+	step := 1 * time.Millisecond
+	// Mock an active backoff condition on the node. This will setup a backoff duration of the 'nodeBackoffInitialDuration'.
+	driver.cs.nodeBackoff.Next(key, tc.Now())
+	pubreq := &csi.ControllerPublishVolumeRequest{
+		VolumeId: testVolumeID + "1",
+		NodeId:   testNodeID,
+		VolumeCapability: &csi.VolumeCapability{
+			AccessType: &csi.VolumeCapability_Mount{
+				Mount: &csi.VolumeCapability_MountVolume{},
+			},
+			AccessMode: &csi.VolumeCapability_AccessMode{
+				Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+			},
+		},
+	}
+	// For the first 199 ms, the backoff condition is true. All controller publish request will be denied with 'Unavailable' error code.
+	for i := 0; i < 199; i++ {
+		tc.Step(step)
+		var err error
+		_, err = driver.cs.ControllerPublishVolume(context.Background(), pubreq)
+		if !isUnavailableError(err) {
+			t.Errorf("unexpected error %v", err)
+		}
+	}
 
-				return gceDriver.cs.queue.Len() == requestCount, nil
-			}); err != nil {
-				if tc.fakeCloudProvider {
-					t.Fatalf("%v requests not processed for node has seen error", gceDriver.cs.queue.Len())
-				} else {
-					t.Fatalf("Only %v requests queued up for node has seen error", gceDriver.cs.queue.Len())
-				}
-			}
+	// Mock clock tick for the 200th millisecond. So backoff condition is no longer true.
+	tc.Step(step)
+	runPublishRequest := func(req *csi.ControllerPublishVolumeRequest) <-chan error {
+		response := make(chan error)
+		go func() {
+			_, err := driver.cs.ControllerPublishVolume(context.Background(), req)
+			response <- err
+		}()
+		return response
+	}
+
+	// For a missing instance the driver should return error code, and the node backoff condition should be set.
+	if config.mockMissingInstance {
+		_, err = driver.cs.ControllerPublishVolume(context.Background(), pubreq)
+		if err == nil {
+			t.Errorf("unexpected error %v", err)
 		}
 
-		if !tc.errorSeenOnNode {
-			if err := wait.Poll(10*time.Nanosecond, 10*time.Millisecond, func() (bool, error) {
-				return gceDriver.cs.queue.Len() != 0, nil
-			}); err == nil {
-				t.Fatalf("%v requests queued up for node hasn't seen error", gceDriver.cs.queue.Len())
-			}
+		t1 := driver.cs.nodeBackoff.Get(key)
+		if t1 == 0 {
+			t.Error("expected delay, got none")
+		}
+		return
+	}
+
+	// mock an error
+	var respPublish <-chan error
+	respPublish = runPublishRequest(pubreq)
+	execute := <-readyToExecute
+	s1 := gcecloudprovider.Signal{ReportError: true}
+	execute <- s1
+	if err := <-respPublish; err == nil {
+		t.Errorf("expected error")
+	}
+
+	// The above failure should cause driver to call Backoff.Next() again and a backoff duration of 400 ms duration is set starting at the 200th millisecond.
+	// For the 200-599 ms, the backoff condition is true, and new controller publish requests will be deined.
+	for i := 0; i < 399; i++ {
+		tc.Step(step)
+		var err error
+		_, err = driver.cs.ControllerPublishVolume(context.Background(), pubreq)
+		if !isUnavailableError(err) {
+			t.Errorf("unexpected error %v", err)
 		}
 	}
+
+	// Mock clock tick for the 600th millisecond. So backoff condition is no longer true.
+	tc.Step(step)
+	// Now mock a successful ControllerUnpublish request, where DetachDisk call succeeds.
+	respPublish = runPublishRequest(pubreq)
+	execute = <-readyToExecute
+	s1 = gcecloudprovider.Signal{}
+	execute <- s1
+	if err := <-respPublish; err != nil {
+		t.Errorf("unexpected error")
+	}
+
+	// Driver is expected to remove the node key from the backoff map.
+	t1 := driver.cs.nodeBackoff.Get(key)
+	if t1 != 0 {
+		t.Error("unexpected delay")
+	}
+}
+
+func isUnavailableError(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	st, ok := status.FromError(err)
+	if !ok {
+		return false
+	}
+
+	return st.Code().String() == "Unavailable"
 }
diff --git a/pkg/gce-pd-csi-driver/gce-pd-driver_test.go b/pkg/gce-pd-csi-driver/gce-pd-driver_test.go
index d645907e..7aa2fd86 100644
--- a/pkg/gce-pd-csi-driver/gce-pd-driver_test.go
+++ b/pkg/gce-pd-csi-driver/gce-pd-driver_test.go
@@ -28,7 +28,7 @@ func initGCEDriver(t *testing.T, cloudDisks []*gce.CloudDisk) *GCEDriver {
 	return initGCEDriverWithCloudProvider(t, fakeCloudProvider)
 }
 
-func initBlockingGCEDriver(t *testing.T, cloudDisks []*gce.CloudDisk, readyToExecute chan chan struct{}) *GCEDriver {
+func initBlockingGCEDriver(t *testing.T, cloudDisks []*gce.CloudDisk, readyToExecute chan chan gce.Signal) *GCEDriver {
 	fakeCloudProvider, err := gce.CreateFakeCloudProvider(project, zone, cloudDisks)
 	if err != nil {
 		t.Fatalf("Failed to create fake cloud provider: %v", err)
diff --git a/test/e2e/tests/single_zone_e2e_test.go b/test/e2e/tests/single_zone_e2e_test.go
index 02317041..abacc564 100644
--- a/test/e2e/tests/single_zone_e2e_test.go
+++ b/test/e2e/tests/single_zone_e2e_test.go
@@ -603,6 +603,8 @@ var _ = Describe("GCE PD CSI Driver", func() {
 			Expect(err).To(BeNil(), "Failed to enable crypto key")
 		}
 
+		// The controller publish failure in above step would set a backoff condition on the node. Wait suffcient amount of time for the driver to accept new controller publish requests.
+		time.Sleep(time.Second)
 		// Make sure attach of PD succeeds
 		err = testAttachWriteReadDetach(volID, volName, controllerInstance, controllerClient, false /* readOnly */)
 		Expect(err).To(BeNil(), "Failed to go through volume lifecycle after restoring CMEK key")