Repository: kubernetes-sigs/gcp-compute-persistent-disk-csi-driver
Pull request #415 (2019-10-23 23:08:37)

Add support for multi-writer PD
/kind feature

/cc @msau42

**What this PR does / why we need it**:
Adds multi writer support (currently alpha in GCP).

**Special notes for your reviewer**:
There's a bit of code duplication due to alpha copies of certain methods. I found it challenging to reduce the duplication but I'd be open to suggestions on how to refactor. On the other hand it might just be a temporary situation until the API's move to beta / GA.

**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
Add support for multi-writer raw block devices.
```


k8s-ci-robot: Welcome @sschmitt! <br><br>It looks like this is your first PR to <a href='https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver'>kubernetes-sigs/gcp-compute-persistent-disk-csi-driver</a> 🎉. Please refer to our [pull request process documentation](https://git.k8s.io/community/contributors/guide/pull-requests.md) to help your PR have a smooth ride to approval. <br><br>You will be prompted by a bot to use commands during the review process. Do not be afraid to follow the prompts! It is okay to experiment. [Here is the bot commands documentation](https://go.k8s.io/bot-commands). <br><br>You can also check if kubernetes-sigs/gcp-compute-persistent-disk-csi-driver has [its own contribution guidelines](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/tree/master/CONTRIBUTING.md). <br><br>You may want to refer to our [testing guide](https://git.k8s.io/community/contributors/devel/sig-testing/testing.md) if you run into trouble with your tests not passing. <br><br>If you are having difficulty getting your pull request seen, please follow the [recommended escalation practices](https://github.com/kubernetes/community/blob/master/contributors/guide/pull-requests.md#why-is-my-pull-request-not-getting-reviewed). Also, for tips and tricks in the contribution process you may want to read the [Kubernetes contributor cheat sheet](https://git.k8s.io/community/contributors/guide/contributor-cheatsheet/README.md). We want to make sure your contribution gets all the attention it needs! <br><br>Thank you, and welcome to Kubernetes. :smiley:
k8s-ci-robot: @sschmitt: PR needs rebase.

<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.
</details>
k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/pull/415#" title="Author self-approved">sschmitt</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **msau42**
You can assign the PR to them by writing `/assign @msau42` in a comment when ready.

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).

<details open>
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)**

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":["msau42"]} -->
k8s-ci-robot: Hi @sschmitt. Thanks for your PR.

I'm waiting for a [kubernetes-sigs](https://github.com/orgs/kubernetes-sigs/people) or [kubernetes](https://github.com/orgs/kubernetes/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://git.k8s.io/community/community-membership.md#member) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fgcp-compute-persistent-disk-csi-driver).

<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.
</details>

msau42: /ok-to-test
/assign @davidz627 
k8s-ci-robot: @sschmitt: The following test **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-e2e | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-e2e/1187146115821080578/) | `/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=415). [Your PR dashboard](https://gubernator.k8s.io/pr/sschmitt). 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: @sschmitt: The following tests **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-e2e | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-e2e/1187146115821080578/) | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`
pull-gcp-compute-persistent-disk-csi-driver-verify | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-verify/1187146115816886279/) | `/test pull-gcp-compute-persistent-disk-csi-driver-verify`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=415). [Your PR dashboard](https://gubernator.k8s.io/pr/sschmitt). 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: @sschmitt: The following tests **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-e2e | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-e2e/1187146115821080578/) | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`
pull-gcp-compute-persistent-disk-csi-driver-verify | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-verify/1187146115816886279/) | `/test pull-gcp-compute-persistent-disk-csi-driver-verify`
pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration/1187146115816886278/) | `/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=415). [Your PR dashboard](https://gubernator.k8s.io/pr/sschmitt). 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: @sschmitt: The following tests **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-e2e | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-e2e/1187146115821080578/) | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`
pull-gcp-compute-persistent-disk-csi-driver-verify | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-verify/1187146115816886279/) | `/test pull-gcp-compute-persistent-disk-csi-driver-verify`
pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration/1187146115816886278/) | `/test pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration`
pull-gcp-compute-persistent-disk-csi-driver-unit | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-unit/1187146115821080576/) | `/test pull-gcp-compute-persistent-disk-csi-driver-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=415). [Your PR dashboard](https://gubernator.k8s.io/pr/sschmitt). 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: @sschmitt: The following tests **failed**, say `/retest` to rerun them all:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-e2e | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-e2e/1187146115821080578/) | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`
pull-gcp-compute-persistent-disk-csi-driver-verify | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-verify/1187146115816886279/) | `/test pull-gcp-compute-persistent-disk-csi-driver-verify`
pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration/1187146115816886278/) | `/test pull-gcp-compute-persistent-disk-csi-driver-kubernetes-integration`
pull-gcp-compute-persistent-disk-csi-driver-unit | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-unit/1187146115821080576/) | `/test pull-gcp-compute-persistent-disk-csi-driver-unit`
pull-gcp-compute-persistent-disk-csi-driver-sanity | 2351b67a0031b492dc568a016ea78f9b8fcd0f8a | [link](https://prow.k8s.io/view/gcs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-compute-persistent-disk-csi-driver/415/pull-gcp-compute-persistent-disk-csi-driver-sanity/1187146115821080577/) | `/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=415). [Your PR dashboard](https://gubernator.k8s.io/pr/sschmitt). 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 -->

Patch:
diff --git a/examples/kubernetes/demo-pod-shared-pd.yaml b/examples/kubernetes/demo-pod-shared-pd.yaml
new file mode 100644
index 00000000..5972feb5
--- /dev/null
+++ b/examples/kubernetes/demo-pod-shared-pd.yaml
@@ -0,0 +1,50 @@
+kind: PersistentVolumeClaim
+apiVersion: v1
+metadata:
+  name: podpvc
+spec:
+  accessModes:
+    - ReadWriteMany
+  volumeMode: Block
+  storageClassName: csi-gce-pd
+  resources:
+    requests:
+      storage: 200Gi
+
+---
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: web-server
+spec:
+  selector:
+    matchLabels:
+      app: web
+  replicas: 2
+  template:
+    metadata:
+      labels:
+        app: web
+    spec:
+      affinity:
+        podAntiAffinity:
+          requiredDuringSchedulingIgnoredDuringExecution:
+          - labelSelector:
+              matchExpressions:
+              - key: app
+                operator: In
+                values:
+                - web
+            topologyKey: "kubernetes.io/hostname"
+      containers:
+        - name: web-server
+          image: nginx
+          volumeDevices:
+            - name: mypvc
+              devicePath: /dev/mypvc
+      volumes:
+        - name: mypvc
+          persistentVolumeClaim:
+            claimName: podpvc
+            readOnly: false
diff --git a/pkg/gce-cloud-provider/compute/cloud-disk.go b/pkg/gce-cloud-provider/compute/cloud-disk.go
index 0fb2a8d0..98e598ec 100644
--- a/pkg/gce-cloud-provider/compute/cloud-disk.go
+++ b/pkg/gce-cloud-provider/compute/cloud-disk.go
@@ -17,12 +17,15 @@ package gcecloudprovider
 import (
 	"strings"
 
+	computealpha "google.golang.org/api/compute/v0.alpha"
 	computev1 "google.golang.org/api/compute/v1"
 )
 
 type CloudDisk struct {
-	ZonalDisk    *computev1.Disk
-	RegionalDisk *computev1.Disk
+	ZonalDisk         *computev1.Disk
+	RegionalDisk      *computev1.Disk
+	ZonalAlphaDisk    *computealpha.Disk
+	RegionalAlphaDisk *computealpha.Disk
 }
 
 type CloudDiskType string
@@ -32,6 +35,10 @@ const (
 	Zonal = "zonal"
 	// Regional key type.
 	Regional = "regional"
+	// ZonalAlpha key type.
+	ZonalAlpha = "zonalAlpha"
+	// RegionalAlpha key type.
+	RegionalAlpha = "regionalAlpha"
 	// Global key type.
 	Global = "global"
 )
@@ -48,12 +55,28 @@ func RegionalCloudDisk(disk *computev1.Disk) *CloudDisk {
 	}
 }
 
+func ZonalAlphaCloudDisk(disk *computealpha.Disk) *CloudDisk {
+	return &CloudDisk{
+		ZonalAlphaDisk: disk,
+	}
+}
+
+func RegionalAlphaCloudDisk(disk *computealpha.Disk) *CloudDisk {
+	return &CloudDisk{
+		RegionalAlphaDisk: disk,
+	}
+}
+
 func (d *CloudDisk) Type() CloudDiskType {
 	switch {
 	case d.ZonalDisk != nil:
 		return Zonal
 	case d.RegionalDisk != nil:
 		return Regional
+	case d.ZonalAlphaDisk != nil:
+		return ZonalAlpha
+	case d.RegionalAlphaDisk != nil:
+		return RegionalAlpha
 	default:
 		return Global
 	}
@@ -65,6 +88,10 @@ func (d *CloudDisk) GetUsers() []string {
 		return d.ZonalDisk.Users
 	case Regional:
 		return d.RegionalDisk.Users
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.Users
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.Users
 	default:
 		return nil
 	}
@@ -76,6 +103,10 @@ func (d *CloudDisk) GetName() string {
 		return d.ZonalDisk.Name
 	case Regional:
 		return d.RegionalDisk.Name
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.Name
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.Name
 	default:
 		return ""
 	}
@@ -87,6 +118,10 @@ func (d *CloudDisk) GetKind() string {
 		return d.ZonalDisk.Kind
 	case Regional:
 		return d.RegionalDisk.Kind
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.Kind
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.Kind
 	default:
 		return ""
 	}
@@ -98,6 +133,10 @@ func (d *CloudDisk) GetStatus() string {
 		return d.ZonalDisk.Status
 	case Regional:
 		return d.RegionalDisk.Status
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.Status
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.Status
 	default:
 		return "Unknown"
 	}
@@ -113,6 +152,10 @@ func (d *CloudDisk) GetPDType() string {
 		pdType = d.ZonalDisk.Type
 	case Regional:
 		pdType = d.RegionalDisk.Type
+	case ZonalAlpha:
+		pdType = d.ZonalAlphaDisk.Type
+	case RegionalAlpha:
+		pdType = d.RegionalAlphaDisk.Type
 	default:
 		return ""
 	}
@@ -126,6 +169,10 @@ func (d *CloudDisk) GetSelfLink() string {
 		return d.ZonalDisk.SelfLink
 	case Regional:
 		return d.RegionalDisk.SelfLink
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.SelfLink
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.SelfLink
 	default:
 		return ""
 	}
@@ -137,6 +184,10 @@ func (d *CloudDisk) GetSizeGb() int64 {
 		return d.ZonalDisk.SizeGb
 	case Regional:
 		return d.RegionalDisk.SizeGb
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.SizeGb
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.SizeGb
 	default:
 		return -1
 	}
@@ -150,6 +201,10 @@ func (d *CloudDisk) setSizeGb(size int64) {
 		d.ZonalDisk.SizeGb = size
 	case Regional:
 		d.RegionalDisk.SizeGb = size
+	case ZonalAlpha:
+		d.ZonalAlphaDisk.SizeGb = size
+	case RegionalAlpha:
+		d.RegionalAlphaDisk.SizeGb = size
 	}
 }
 
@@ -159,6 +214,10 @@ func (d *CloudDisk) GetZone() string {
 		return d.ZonalDisk.Zone
 	case Regional:
 		return d.RegionalDisk.Zone
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.Zone
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.Zone
 	default:
 		return ""
 	}
@@ -170,6 +229,10 @@ func (d *CloudDisk) GetSnapshotId() string {
 		return d.ZonalDisk.SourceSnapshotId
 	case Regional:
 		return d.RegionalDisk.SourceSnapshotId
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.SourceSnapshotId
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.SourceSnapshotId
 	default:
 		return ""
 	}
@@ -177,16 +240,39 @@ func (d *CloudDisk) GetSnapshotId() string {
 
 func (d *CloudDisk) GetKMSKeyName() string {
 	var dek *computev1.CustomerEncryptionKey
+	var dekAlpha *computealpha.CustomerEncryptionKey
 	switch d.Type() {
 	case Zonal:
 		dek = d.ZonalDisk.DiskEncryptionKey
 	case Regional:
 		dek = d.RegionalDisk.DiskEncryptionKey
+	case ZonalAlpha:
+		dekAlpha = d.ZonalAlphaDisk.DiskEncryptionKey
+	case RegionalAlpha:
+		dekAlpha = d.RegionalAlphaDisk.DiskEncryptionKey
 	default:
 		return ""
 	}
-	if dek == nil {
-		return ""
+
+	if dek != nil {
+		return dek.KmsKeyName
+	} else if dekAlpha != nil {
+		return dekAlpha.KmsKeyName
+	}
+	return ""
+}
+
+func (d *CloudDisk) GetMultiWriter() bool {
+	switch d.Type() {
+	case Zonal:
+		return false
+	case Regional:
+		return false
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.MultiWriter
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.MultiWriter
+	default:
+		return false
 	}
-	return dek.KmsKeyName
 }
diff --git a/pkg/gce-cloud-provider/compute/fake-gce.go b/pkg/gce-cloud-provider/compute/fake-gce.go
index bfbab953..e61e0686 100644
--- a/pkg/gce-cloud-provider/compute/fake-gce.go
+++ b/pkg/gce-cloud-provider/compute/fake-gce.go
@@ -212,7 +212,7 @@ func (cloud *FakeCloudProvider) ListSnapshots(ctx context.Context, filter string
 }
 
 // Disk Methods
-func (cloud *FakeCloudProvider) GetDisk(ctx context.Context, volKey *meta.Key) (*CloudDisk, error) {
+func (cloud *FakeCloudProvider) GetDisk(ctx context.Context, volKey *meta.Key, api GCEAPIVersion) (*CloudDisk, error) {
 	disk, ok := cloud.disks[volKey.Name]
 	if !ok {
 		return nil, notFoundError()
@@ -220,7 +220,7 @@ func (cloud *FakeCloudProvider) GetDisk(ctx context.Context, volKey *meta.Key) (
 	return disk, nil
 }
 
-func (cloud *FakeCloudProvider) ValidateExistingDisk(ctx context.Context, resp *CloudDisk, params common.DiskParameters, reqBytes, limBytes int64) error {
+func (cloud *FakeCloudProvider) ValidateExistingDisk(ctx context.Context, resp *CloudDisk, params common.DiskParameters, reqBytes, limBytes int64, multiWriter bool) error {
 	if resp == nil {
 		return fmt.Errorf("disk does not exist")
 	}
@@ -232,14 +232,29 @@ func (cloud *FakeCloudProvider) ValidateExistingDisk(ctx context.Context, resp *
 			reqBytes, common.GbToBytes(resp.GetSizeGb()), limBytes)
 	}
 
+	respType := strings.Split(resp.GetPDType(), "/")
+	typeMatch := strings.TrimSpace(respType[len(respType)-1]) == strings.TrimSpace(params.DiskType)
+	typeDefault := params.DiskType == "" && strings.TrimSpace(respType[len(respType)-1]) == "pd-standard"
+	if !typeMatch && !typeDefault {
+		return fmt.Errorf("disk already exists with incompatible type. Need %v. Got %v",
+			params.DiskType, respType[len(respType)-1])
+	}
+
+	// We are assuming here that a multiWriter disk could be used as non-multiWriter
+	if multiWriter && !resp.GetMultiWriter() {
+		return fmt.Errorf("disk already exists with incompatible capability. Need MultiWriter. Got non-MultiWriter")
+	}
+
+	klog.V(4).Infof("Compatible disk already exists")
 	return ValidateDiskParameters(resp, params)
 }
 
-func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string) error {
+func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, multiWriter bool) error {
 	if disk, ok := cloud.disks[volKey.Name]; ok {
 		err := cloud.ValidateExistingDisk(ctx, disk, params,
 			int64(capacityRange.GetRequiredBytes()),
-			int64(capacityRange.GetLimitBytes()))
+			int64(capacityRange.GetLimitBytes()),
+			multiWriter)
 		if err != nil {
 			return err
 		}
diff --git a/pkg/gce-cloud-provider/compute/gce-compute.go b/pkg/gce-cloud-provider/compute/gce-compute.go
index 37c189c5..37b173b0 100644
--- a/pkg/gce-cloud-provider/compute/gce-compute.go
+++ b/pkg/gce-cloud-provider/compute/gce-compute.go
@@ -24,6 +24,7 @@ import (
 
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
+	computealpha "google.golang.org/api/compute/v0.alpha"
 	computev1 "google.golang.org/api/compute/v1"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -39,15 +40,24 @@ const (
 	cryptoKeyVerDelimiter          = "/cryptoKeyVersions"
 )
 
+type GCEAPIVersion string
+
+const (
+	// V1 key type
+	GCEAPIVersionV1 GCEAPIVersion = "v1"
+	// Alpha key type
+	GCEAPIVersionAlpha GCEAPIVersion = "alpha"
+)
+
 type GCECompute interface {
 	// Metadata information
 	GetDefaultProject() string
 	GetDefaultZone() string
 	// Disk Methods
-	GetDisk(ctx context.Context, volumeKey *meta.Key) (*CloudDisk, error)
+	GetDisk(ctx context.Context, volumeKey *meta.Key, gceAPIVersion GCEAPIVersion) (*CloudDisk, error)
 	RepairUnderspecifiedVolumeKey(ctx context.Context, volumeKey *meta.Key) (*meta.Key, error)
-	ValidateExistingDisk(ctx context.Context, disk *CloudDisk, params common.DiskParameters, reqBytes, limBytes int64) error
-	InsertDisk(ctx context.Context, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string) error
+	ValidateExistingDisk(ctx context.Context, disk *CloudDisk, params common.DiskParameters, reqBytes, limBytes int64, multiWriter bool) error
+	InsertDisk(ctx context.Context, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, multiWriter bool) error
 	DeleteDisk(ctx context.Context, volumeKey *meta.Key) error
 	AttachDisk(ctx context.Context, volKey *meta.Key, readWrite, diskType, instanceZone, instanceName string) error
 	DetachDisk(ctx context.Context, deviceName string, instanceZone, instanceName string) error
@@ -179,15 +189,25 @@ func (cloud *CloudProvider) ListSnapshots(ctx context.Context, filter string, ma
 
 }
 
-func (cloud *CloudProvider) GetDisk(ctx context.Context, key *meta.Key) (*CloudDisk, error) {
+func (cloud *CloudProvider) GetDisk(ctx context.Context, key *meta.Key, gceAPIVersion GCEAPIVersion) (*CloudDisk, error) {
 	klog.V(5).Infof("Getting disk %v", key)
 	switch key.Type() {
 	case meta.Zonal:
-		disk, err := cloud.getZonalDiskOrError(ctx, key.Zone, key.Name)
-		return ZonalCloudDisk(disk), err
+		if gceAPIVersion == GCEAPIVersionAlpha {
+			disk, err := cloud.getZonalAlphaDiskOrError(ctx, key.Zone, key.Name)
+			return ZonalAlphaCloudDisk(disk), err
+		} else {
+			disk, err := cloud.getZonalDiskOrError(ctx, key.Zone, key.Name)
+			return ZonalCloudDisk(disk), err
+		}
 	case meta.Regional:
-		disk, err := cloud.getRegionalDiskOrError(ctx, key.Region, key.Name)
-		return RegionalCloudDisk(disk), err
+		if gceAPIVersion == GCEAPIVersionAlpha {
+			disk, err := cloud.getRegionalAlphaDiskOrError(ctx, key.Region, key.Name)
+			return RegionalAlphaCloudDisk(disk), err
+		} else {
+			disk, err := cloud.getRegionalDiskOrError(ctx, key.Region, key.Name)
+			return RegionalCloudDisk(disk), err
+		}
 	default:
 		return nil, fmt.Errorf("key was neither zonal nor regional, got: %v", key.String())
 	}
@@ -213,6 +233,24 @@ func (cloud *CloudProvider) getRegionalDiskOrError(ctx context.Context, volumeRe
 	return disk, nil
 }
 
+func (cloud *CloudProvider) getZonalAlphaDiskOrError(ctx context.Context, volumeZone, volumeName string) (*computealpha.Disk, error) {
+	project := cloud.project
+	disk, err := cloud.alphaService.Disks.Get(project, volumeZone, volumeName).Context(ctx).Do()
+	if err != nil {
+		return nil, err
+	}
+	return disk, nil
+}
+
+func (cloud *CloudProvider) getRegionalAlphaDiskOrError(ctx context.Context, volumeRegion, volumeName string) (*computealpha.Disk, error) {
+	project := cloud.project
+	disk, err := cloud.alphaService.RegionDisks.Get(project, volumeRegion, volumeName).Context(ctx).Do()
+	if err != nil {
+		return nil, err
+	}
+	return disk, nil
+}
+
 func (cloud *CloudProvider) GetReplicaZoneURI(zone string) string {
 	return cloud.service.BasePath + fmt.Sprintf(
 		replicaZoneURITemplateSingleZone,
@@ -227,7 +265,7 @@ func (cloud *CloudProvider) getRegionURI(region string) string {
 		region)
 }
 
-func (cloud *CloudProvider) ValidateExistingDisk(ctx context.Context, resp *CloudDisk, params common.DiskParameters, reqBytes, limBytes int64) error {
+func (cloud *CloudProvider) ValidateExistingDisk(ctx context.Context, resp *CloudDisk, params common.DiskParameters, reqBytes, limBytes int64, multiWriter bool) error {
 	klog.V(5).Infof("Validating existing disk %v with diskType: %s, reqested bytes: %v, limit bytes: %v", resp, params.DiskType, reqBytes, limBytes)
 	if resp == nil {
 		return fmt.Errorf("disk does not exist")
@@ -240,6 +278,11 @@ func (cloud *CloudProvider) ValidateExistingDisk(ctx context.Context, resp *Clou
 			reqBytes, common.GbToBytes(resp.GetSizeGb()), limBytes)
 	}
 
+	// We are assuming here that a multiWriter disk could be used as non-multiWriter
+	if multiWriter && !resp.GetMultiWriter() {
+		return fmt.Errorf("disk already exists with incompatible capability. Need MultiWriter. Got non-MultiWriter")
+	}
+
 	return ValidateDiskParameters(resp, params)
 }
 
@@ -250,11 +293,11 @@ func ValidateDiskParameters(disk *CloudDisk, params common.DiskParameters) error
 		return fmt.Errorf("actual pd type %s did not match the expected param %s", disk.GetPDType(), params.DiskType)
 	}
 
-	if params.ReplicationType == "none" && disk.Type() != Zonal {
+	if params.ReplicationType == "none" && disk.Type() != Zonal && disk.Type() != ZonalAlpha {
 		return fmt.Errorf("actual disk replication type %v did not match expected param %s", disk.Type(), params.ReplicationType)
 	}
 
-	if params.ReplicationType == "regional-pd" && disk.Type() != Regional {
+	if params.ReplicationType == "regional-pd" && disk.Type() != Regional && disk.Type() != RegionalAlpha {
 		return fmt.Errorf("actual disk replication type %v did not match expected param %s", disk.Type(), "regional-pd")
 	}
 
@@ -267,7 +310,7 @@ func ValidateDiskParameters(disk *CloudDisk, params common.DiskParameters) error
 	return nil
 }
 
-func (cloud *CloudProvider) InsertDisk(ctx context.Context, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string) error {
+func (cloud *CloudProvider) InsertDisk(ctx context.Context, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, multiWriter bool) error {
 	klog.V(5).Infof("Inserting disk %v", volKey)
 
 	description, err := encodeDiskTags(params.Tags)
@@ -280,17 +323,46 @@ func (cloud *CloudProvider) InsertDisk(ctx context.Context, volKey *meta.Key, pa
 		if description == "" {
 			description = "Disk created by GCE-PD CSI Driver"
 		}
-		return cloud.insertZonalDisk(ctx, volKey, params, capBytes, capacityRange, snapshotID, description)
+		return cloud.insertZonalDisk(ctx, volKey, params, capBytes, capacityRange, snapshotID, description, multiWriter)
 	case meta.Regional:
 		if description == "" {
 			description = "Regional disk created by GCE-PD CSI Driver"
 		}
-		return cloud.insertRegionalDisk(ctx, volKey, params, capBytes, capacityRange, replicaZones, snapshotID, description)
+		return cloud.insertRegionalDisk(ctx, volKey, params, capBytes, capacityRange, replicaZones, snapshotID, description, multiWriter)
 	default:
 		return fmt.Errorf("could not insert disk, key was neither zonal nor regional, instead got: %v", volKey.String())
 	}
 }
 
+func convertV1CustomerEncryptionKeyToAlpha(v1Key *computev1.CustomerEncryptionKey) *computealpha.CustomerEncryptionKey {
+	return &computealpha.CustomerEncryptionKey{
+		KmsKeyName:      v1Key.KmsKeyName,
+		RawKey:          v1Key.RawKey,
+		Sha256:          v1Key.Sha256,
+		ForceSendFields: v1Key.ForceSendFields,
+		NullFields:      v1Key.NullFields,
+	}
+}
+
+func convertV1DiskToAlphaDisk(v1Disk *computev1.Disk) *computealpha.Disk {
+	var dek *computealpha.CustomerEncryptionKey = nil
+
+	if v1Disk.DiskEncryptionKey != nil {
+		dek = convertV1CustomerEncryptionKeyToAlpha(v1Disk.DiskEncryptionKey)
+	}
+
+	// Note: this is an incomplete list. It only includes the fields we use for disk creation.
+	return &computealpha.Disk{
+		Name:              v1Disk.Name,
+		SizeGb:            v1Disk.SizeGb,
+		Description:       v1Disk.Description,
+		Type:              v1Disk.Type,
+		SourceSnapshot:    v1Disk.SourceSnapshot,
+		ReplicaZones:      v1Disk.ReplicaZones,
+		DiskEncryptionKey: dek,
+	}
+}
+
 func (cloud *CloudProvider) insertRegionalDisk(
 	ctx context.Context,
 	volKey *meta.Key,
@@ -299,7 +371,17 @@ func (cloud *CloudProvider) insertRegionalDisk(
 	capacityRange *csi.CapacityRange,
 	replicaZones []string,
 	snapshotID string,
-	description string) error {
+	description string,
+	multiWriter bool) error {
+	var (
+		err           error
+		opName        string
+		gceAPIVersion = GCEAPIVersionV1
+	)
+
+	if multiWriter {
+		gceAPIVersion = GCEAPIVersionAlpha
+	}
 
 	diskToCreate := &computev1.Disk{
 		Name:        volKey.Name,
@@ -319,16 +401,31 @@ func (cloud *CloudProvider) insertRegionalDisk(
 		}
 	}
 
-	insertOp, err := cloud.service.RegionDisks.Insert(cloud.project, volKey.Region, diskToCreate).Context(ctx).Do()
+	if gceAPIVersion == GCEAPIVersionAlpha {
+		var insertOp *computealpha.Operation
+		alphaDiskToCreate := convertV1DiskToAlphaDisk(diskToCreate)
+		alphaDiskToCreate.MultiWriter = multiWriter
+		insertOp, err = cloud.alphaService.RegionDisks.Insert(cloud.project, volKey.Region, alphaDiskToCreate).Context(ctx).Do()
+		if insertOp != nil {
+			opName = insertOp.Name
+		}
+	} else {
+		var insertOp *computev1.Operation
+		insertOp, err = cloud.service.RegionDisks.Insert(cloud.project, volKey.Region, diskToCreate).Context(ctx).Do()
+		if insertOp != nil {
+			opName = insertOp.Name
+		}
+	}
 	if err != nil {
 		if IsGCEError(err, "alreadyExists") {
-			disk, err := cloud.GetDisk(ctx, volKey)
+			disk, err := cloud.GetDisk(ctx, volKey, gceAPIVersion)
 			if err != nil {
 				return err
 			}
 			err = cloud.ValidateExistingDisk(ctx, disk, params,
 				int64(capacityRange.GetRequiredBytes()),
-				int64(capacityRange.GetLimitBytes()))
+				int64(capacityRange.GetLimitBytes()),
+				multiWriter)
 			if err != nil {
 				return err
 			}
@@ -338,16 +435,17 @@ func (cloud *CloudProvider) insertRegionalDisk(
 		return status.Error(codes.Internal, fmt.Sprintf("unknown Insert disk error: %v", err))
 	}
 
-	err = cloud.waitForRegionalOp(ctx, insertOp, volKey.Region)
+	err = cloud.waitForRegionalOp(ctx, opName, volKey.Region)
 	if err != nil {
 		if IsGCEError(err, "alreadyExists") {
-			disk, err := cloud.GetDisk(ctx, volKey)
+			disk, err := cloud.GetDisk(ctx, volKey, gceAPIVersion)
 			if err != nil {
 				return err
 			}
 			err = cloud.ValidateExistingDisk(ctx, disk, params,
 				int64(capacityRange.GetRequiredBytes()),
-				int64(capacityRange.GetLimitBytes()))
+				int64(capacityRange.GetLimitBytes()),
+				multiWriter)
 			if err != nil {
 				return err
 			}
@@ -366,7 +464,18 @@ func (cloud *CloudProvider) insertZonalDisk(
 	capBytes int64,
 	capacityRange *csi.CapacityRange,
 	snapshotID string,
-	description string) error {
+	description string,
+	multiWriter bool) error {
+	var (
+		err           error
+		opName        string
+		gceAPIVersion = GCEAPIVersionV1
+	)
+
+	if multiWriter {
+		gceAPIVersion = GCEAPIVersionAlpha
+	}
+
 	diskToCreate := &computev1.Disk{
 		Name:        volKey.Name,
 		SizeGb:      common.BytesToGb(capBytes),
@@ -384,17 +493,32 @@ func (cloud *CloudProvider) insertZonalDisk(
 		}
 	}
 
-	op, err := cloud.service.Disks.Insert(cloud.project, volKey.Zone, diskToCreate).Context(ctx).Do()
+	if gceAPIVersion == GCEAPIVersionAlpha {
+		var insertOp *computealpha.Operation
+		alphaDiskToCreate := convertV1DiskToAlphaDisk(diskToCreate)
+		alphaDiskToCreate.MultiWriter = multiWriter
+		insertOp, err = cloud.alphaService.Disks.Insert(cloud.project, volKey.Zone, alphaDiskToCreate).Context(ctx).Do()
+		if insertOp != nil {
+			opName = insertOp.Name
+		}
+	} else {
+		var insertOp *computev1.Operation
+		insertOp, err = cloud.service.Disks.Insert(cloud.project, volKey.Zone, diskToCreate).Context(ctx).Do()
+		if insertOp != nil {
+			opName = insertOp.Name
+		}
+	}
 
 	if err != nil {
 		if IsGCEError(err, "alreadyExists") {
-			disk, err := cloud.GetDisk(ctx, volKey)
+			disk, err := cloud.GetDisk(ctx, volKey, gceAPIVersion)
 			if err != nil {
 				return err
 			}
 			err = cloud.ValidateExistingDisk(ctx, disk, params,
 				int64(capacityRange.GetRequiredBytes()),
-				int64(capacityRange.GetLimitBytes()))
+				int64(capacityRange.GetLimitBytes()),
+				multiWriter)
 			if err != nil {
 				return err
 			}
@@ -404,17 +528,18 @@ func (cloud *CloudProvider) insertZonalDisk(
 		return fmt.Errorf("unknown Insert disk error: %v", err)
 	}
 
-	err = cloud.waitForZonalOp(ctx, op, volKey.Zone)
+	err = cloud.waitForZonalOp(ctx, opName, volKey.Zone)
 
 	if err != nil {
 		if IsGCEError(err, "alreadyExists") {
-			disk, err := cloud.GetDisk(ctx, volKey)
+			disk, err := cloud.GetDisk(ctx, volKey, gceAPIVersion)
 			if err != nil {
 				return err
 			}
 			err = cloud.ValidateExistingDisk(ctx, disk, params,
 				int64(capacityRange.GetRequiredBytes()),
-				int64(capacityRange.GetLimitBytes()))
+				int64(capacityRange.GetLimitBytes()),
+				multiWriter)
 			if err != nil {
 				return err
 			}
@@ -447,7 +572,7 @@ func (cloud *CloudProvider) deleteZonalDisk(ctx context.Context, zone, name stri
 		}
 		return err
 	}
-	err = cloud.waitForZonalOp(ctx, op, zone)
+	err = cloud.waitForZonalOp(ctx, op.Name, zone)
 	if err != nil {
 		return err
 	}
@@ -463,7 +588,7 @@ func (cloud *CloudProvider) deleteRegionalDisk(ctx context.Context, region, name
 		}
 		return err
 	}
-	err = cloud.waitForRegionalOp(ctx, op, region)
+	err = cloud.waitForRegionalOp(ctx, op.Name, region)
 	if err != nil {
 		return err
 	}
@@ -490,7 +615,7 @@ func (cloud *CloudProvider) AttachDisk(ctx context.Context, volKey *meta.Key, re
 	if err != nil {
 		return fmt.Errorf("failed cloud service attach disk call: %v", err)
 	}
-	err = cloud.waitForZonalOp(ctx, op, instanceZone)
+	err = cloud.waitForZonalOp(ctx, op.Name, instanceZone)
 	if err != nil {
 		return fmt.Errorf("failed when waiting for zonal op: %v", err)
 	}
@@ -503,7 +628,7 @@ func (cloud *CloudProvider) DetachDisk(ctx context.Context, deviceName, instance
 	if err != nil {
 		return err
 	}
-	err = cloud.waitForZonalOp(ctx, op, instanceZone)
+	err = cloud.waitForZonalOp(ctx, op.Name, instanceZone)
 	if err != nil {
 		return err
 	}
@@ -556,13 +681,14 @@ func (cloud *CloudProvider) getRegionalDiskTypeURI(region, diskType string) stri
 	return cloud.service.BasePath + fmt.Sprintf(diskTypeURITemplateRegional, cloud.project, region, diskType)
 }
 
-func (cloud *CloudProvider) waitForZonalOp(ctx context.Context, op *computev1.Operation, zone string) error {
+func (cloud *CloudProvider) waitForZonalOp(ctx context.Context, opName string, zone string) error {
+	// The v1 API can query for v1, alpha, or beta operations.
 	svc := cloud.service
 	project := cloud.project
 	return wait.Poll(3*time.Second, 5*time.Minute, func() (bool, error) {
-		pollOp, err := svc.ZoneOperations.Get(project, zone, op.Name).Context(ctx).Do()
+		pollOp, err := svc.ZoneOperations.Get(project, zone, opName).Context(ctx).Do()
 		if err != nil {
-			klog.Errorf("WaitForOp(op: %#v, zone: %#v) failed to poll the operation", op, zone)
+			klog.Errorf("WaitForOp(op: %s, zone: %#v) failed to poll the operation", opName, zone)
 			return false, err
 		}
 		done, err := opIsDone(pollOp)
@@ -570,11 +696,12 @@ func (cloud *CloudProvider) waitForZonalOp(ctx context.Context, op *computev1.Op
 	})
 }
 
-func (cloud *CloudProvider) waitForRegionalOp(ctx context.Context, op *computev1.Operation, region string) error {
+func (cloud *CloudProvider) waitForRegionalOp(ctx context.Context, opName string, region string) error {
+	// The v1 API can query for v1, alpha, or beta operations.
 	return wait.Poll(3*time.Second, 5*time.Minute, func() (bool, error) {
-		pollOp, err := cloud.service.RegionOperations.Get(cloud.project, region, op.Name).Context(ctx).Do()
+		pollOp, err := cloud.service.RegionOperations.Get(cloud.project, region, opName).Context(ctx).Do()
 		if err != nil {
-			klog.Errorf("WaitForOp(op: %#v, region: %#v) failed to poll the operation", op, region)
+			klog.Errorf("WaitForOp(op: %s, region: %#v) failed to poll the operation", opName, region)
 			return false, err
 		}
 		done, err := opIsDone(pollOp)
@@ -582,13 +709,13 @@ func (cloud *CloudProvider) waitForRegionalOp(ctx context.Context, op *computev1
 	})
 }
 
-func (cloud *CloudProvider) waitForGlobalOp(ctx context.Context, op *computev1.Operation) error {
+func (cloud *CloudProvider) waitForGlobalOp(ctx context.Context, opName string) error {
 	svc := cloud.service
 	project := cloud.project
 	return wait.Poll(3*time.Second, 5*time.Minute, func() (bool, error) {
-		pollOp, err := svc.GlobalOperations.Get(project, op.Name).Context(ctx).Do()
+		pollOp, err := svc.GlobalOperations.Get(project, opName).Context(ctx).Do()
 		if err != nil {
-			klog.Errorf("waitForGlobalOp(op: %#v) failed to poll the operation", op)
+			klog.Errorf("waitForGlobalOp(op: %s) failed to poll the operation", opName)
 			return false, err
 		}
 		done, err := opIsDone(pollOp)
@@ -601,7 +728,7 @@ func (cloud *CloudProvider) WaitForAttach(ctx context.Context, volKey *meta.Key,
 	start := time.Now()
 	return wait.Poll(5*time.Second, 2*time.Minute, func() (bool, error) {
 		klog.V(6).Infof("Polling for attach of disk %v to instance %v to complete for %v", volKey.Name, instanceName, time.Since(start))
-		disk, err := cloud.GetDisk(ctx, volKey)
+		disk, err := cloud.GetDisk(ctx, volKey, GCEAPIVersionV1)
 		if err != nil {
 			return false, fmt.Errorf("GetDisk failed to get disk: %v", err)
 		}
@@ -661,7 +788,7 @@ func (cloud *CloudProvider) DeleteSnapshot(ctx context.Context, snapshotName str
 		}
 		return err
 	}
-	err = cloud.waitForGlobalOp(ctx, op)
+	err = cloud.waitForGlobalOp(ctx, op.Name)
 	if err != nil {
 		return err
 	}
@@ -686,7 +813,7 @@ func (cloud *CloudProvider) CreateSnapshot(ctx context.Context, volKey *meta.Key
 // k8s.io/apimachinery/quantity package for better size handling
 func (cloud *CloudProvider) ResizeDisk(ctx context.Context, volKey *meta.Key, requestBytes int64) (int64, error) {
 	klog.V(5).Infof("Resizing disk %v to size %v", volKey, requestBytes)
-	cloudDisk, err := cloud.GetDisk(ctx, volKey)
+	cloudDisk, err := cloud.GetDisk(ctx, volKey, GCEAPIVersionV1)
 	if err != nil {
 		return -1, fmt.Errorf("failed to get disk: %v", err)
 	}
@@ -719,7 +846,7 @@ func (cloud *CloudProvider) resizeZonalDisk(ctx context.Context, volKey *meta.Ke
 		return -1, fmt.Errorf("failed to resize zonal volume %v: %v", volKey.String(), err)
 	}
 
-	err = cloud.waitForZonalOp(ctx, op, volKey.Zone)
+	err = cloud.waitForZonalOp(ctx, op.Name, volKey.Zone)
 	if err != nil {
 		return -1, fmt.Errorf("failed waiting for op for zonal resize for %s: %v", volKey.String(), err)
 	}
@@ -737,7 +864,7 @@ func (cloud *CloudProvider) resizeRegionalDisk(ctx context.Context, volKey *meta
 		return -1, fmt.Errorf("failed to resize regional volume %v: %v", volKey.String(), err)
 	}
 
-	err = cloud.waitForRegionalOp(ctx, op, volKey.Region)
+	err = cloud.waitForRegionalOp(ctx, op.Name, volKey.Region)
 	if err != nil {
 		return -1, fmt.Errorf("failed waiting for op for regional resize for %s: %v", volKey.String(), err)
 	}
diff --git a/pkg/gce-cloud-provider/compute/gce.go b/pkg/gce-cloud-provider/compute/gce.go
index aaee99ba..0686a68e 100644
--- a/pkg/gce-cloud-provider/compute/gce.go
+++ b/pkg/gce-cloud-provider/compute/gce.go
@@ -27,6 +27,7 @@ import (
 
 	"cloud.google.com/go/compute/metadata"
 	"golang.org/x/oauth2"
+	alpha "google.golang.org/api/compute/v0.alpha"
 	"google.golang.org/api/compute/v1"
 	"google.golang.org/api/googleapi"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -42,16 +43,18 @@ const (
 
 	regionURITemplate = "projects/%s/regions/%s"
 
-	GCEComputeAPIEndpoint     = "https://www.googleapis.com/compute/v1/"
-	GCEComputeBetaAPIEndpoint = "https://www.googleapis.com/compute/beta/"
+	GCEComputeAPIEndpoint      = "https://www.googleapis.com/compute/v1/"
+	GCEComputeBetaAPIEndpoint  = "https://www.googleapis.com/compute/beta/"
+	GCEComputeAlphaAPIEndpoint = "https://www.googleapis.com/compute/alpha/"
 
 	replicaZoneURITemplateSingleZone = "%s/zones/%s" // {gce.projectID}/zones/{disk.Zone}
 )
 
 type CloudProvider struct {
-	service *compute.Service
-	project string
-	zone    string
+	service      *compute.Service
+	alphaService *alpha.Service
+	project      string
+	zone         string
 
 	zonesCache map[string]([]string)
 }
@@ -89,16 +92,22 @@ func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath s
 		return nil, err
 	}
 
+	alphasvc, err := createAlphaCloudService(ctx, vendorVersion, tokenSource)
+	if err != nil {
+		return nil, err
+	}
+
 	project, zone, err := getProjectAndZone(configFile)
 	if err != nil {
 		return nil, fmt.Errorf("Failed getting Project and Zone: %v", err)
 	}
 
 	return &CloudProvider{
-		service:    svc,
-		project:    project,
-		zone:       zone,
-		zonesCache: make(map[string]([]string)),
+		service:      svc,
+		alphaService: alphasvc,
+		project:      project,
+		zone:         zone,
+		zonesCache:   make(map[string]([]string)),
 	}, nil
 
 }
@@ -149,6 +158,19 @@ func readConfig(configPath string) (*ConfigFile, error) {
 	return cfg, nil
 }
 
+func createAlphaCloudService(ctx context.Context, vendorVersion string, tokenSource oauth2.TokenSource) (*alpha.Service, error) {
+	client, err := newOauthClient(ctx, tokenSource)
+	if err != nil {
+		return nil, err
+	}
+	service, err := alpha.New(client)
+	if err != nil {
+		return nil, err
+	}
+	service.UserAgent = fmt.Sprintf("GCE CSI Driver/%s (%s %s)", vendorVersion, runtime.GOOS, runtime.GOARCH)
+	return service, nil
+}
+
 func createCloudService(ctx context.Context, vendorVersion string, tokenSource oauth2.TokenSource) (*compute.Service, error) {
 	svc, err := createCloudServiceWithDefaultServiceAccount(ctx, vendorVersion, tokenSource)
 	return svc, err
diff --git a/pkg/gce-pd-csi-driver/controller.go b/pkg/gce-pd-csi-driver/controller.go
index 8be34ab6..713eb3ba 100644
--- a/pkg/gce-pd-csi-driver/controller.go
+++ b/pkg/gce-pd-csi-driver/controller.go
@@ -114,6 +114,12 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 	if err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "failed to extract parameters: %v", err)
 	}
+	// Determine multiWriter
+	gceAPIVersion := gce.GCEAPIVersionV1
+	multiWriter, _ := getMultiWriterFromCapabilities(volumeCapabilities)
+	if multiWriter {
+		gceAPIVersion = gce.GCEAPIVersionAlpha
+	}
 	// Determine the zone or zones+region of the disk
 	var zones []string
 	var volKey *meta.Key
@@ -152,7 +158,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 	defer gceCS.volumeLocks.Release(volumeID)
 
 	// Validate if disk already exists
-	existingDisk, err := gceCS.CloudProvider.GetDisk(ctx, volKey)
+	existingDisk, err := gceCS.CloudProvider.GetDisk(ctx, volKey, gceAPIVersion)
 	if err != nil {
 		if !gce.IsGCEError(err, "notFound") {
 			return nil, status.Error(codes.Internal, fmt.Sprintf("CreateVolume unknown get disk error when validating: %v", err))
@@ -162,7 +168,8 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 		// There was no error so we want to validate the disk that we find
 		err = gceCS.CloudProvider.ValidateExistingDisk(ctx, existingDisk, params,
 			int64(capacityRange.GetRequiredBytes()),
-			int64(capacityRange.GetLimitBytes()))
+			int64(capacityRange.GetLimitBytes()),
+			multiWriter)
 		if err != nil {
 			return nil, status.Error(codes.AlreadyExists, fmt.Sprintf("CreateVolume disk already exists with same name and is incompatible: %v", err))
 		}
@@ -205,7 +212,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 		if len(zones) != 1 {
 			return nil, status.Error(codes.Internal, fmt.Sprintf("CreateVolume failed to get a single zone for creating zonal disk, instead got: %v", zones))
 		}
-		disk, err = createSingleZoneDisk(ctx, gceCS.CloudProvider, name, zones, params, capacityRange, capBytes, snapshotID)
+		disk, err = createSingleZoneDisk(ctx, gceCS.CloudProvider, name, zones, params, capacityRange, capBytes, snapshotID, multiWriter)
 		if err != nil {
 			return nil, status.Error(codes.Internal, fmt.Sprintf("CreateVolume failed to create single zonal disk %#v: %v", name, err))
 		}
@@ -213,7 +220,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 		if len(zones) != 2 {
 			return nil, status.Errorf(codes.Internal, fmt.Sprintf("CreateVolume failed to get a 2 zones for creating regional disk, instead got: %v", zones))
 		}
-		disk, err = createRegionalDisk(ctx, gceCS.CloudProvider, name, zones, params, capacityRange, capBytes, snapshotID)
+		disk, err = createRegionalDisk(ctx, gceCS.CloudProvider, name, zones, params, capacityRange, capBytes, snapshotID, multiWriter)
 		if err != nil {
 			return nil, status.Error(codes.Internal, fmt.Sprintf("CreateVolume failed to create regional disk %#v: %v", name, err))
 		}
@@ -318,7 +325,7 @@ func (gceCS *GCEControllerServer) ControllerPublishVolume(ctx context.Context, r
 		PublishContext: nil,
 	}
 
-	_, err = gceCS.CloudProvider.GetDisk(ctx, volKey)
+	_, err = gceCS.CloudProvider.GetDisk(ctx, volKey, gce.GCEAPIVersionV1)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
 			return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find disk %v: %v", volKey.String(), err))
@@ -452,7 +459,7 @@ func (gceCS *GCEControllerServer) ValidateVolumeCapabilities(ctx context.Context
 	}
 	defer gceCS.volumeLocks.Release(volumeID)
 
-	disk, err := gceCS.CloudProvider.GetDisk(ctx, volKey)
+	disk, err := gceCS.CloudProvider.GetDisk(ctx, volKey, gce.GCEAPIVersionV1)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
 			return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find disk %v: %v", volKey.Name, err))
@@ -573,7 +580,7 @@ func (gceCS *GCEControllerServer) CreateSnapshot(ctx context.Context, req *csi.C
 	defer gceCS.volumeLocks.Release(volumeID)
 
 	// Check if volume exists
-	_, err = gceCS.CloudProvider.GetDisk(ctx, volKey)
+	_, err = gceCS.CloudProvider.GetDisk(ctx, volKey, gce.GCEAPIVersionV1)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
 			return nil, status.Error(codes.NotFound, fmt.Sprintf("CreateSnapshot could not find disk %v: %v", volKey.String(), err))
@@ -1006,10 +1013,11 @@ func generateCreateVolumeResponse(disk *gce.CloudDisk, capBytes int64, zones []s
 
 func cleanSelfLink(selfLink string) string {
 	temp := strings.TrimPrefix(selfLink, gce.GCEComputeAPIEndpoint)
-	return strings.TrimPrefix(temp, gce.GCEComputeBetaAPIEndpoint)
+	temp = strings.TrimPrefix(temp, gce.GCEComputeBetaAPIEndpoint)
+	return strings.TrimPrefix(temp, gce.GCEComputeAlphaAPIEndpoint)
 }
 
-func createRegionalDisk(ctx context.Context, cloudProvider gce.GCECompute, name string, zones []string, params common.DiskParameters, capacityRange *csi.CapacityRange, capBytes int64, snapshotID string) (*gce.CloudDisk, error) {
+func createRegionalDisk(ctx context.Context, cloudProvider gce.GCECompute, name string, zones []string, params common.DiskParameters, capacityRange *csi.CapacityRange, capBytes int64, snapshotID string, multiWriter bool) (*gce.CloudDisk, error) {
 	region, err := common.GetRegionFromZones(zones)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get region from zones: %v", err)
@@ -1021,28 +1029,38 @@ func createRegionalDisk(ctx context.Context, cloudProvider gce.GCECompute, name
 			fullyQualifiedReplicaZones, cloudProvider.GetReplicaZoneURI(replicaZone))
 	}
 
-	err = cloudProvider.InsertDisk(ctx, meta.RegionalKey(name, region), params, capBytes, capacityRange, fullyQualifiedReplicaZones, snapshotID)
+	err = cloudProvider.InsertDisk(ctx, meta.RegionalKey(name, region), params, capBytes, capacityRange, fullyQualifiedReplicaZones, snapshotID, multiWriter)
 	if err != nil {
 		return nil, fmt.Errorf("failed to insert regional disk: %v", err)
 	}
 
-	disk, err := cloudProvider.GetDisk(ctx, meta.RegionalKey(name, region))
+	gceAPIVersion := gce.GCEAPIVersionV1
+	if multiWriter {
+		gceAPIVersion = gce.GCEAPIVersionAlpha
+	}
+
+	disk, err := cloudProvider.GetDisk(ctx, meta.RegionalKey(name, region), gceAPIVersion)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get disk after creating regional disk: %v", err)
 	}
 	return disk, nil
 }
 
-func createSingleZoneDisk(ctx context.Context, cloudProvider gce.GCECompute, name string, zones []string, params common.DiskParameters, capacityRange *csi.CapacityRange, capBytes int64, snapshotID string) (*gce.CloudDisk, error) {
+func createSingleZoneDisk(ctx context.Context, cloudProvider gce.GCECompute, name string, zones []string, params common.DiskParameters, capacityRange *csi.CapacityRange, capBytes int64, snapshotID string, multiWriter bool) (*gce.CloudDisk, error) {
 	if len(zones) != 1 {
 		return nil, fmt.Errorf("got wrong number of zones for zonal create volume: %v", len(zones))
 	}
 	diskZone := zones[0]
-	err := cloudProvider.InsertDisk(ctx, meta.ZonalKey(name, diskZone), params, capBytes, capacityRange, nil, snapshotID)
+	err := cloudProvider.InsertDisk(ctx, meta.ZonalKey(name, diskZone), params, capBytes, capacityRange, nil, snapshotID, multiWriter)
 	if err != nil {
 		return nil, fmt.Errorf("failed to insert zonal disk: %v", err)
 	}
-	disk, err := cloudProvider.GetDisk(ctx, meta.ZonalKey(name, diskZone))
+
+	gceAPIVersion := gce.GCEAPIVersionV1
+	if multiWriter {
+		gceAPIVersion = gce.GCEAPIVersionAlpha
+	}
+	disk, err := cloudProvider.GetDisk(ctx, meta.ZonalKey(name, diskZone), gceAPIVersion)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/gce-pd-csi-driver/gce-pd-driver.go b/pkg/gce-pd-csi-driver/gce-pd-driver.go
index 3c8d4e16..92c12cbc 100644
--- a/pkg/gce-pd-csi-driver/gce-pd-driver.go
+++ b/pkg/gce-pd-csi-driver/gce-pd-driver.go
@@ -54,6 +54,7 @@ func (gceDriver *GCEDriver) SetupGCEDriver(name, vendorVersion string, identityS
 	vcam := []csi.VolumeCapability_AccessMode_Mode{
 		csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
 		csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
+		csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER,
 	}
 	gceDriver.AddVolumeCapabilityAccessModes(vcam)
 	csc := []csi.ControllerServiceCapability_RPC_Type{
diff --git a/pkg/gce-pd-csi-driver/utils.go b/pkg/gce-pd-csi-driver/utils.go
index 76d5be9c..32ce2993 100644
--- a/pkg/gce-pd-csi-driver/utils.go
+++ b/pkg/gce-pd-csi-driver/utils.go
@@ -95,12 +95,16 @@ func validateVolumeCapability(vc *csi.VolumeCapability) error {
 	}
 	blk := vc.GetBlock()
 	mnt := vc.GetMount()
+	mod := vc.GetAccessMode().GetMode()
 	if mnt == nil && blk == nil {
 		return errors.New("must specify an access type")
 	}
 	if mnt != nil && blk != nil {
 		return errors.New("specified both mount and block access types")
 	}
+	if mnt != nil && mod == csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER {
+		return errors.New("specified multi writer with mount access type")
+	}
 	return nil
 }
 
@@ -114,9 +118,32 @@ func validateAccessMode(am *csi.VolumeCapability_AccessMode) error {
 	case csi.VolumeCapability_AccessMode_SINGLE_NODE_READER_ONLY:
 	case csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY:
 	case csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER:
-		return errors.New("MULTI_NODE_MULTI_WRITER access mode is not yet supported for PD")
 	default:
 		return fmt.Errorf("%v access mode is not supported for for PD", am.GetMode())
 	}
 	return nil
 }
+
+func getMultiWriterFromCapability(vc *csi.VolumeCapability) (bool, error) {
+	if vc.GetAccessMode() == nil {
+		return false, errors.New("access mode is nil")
+	}
+	mode := vc.GetAccessMode().GetMode()
+	return (mode == csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER), nil
+}
+
+func getMultiWriterFromCapabilities(vcs []*csi.VolumeCapability) (bool, error) {
+	if vcs == nil {
+		return false, errors.New("volume capabilities is nil")
+	}
+	for _, vc := range vcs {
+		multiWriter, err := getMultiWriterFromCapability(vc)
+		if err != nil {
+			return false, err
+		}
+		if multiWriter {
+			return true, nil
+		}
+	}
+	return false, nil
+}


Test patch:
diff --git a/pkg/gce-pd-csi-driver/controller_test.go b/pkg/gce-pd-csi-driver/controller_test.go
index c415144e..1e7bf863 100644
--- a/pkg/gce-pd-csi-driver/controller_test.go
+++ b/pkg/gce-pd-csi-driver/controller_test.go
@@ -346,7 +346,7 @@ func TestCreateVolumeArguments(t *testing.T) {
 			},
 		},
 		{
-			name: "fail with MULTI_NODE_MULTI_WRITER capability",
+			name: "fail with mount/MULTI_NODE_MULTI_WRITER capabilities",
 			req: &csi.CreateVolumeRequest{
 				Name:               "test-name",
 				CapacityRange:      stdCapRange,
@@ -355,6 +355,21 @@ func TestCreateVolumeArguments(t *testing.T) {
 			},
 			expErrCode: codes.InvalidArgument,
 		},
+		{
+			name: "success with block/MULTI_NODE_MULTI_WRITER capabilities",
+			req: &csi.CreateVolumeRequest{
+				Name:               "test-name",
+				CapacityRange:      stdCapRange,
+				VolumeCapabilities: createBlockVolumeCapabilities(csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER),
+				Parameters:         stdParams,
+			},
+			expVol: &csi.Volume{
+				CapacityBytes:      common.GbToBytes(20),
+				VolumeId:           testVolumeID,
+				VolumeContext:      nil,
+				AccessibleTopology: stdTopology,
+			},
+		},
 		{
 			name: "fail no name",
 			req: &csi.CreateVolumeRequest{
@@ -605,18 +620,9 @@ func TestCreateVolumeArguments(t *testing.T) {
 		{
 			name: "success with block volume capability",
 			req: &csi.CreateVolumeRequest{
-				Name:          name,
-				CapacityRange: stdCapRange,
-				VolumeCapabilities: []*csi.VolumeCapability{
-					{
-						AccessType: &csi.VolumeCapability_Block{
-							Block: &csi.VolumeCapability_BlockVolume{},
-						},
-						AccessMode: &csi.VolumeCapability_AccessMode{
-							Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
-						},
-					},
-				},
+				Name:               name,
+				CapacityRange:      stdCapRange,
+				VolumeCapabilities: createBlockVolumeCapabilities(csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER),
 			},
 			expVol: &csi.Volume{
 				CapacityBytes:      common.GbToBytes(20),
diff --git a/pkg/gce-pd-csi-driver/node_test.go b/pkg/gce-pd-csi-driver/node_test.go
index 8d1cda22..8ff391b2 100644
--- a/pkg/gce-pd-csi-driver/node_test.go
+++ b/pkg/gce-pd-csi-driver/node_test.go
@@ -225,7 +225,7 @@ func TestNodePublishVolume(t *testing.T) {
 				TargetPath:        targetPath,
 				StagingTargetPath: stagingPath,
 				Readonly:          false,
-				VolumeCapability:  createVolumeCapability(csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER),
+				VolumeCapability:  createVolumeCapability(csi.VolumeCapability_AccessMode_UNKNOWN),
 			},
 			expErrCode: codes.InvalidArgument,
 		},
@@ -378,6 +378,15 @@ func TestNodeStageVolume(t *testing.T) {
 				VolumeCapability:  stdVolCap,
 			},
 		},
+		{
+			name: "Invalid request (Bad Access Mode)",
+			req: &csi.NodeStageVolumeRequest{
+				VolumeId:          volumeID,
+				StagingTargetPath: stagingPath,
+				VolumeCapability:  createVolumeCapability(csi.VolumeCapability_AccessMode_UNKNOWN),
+			},
+			expErrCode: codes.InvalidArgument,
+		},
 		{
 			name: "Invalid request (Bad Access Mode)",
 			req: &csi.NodeStageVolumeRequest{
diff --git a/pkg/gce-pd-csi-driver/utils_test.go b/pkg/gce-pd-csi-driver/utils_test.go
index bc90558a..53c134b1 100644
--- a/pkg/gce-pd-csi-driver/utils_test.go
+++ b/pkg/gce-pd-csi-driver/utils_test.go
@@ -54,6 +54,23 @@ func createVolumeCapability(am csi.VolumeCapability_AccessMode_Mode) *csi.Volume
 	}
 }
 
+func createBlockVolumeCapabilities(am csi.VolumeCapability_AccessMode_Mode) []*csi.VolumeCapability {
+	return []*csi.VolumeCapability{
+		createBlockVolumeCapability(am),
+	}
+}
+
+func createBlockVolumeCapability(am csi.VolumeCapability_AccessMode_Mode) *csi.VolumeCapability {
+	return &csi.VolumeCapability{
+		AccessType: &csi.VolumeCapability_Block{
+			Block: &csi.VolumeCapability_BlockVolume{},
+		},
+		AccessMode: &csi.VolumeCapability_AccessMode{
+			Mode: am,
+		},
+	}
+}
+
 func TestValidateVolumeCapabilities(t *testing.T) {
 	testCases := []struct {
 		name   string
@@ -127,16 +144,16 @@ func TestValidateVolumeCapabilities(t *testing.T) {
 		},
 		{
 			name: "success with block capabilities",
-			vc: []*csi.VolumeCapability{
-				{
-					AccessType: &csi.VolumeCapability_Block{
-						Block: &csi.VolumeCapability_BlockVolume{},
-					},
-					AccessMode: &csi.VolumeCapability_AccessMode{
-						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
-					},
-				},
-			},
+			vc:   createBlockVolumeCapabilities(csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER),
+		},
+		{
+			name: "success with block/MULTI_NODE_MULTI_WRITER capabilities",
+			vc:   createBlockVolumeCapabilities(csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER),
+		},
+		{
+			name:   "fail with block/MULTI_NODE_SINGLE_WRITER capabilities",
+			vc:     createBlockVolumeCapabilities(csi.VolumeCapability_AccessMode_MULTI_NODE_SINGLE_WRITER),
+			expErr: true,
 		},
 		{
 			name: "success with reader + writer capabilities",
@@ -164,5 +181,56 @@ func TestValidateVolumeCapabilities(t *testing.T) {
 			t.Fatalf("Did not expect error but got: %v", err)
 		}
 	}
+}
+
+func TestGetMultiWriterFromCapabilities(t *testing.T) {
+	testCases := []struct {
+		name   string
+		vc     []*csi.VolumeCapability
+		expVal bool
+		expErr bool
+	}{
+		{
+			name:   "false with empty capabilities",
+			vc:     []*csi.VolumeCapability{},
+			expVal: false,
+		},
+		{
+			name: "fail with capabilities no access mode",
+			vc: []*csi.VolumeCapability{
+				{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+				},
+			},
+			expErr: true,
+		},
+		{
+			name:   "false with mount/SINGLE_NODE_WRITER capabilities",
+			vc:     createVolumeCapabilities(csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER),
+			expVal: false,
+		},
+		{
+			name:   "true with block/MULTI_NODE_MULTI_WRITER capabilities",
+			vc:     createBlockVolumeCapabilities(csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER),
+			expVal: true,
+		},
+	}
 
+	for _, tc := range testCases {
+		t.Logf("Running test: %v", tc.name)
+		val, err := getMultiWriterFromCapabilities(tc.vc)
+		if tc.expErr && err == nil {
+			t.Fatalf("Expected error but didn't get any")
+		}
+		if !tc.expErr && err != nil {
+			t.Fatalf("Did not expect error but got: %v", err)
+		}
+		if err != nil {
+			if tc.expVal != val {
+				t.Fatalf("Expected '%t' but got '%t'", tc.expVal, val)
+			}
+		}
+	}
 }
diff --git a/test/e2e/tests/multi_zone_e2e_test.go b/test/e2e/tests/multi_zone_e2e_test.go
index a38a09dd..991072ce 100644
--- a/test/e2e/tests/multi_zone_e2e_test.go
+++ b/test/e2e/tests/multi_zone_e2e_test.go
@@ -257,7 +257,11 @@ func testLifecycleWithVerify(volID string, volName string, instance *remote.Inst
 	if secondMountVerify != nil {
 		// Mount disk somewhere else
 		secondPublishDir := filepath.Join("/tmp/", volName, "secondmount")
-		err = client.NodePublishVolume(volID, stageDir, secondPublishDir)
+		if useBlock {
+			err = client.NodePublishBlockVolume(volID, stageDir, secondPublishDir)
+		} else {
+			err = client.NodePublishVolume(volID, stageDir, secondPublishDir)
+		}
 		if err != nil {
 			return fmt.Errorf("NodePublishVolume failed with error: %v", err)
 		}
@@ -269,7 +273,10 @@ func testLifecycleWithVerify(volID string, volName string, instance *remote.Inst
 		b := verifyArgs{
 			publishDir: secondPublishDir,
 		}
-		secondMountVerify(b)
+		err = secondMountVerify(b)
+		if err != nil {
+			return fmt.Errorf("failed to verify after second mount to %s: %v", publishDir, err)
+		}
 
 		// Unmount Disk
 		err = client.NodeUnpublishVolume(volID, secondPublishDir)
diff --git a/test/e2e/tests/setup_e2e_test.go b/test/e2e/tests/setup_e2e_test.go
index 1b067939..ce15319f 100644
--- a/test/e2e/tests/setup_e2e_test.go
+++ b/test/e2e/tests/setup_e2e_test.go
@@ -25,6 +25,7 @@ import (
 	cloudkms "cloud.google.com/go/kms/apiv1"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	computealpha "google.golang.org/api/compute/v0.alpha"
 	compute "google.golang.org/api/compute/v1"
 	"k8s.io/klog"
 	testutils "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/test/e2e/utils"
@@ -37,9 +38,10 @@ var (
 	runInProw       = flag.Bool("run-in-prow", false, "If true, use a Boskos loaned project and special CI service accounts and ssh keys")
 	deleteInstances = flag.Bool("delete-instances", false, "Delete the instances after tests run")
 
-	testContexts   = []*remote.TestContext{}
-	computeService *compute.Service
-	kmsClient      *cloudkms.KeyManagementClient
+	testContexts        = []*remote.TestContext{}
+	computeService      *compute.Service
+	computeAlphaService *computealpha.Service
+	kmsClient           *cloudkms.KeyManagementClient
 )
 
 func init() {
@@ -65,6 +67,9 @@ var _ = BeforeSuite(func() {
 	computeService, err = remote.GetComputeClient()
 	Expect(err).To(BeNil())
 
+	computeAlphaService, err = remote.GetComputeAlphaClient()
+	Expect(err).To(BeNil())
+
 	// Create the KMS client.
 	kmsClient, err = cloudkms.NewKeyManagementClient(context.Background())
 	Expect(err).To(BeNil())
diff --git a/test/e2e/tests/single_zone_e2e_test.go b/test/e2e/tests/single_zone_e2e_test.go
index 51196ce0..3ee47992 100644
--- a/test/e2e/tests/single_zone_e2e_test.go
+++ b/test/e2e/tests/single_zone_e2e_test.go
@@ -44,6 +44,7 @@ const (
 
 	defaultSizeGb      int64 = 5
 	defaultRepdSizeGb  int64 = 200
+	defaultMwSizeGb    int64 = 200
 	readyState               = "READY"
 	standardDiskType         = "pd-standard"
 	ssdDiskType              = "pd-ssd"
@@ -676,6 +677,75 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		Expect(err).To(BeNil(), "Failed to go through volume lifecycle")
 	})
 
+	// Pending while multi-writer feature is in Alpha
+	PIt("Should create and delete multi-writer disk", func() {
+		Expect(testContexts).ToNot(BeEmpty())
+		testContext := getRandomTestContext()
+
+		p, _, _ := testContext.Instance.GetIdentity()
+		client := testContext.Client
+
+		// Hardcode to us-east1-a while feature is in alpha
+		zone := "us-east1-a"
+
+		// Create and Validate Disk
+		volName, volID := createAndValidateUniqueZonalMultiWriterDisk(client, p, zone)
+
+		defer func() {
+			// Delete Disk
+			err := client.DeleteVolume(volID)
+			Expect(err).To(BeNil(), "DeleteVolume failed")
+
+			// Validate Disk Deleted
+			_, err = computeAlphaService.Disks.Get(p, zone, volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found")
+		}()
+	})
+
+	// Pending while multi-writer feature is in Alpha
+	PIt("Should complete entire disk lifecycle with multi-writer disk", func() {
+		testContext := getRandomTestContext()
+
+		p, z, _ := testContext.Instance.GetIdentity()
+		client := testContext.Client
+		instance := testContext.Instance
+
+		// Create and Validate Disk
+		volName, volID := createAndValidateUniqueZonalMultiWriterDisk(client, p, z)
+
+		defer func() {
+			// Delete Disk
+			err := client.DeleteVolume(volID)
+			Expect(err).To(BeNil(), "DeleteVolume failed")
+
+			// Validate Disk Deleted
+			_, err = computeService.Disks.Get(p, z, volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found")
+		}()
+
+		// Attach Disk
+		testFileContents := "test"
+		writeFunc := func(a verifyArgs) error {
+			err := testutils.WriteBlock(instance, a.publishDir, testFileContents)
+			if err != nil {
+				return fmt.Errorf("Failed to write file: %v", err)
+			}
+			return nil
+		}
+		verifyReadFunc := func(a verifyArgs) error {
+			readContents, err := testutils.ReadBlock(instance, a.publishDir, len(testFileContents))
+			if err != nil {
+				return fmt.Errorf("ReadFile failed with error: %v", err)
+			}
+			if strings.TrimSpace(string(readContents)) != testFileContents {
+				return fmt.Errorf("wanted test file content: %s, got content: %s", testFileContents, readContents)
+			}
+			return nil
+		}
+		err := testLifecycleWithVerify(volID, volName, instance, client, false /* readOnly */, true /* block */, writeFunc, verifyReadFunc)
+		Expect(err).To(BeNil(), "Failed to go through volume lifecycle")
+	})
+
 	It("Should successfully create disk with PVC/PV tags", func() {
 		Expect(testContexts).ToNot(BeEmpty())
 		testContext := getRandomTestContext()
@@ -756,3 +826,38 @@ func deleteVolumeOrError(client *remote.CsiClient, volID, project string) {
 	_, err = computeService.Disks.Get(project, key.Zone, key.Name).Do()
 	Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found")
 }
+
+func createAndValidateUniqueZonalMultiWriterDisk(client *remote.CsiClient, project, zone string) (string, string) {
+	// Create Disk
+	volName := testNamePrefix + string(uuid.NewUUID())
+	volID, err := client.CreateVolumeWithCaps(volName, nil, defaultMwSizeGb,
+		&csi.TopologyRequirement{
+			Requisite: []*csi.Topology{
+				{
+					Segments: map[string]string{common.TopologyKeyZone: zone},
+				},
+			},
+		},
+		[]*csi.VolumeCapability{
+			{
+				AccessType: &csi.VolumeCapability_Block{
+					Block: &csi.VolumeCapability_BlockVolume{},
+				},
+				AccessMode: &csi.VolumeCapability_AccessMode{
+					Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER,
+				},
+			},
+		})
+	Expect(err).To(BeNil(), "CreateVolume failed with error: %v", err)
+
+	// Validate Disk Created
+	cloudDisk, err := computeAlphaService.Disks.Get(project, zone, volName).Do()
+	Expect(err).To(BeNil(), "Could not get disk from cloud directly")
+	Expect(cloudDisk.Type).To(ContainSubstring(standardDiskType))
+	Expect(cloudDisk.Status).To(Equal(readyState))
+	Expect(cloudDisk.SizeGb).To(Equal(defaultMwSizeGb))
+	Expect(cloudDisk.Name).To(Equal(volName))
+	Expect(cloudDisk.MultiWriter).To(Equal(true))
+
+	return volName, volID
+}
diff --git a/test/e2e/utils/utils.go b/test/e2e/utils/utils.go
index 4c311ad4..f13c3195 100644
--- a/test/e2e/utils/utils.go
+++ b/test/e2e/utils/utils.go
@@ -171,6 +171,23 @@ func ReadFile(instance *remote.InstanceInfo, filePath string) (string, error) {
 	return output, nil
 }
 
+func WriteBlock(instance *remote.InstanceInfo, path, fileContents string) error {
+	output, err := instance.SSHNoSudo("echo", fileContents, "|", "dd", "of="+path)
+	if err != nil {
+		return fmt.Errorf("failed to write test file %s. Output: %v, errror: %v", path, output, err)
+	}
+	return nil
+}
+
+func ReadBlock(instance *remote.InstanceInfo, path string, length int) (string, error) {
+	lengthStr := strconv.Itoa(length)
+	output, err := instance.SSHNoSudo("dd", "if="+path, "bs="+lengthStr, "count=1", "2>", "/dev/null")
+	if err != nil {
+		return "", fmt.Errorf("failed to read test file %s. Output: %v, errror: %v", path, output, err)
+	}
+	return output, nil
+}
+
 func GetFSSizeInGb(instance *remote.InstanceInfo, mountPath string) (int64, error) {
 	output, err := instance.SSH("df", "--output=size", "-BG", mountPath, "|", "awk", "'NR==2'")
 	if err != nil {
diff --git a/test/remote/client-wrappers.go b/test/remote/client-wrappers.go
index fb4366dc..152570ba 100644
--- a/test/remote/client-wrappers.go
+++ b/test/remote/client-wrappers.go
@@ -99,13 +99,13 @@ func (c *CsiClient) CloseConn() error {
 	return c.conn.Close()
 }
 
-func (c *CsiClient) CreateVolume(volName string, params map[string]string, sizeInGb int64, topReq *csipb.TopologyRequirement) (string, error) {
+func (c *CsiClient) CreateVolumeWithCaps(volName string, params map[string]string, sizeInGb int64, topReq *csipb.TopologyRequirement, caps []*csipb.VolumeCapability) (string, error) {
 	capRange := &csipb.CapacityRange{
 		RequiredBytes: common.GbToBytes(sizeInGb),
 	}
 	cvr := &csipb.CreateVolumeRequest{
 		Name:               volName,
-		VolumeCapabilities: stdVolCaps,
+		VolumeCapabilities: caps,
 		Parameters:         params,
 		CapacityRange:      capRange,
 	}
@@ -119,6 +119,10 @@ func (c *CsiClient) CreateVolume(volName string, params map[string]string, sizeI
 	return cresp.GetVolume().GetVolumeId(), nil
 }
 
+func (c *CsiClient) CreateVolume(volName string, params map[string]string, sizeInGb int64, topReq *csipb.TopologyRequirement) (string, error) {
+	return c.CreateVolumeWithCaps(volName, params, sizeInGb, topReq, stdVolCaps)
+}
+
 func (c *CsiClient) DeleteVolume(volId string) error {
 	dvr := &csipb.DeleteVolumeRequest{
 		VolumeId: volId,
diff --git a/test/remote/instance.go b/test/remote/instance.go
index a078c672..af4f0eea 100644
--- a/test/remote/instance.go
+++ b/test/remote/instance.go
@@ -27,6 +27,7 @@ import (
 	"time"
 
 	"golang.org/x/oauth2/google"
+	computealpha "google.golang.org/api/compute/v0.alpha"
 	compute "google.golang.org/api/compute/v1"
 	"google.golang.org/api/googleapi"
 	"k8s.io/apimachinery/pkg/util/uuid"
@@ -281,6 +282,36 @@ func GetComputeClient() (*compute.Service, error) {
 	return nil, err
 }
 
+func GetComputeAlphaClient() (*computealpha.Service, error) {
+	const retries = 10
+	const backoff = time.Second * 6
+
+	klog.V(4).Infof("Getting compute client...")
+
+	// Setup the gce client for provisioning instances
+	// Getting credentials on gce jenkins is flaky, so try a couple times
+	var err error
+	var cs *computealpha.Service
+	for i := 0; i < retries; i++ {
+		if i > 0 {
+			time.Sleep(backoff)
+		}
+
+		var client *http.Client
+		client, err = google.DefaultClient(context.Background(), computealpha.ComputeScope)
+		if err != nil {
+			continue
+		}
+
+		cs, err = computealpha.New(client)
+		if err != nil {
+			continue
+		}
+		return cs, nil
+	}
+	return nil, err
+}
+
 func generateMetadataWithPublicKey(pubKeyFile string) (*compute.Metadata, error) {
 	publicKeyByte, err := ioutil.ReadFile(pubKeyFile)
 	if err != nil {