Repository: kubernetes-sigs/gcp-filestore-csi-driver
Pull request #229 (2022-02-25 04:30:05)

create volume changes
<!--  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**:
Layout the code for CSI create volume for multishare instances.
1. Multishare controller handles the CSI Create Volume call.
2. Multishare Ops manager handles cache consistency, manages lifecycle of instance and share ops.
3. cloud provider providers skeleton code for multishare APIs, and multi-share instance and share definitions.

**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**:

**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
NONE
```


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

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/pull/229#" 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-filestore-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-filestore-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":[]} -->
saikat-royc: /hold
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-filestore-csi-driver-verify | 6ffad6757adcf9d3a1c7d5598051158ac51b63d5 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-verify/1497066349338300416/) | true | `/test pull-gcp-filestore-csi-driver-verify`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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-filestore-csi-driver-verify | 6ffad6757adcf9d3a1c7d5598051158ac51b63d5 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-verify/1497066349338300416/) | true | `/test pull-gcp-filestore-csi-driver-verify`
pull-gcp-filestore-csi-driver-e2e | 6ffad6757adcf9d3a1c7d5598051158ac51b63d5 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1497066349246025728/) | true | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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 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-filestore-csi-driver-e2e | 19cfaa7a896f36d1bf7c9ad7a87ef60804d60896 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1497457993237663744/) | true | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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 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-filestore-csi-driver-e2e | 1c6a0b0c8400bec1c9496d46e0735968236c3680 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1497459672418881536/) | true | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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: /cc @leiyiz 
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-filestore-csi-driver-e2e | ca8b6a8a8332f15f6e803c262cdc16457868f5bb | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1498766323109335040/) | true | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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 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-filestore-csi-driver-e2e | f2aedaf82b88c4c41ce1dc186f02c4a0bdf59780 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1499205627178651648/) | true | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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-filestore-csi-driver-e2e | f2aedaf82b88c4c41ce1dc186f02c4a0bdf59780 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1499205627178651648/) | true | `/test pull-gcp-filestore-csi-driver-e2e`
pull-gcp-filestore-csi-driver-kubernetes-integration | f2aedaf82b88c4c41ce1dc186f02c4a0bdf59780 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-kubernetes-integration/1499205627287703552/) | false | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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-filestore-csi-driver-kubernetes-integration | f2aedaf82b88c4c41ce1dc186f02c4a0bdf59780 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-kubernetes-integration/1499205627287703552/) | false | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`
pull-gcp-filestore-csi-driver-e2e | 8ccc276d51d805037bcc813e71fea659b59ed5c9 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1499298429988245504/) | true | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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 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-filestore-csi-driver-e2e | 1bd6dbcc1d0036c96e2f2114da1ca48b36924e57 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1499501953707675648/) | true | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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 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-filestore-csi-driver-e2e | cf752391fae679cb79f0147db567d58a1e0a7bb8 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1499516674661945344/) | true | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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: /cc @leiyiz  ready for first round of reviews. more unit tests to be added.
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-filestore-csi-driver-kubernetes-integration | 19ec0e8dcf5a3133432f6dd8eba32a28322ba82b | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-kubernetes-integration/1499650016279531520/) | false | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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-filestore-csi-driver-kubernetes-integration | 19ec0e8dcf5a3133432f6dd8eba32a28322ba82b | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-kubernetes-integration/1499650016279531520/) | false | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`
pull-gcp-filestore-csi-driver-e2e | eb3b6e6df2df19fa500cebe59fe17c67d5c1de13 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1499965322545860608/) | true | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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-filestore-csi-driver-e2e | eb3b6e6df2df19fa500cebe59fe17c67d5c1de13 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1499965322545860608/) | true | `/test pull-gcp-filestore-csi-driver-e2e`
pull-gcp-filestore-csi-driver-unit | 0afd81d8a293afb0e4ad832c41e5c8f2cec30d28 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-unit/1501369312852578304/) | true | `/test pull-gcp-filestore-csi-driver-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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 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-filestore-csi-driver-unit | 7c42c73d37aaefe6769c87cbe8f1d9cd406193b8 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-unit/1501449406379986944/) | true | `/test pull-gcp-filestore-csi-driver-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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-filestore-csi-driver-unit | 7c42c73d37aaefe6769c87cbe8f1d9cd406193b8 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-unit/1501449406379986944/) | true | `/test pull-gcp-filestore-csi-driver-unit`
pull-gcp-filestore-csi-driver-e2e | 7c42c73d37aaefe6769c87cbe8f1d9cd406193b8 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1501449406321266688/) | true | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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-filestore-csi-driver-e2e | 7c42c73d37aaefe6769c87cbe8f1d9cd406193b8 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1501449406321266688/) | true | `/test pull-gcp-filestore-csi-driver-e2e`
pull-gcp-filestore-csi-driver-unit | 7c42c73d37aaefe6769c87cbe8f1d9cd406193b8 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-unit/1501670818294272000/) | true | `/test pull-gcp-filestore-csi-driver-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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: /cancel hold
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-filestore-csi-driver-e2e | 0a6ed880ace4295f50b66ae8a92556ef9b505c61 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1501687409669050368/) | true | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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 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-filestore-csi-driver-e2e | 0a6ed880ace4295f50b66ae8a92556ef9b505c61 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1501688140971118592/) | true | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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 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-filestore-csi-driver-e2e | 0a6ed880ace4295f50b66ae8a92556ef9b505c61 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1501720462491652096/) | true | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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 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-filestore-csi-driver-e2e | 6a02a604d32bf26d6701511931035de993f896dd | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/229/pull-gcp-filestore-csi-driver-e2e/1501804973258706944/) | true | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=229). [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-filestore-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/cmd/main.go b/cmd/main.go
index a204b365..8ace3e78 100644
--- a/cmd/main.go
+++ b/cmd/main.go
@@ -40,6 +40,7 @@ var (
 	cloudConfigFilePath = flag.String("cloud-config", "", "Path to GCE cloud provider config")
 	httpEndpoint        = flag.String("http-endpoint", "", "The TCP network address where the prometheus metrics endpoint will listen (example: `:8080`). The default is empty string, which means metrics endpoint is disabled.")
 	metricsPath         = flag.String("metrics-path", "/metrics", "The HTTP path where prometheus metrics will be exposed. Default is `/metrics`.")
+	enableMultishare    = flag.Bool("enable-multishare", false, "if set to true, the driver will support multishare instance provisioning")
 	// This is set at compile time
 	version = "unknown"
 )
@@ -56,6 +57,9 @@ func main() {
 	defer cancel()
 	var meta metadata.Service
 	if *runController {
+		if *enableMultishare {
+			klog.Fatalf("provisioning of multishare instances not supported")
+		}
 		if *httpEndpoint != "" && metrics.IsGKEComponentVersionAvailable() {
 			mm := metrics.NewMetricsManager()
 			mm.InitializeHttpHandler(*httpEndpoint, *metricsPath)
@@ -80,14 +84,15 @@ func main() {
 
 	mounter := mount.New("")
 	config := &driver.GCFSDriverConfig{
-		Name:            driverName,
-		Version:         version,
-		NodeID:          *nodeID,
-		RunController:   *runController,
-		RunNode:         *runNode,
-		Mounter:         mounter,
-		Cloud:           provider,
-		MetadataService: meta,
+		Name:             driverName,
+		Version:          version,
+		NodeID:           *nodeID,
+		RunController:    *runController,
+		RunNode:          *runNode,
+		Mounter:          mounter,
+		Cloud:            provider,
+		MetadataService:  meta,
+		EnableMultishare: *enableMultishare,
 	}
 
 	gcfsDriver, err := driver.NewGCFSDriver(config)
diff --git a/deploy/kubernetes/cluster_setup.sh b/deploy/kubernetes/cluster_setup.sh
index adb8e439..52a9951e 100755
--- a/deploy/kubernetes/cluster_setup.sh
+++ b/deploy/kubernetes/cluster_setup.sh
@@ -26,12 +26,6 @@ if [ "${DEPLOY_VERSION}" != dev ]; then
   if ! kubectl get secret gcp-filestore-csi-driver-sa --namespace=$GCFS_NS; then
     kubectl create secret generic gcp-filestore-csi-driver-sa --from-file="$GCFS_SA_FILE" --namespace=$GCFS_NS
   fi
-else
-  $mydir/webhook-example/create-cert.sh --namespace ${GCFS_NS} --service fs-validation
-  webhook_config="$mydir/overlays/${DEPLOY_VERSION}/mutation-configuration.yaml"
-  cat $mydir/mutation-webhook-configuration-template | $mydir/webhook-example/patch-ca-bundle.sh > $webhook_config
-  kubectl apply -f $webhook_config
-  rm $webhook_config
 fi
 
 readonly tmp_spec=/tmp/gcp-filestore-csi-driver-specs-generated.yaml
diff --git a/deploy/kubernetes/mutation-webhook-configuration-template b/deploy/kubernetes/mutation-webhook-configuration-template
deleted file mode 100644
index 071235bf..00000000
--- a/deploy/kubernetes/mutation-webhook-configuration-template
+++ /dev/null
@@ -1,24 +0,0 @@
-apiVersion: admissionregistration.k8s.io/v1
-kind: MutatingWebhookConfiguration
-metadata:
-  name: filestorecsi-mutation-webhook.storage.k8s.io
-webhooks:
-- name: filestorecsi-validation-webhook.storage.k8s.io
-  rules:
-  - apiGroups:   ["storage.k8s.io"]
-    apiVersions: ["v1"]
-    operations:  ["CREATE"]
-    resources:   ["storageclasses"]
-    scope:       "*"
-  clientConfig:
-    caBundle: ${CA_BUNDLE}
-    service:
-      namespace: gcp-filestore-csi-driver
-      name: "fs-validation"
-      path: "/storageclasses"
-      port: 443
-  admissionReviewVersions: ["v1"]
-  sideEffects: None
-  reinvocationPolicy: Never
-  failurePolicy: Ignore
-  timeoutSeconds: 2
\ No newline at end of file
diff --git a/deploy/kubernetes/overlays/dev/kustomization.yaml b/deploy/kubernetes/overlays/dev/kustomization.yaml
index 0f92f0b1..32fbf2f8 100644
--- a/deploy/kubernetes/overlays/dev/kustomization.yaml
+++ b/deploy/kubernetes/overlays/dev/kustomization.yaml
@@ -2,7 +2,6 @@ apiVersion: kustomize.config.k8s.io/v1beta1
 kind: Kustomization
 resources:
 - ../stable-master
-- webhook-deployment.yaml
 patchesStrategicMerge:
 - controller_always_pull.yaml
 - node_always_pull.yaml
@@ -10,10 +9,6 @@ patchesStrategicMerge:
 - noauth.yaml
 
 namespace: gcp-filestore-csi-driver
-
-images:
-- name: gcr.io/k8s-staging-cloud-provider-gcp/gcp-filestore-csi-driver-webhook
-  newTag: "canary"
 # To change the dev image, add something like the following.
 #images:
 #- name: gcr.io/gke-release/gcp-filestore-csi-driver
diff --git a/deploy/kubernetes/overlays/dev/webhook-deployment.yaml b/deploy/kubernetes/overlays/dev/webhook-deployment.yaml
deleted file mode 100644
index 8584efdb..00000000
--- a/deploy/kubernetes/overlays/dev/webhook-deployment.yaml
+++ /dev/null
@@ -1,42 +0,0 @@
-apiVersion: apps/v1
-kind: Deployment
-metadata:
-  name: filestorecsi-validation-webhook
-spec:
-  replicas: 1
-  selector:
-    matchLabels:
-      app: filestorecsi-validation
-  template:
-    metadata:
-      labels:
-        app: filestorecsi-validation
-    spec:
-      containers:
-      - name: filestorecsi-validation
-        image: gcr.io/k8s-staging-cloud-provider-gcp/gcp-filestore-csi-driver-webhook
-        imagePullPolicy: Always
-        args: ['--tls-cert-file=/etc/filestorecsi-validation-webhook/certs/cert.pem', '--tls-private-key-file=/etc/filestorecsi-validation-webhook/certs/key.pem']
-        ports:
-        - containerPort: 443 # change the port as needed
-        volumeMounts:
-          - name: filestorecsi-validation-webhook-certs
-            mountPath: /etc/filestorecsi-validation-webhook/certs
-            readOnly: true
-      volumes:
-        - name: filestorecsi-validation-webhook-certs
-          secret:
-            secretName: filestorecsi-validation-secret
----
-apiVersion: v1
-kind: Service
-metadata:
-  name: fs-validation
-  namespace: default
-spec:
-  selector:
-    app: filestorecsi-validation
-  ports:
-    - protocol: TCP
-      port: 443 # Change if needed
-      targetPort: 443 # Change if the webserver image expects a different port
diff --git a/pkg/cloud_provider/fake.go b/pkg/cloud_provider/fake.go
index d6934b60..71bfe7fa 100644
--- a/pkg/cloud_provider/fake.go
+++ b/pkg/cloud_provider/fake.go
@@ -34,3 +34,11 @@ func NewFakeCloud() (*Cloud, error) {
 		Zone:    "us-central1-c",
 	}, nil
 }
+
+func NewFakeCloudWithFiler(filer file.Service, project, location string) (*Cloud, error) {
+	return &Cloud{
+		File:    filer,
+		Project: project,
+		Zone:    location,
+	}, nil
+}
diff --git a/pkg/cloud_provider/file/fake.go b/pkg/cloud_provider/file/fake.go
index d511b251..14f348f2 100644
--- a/pkg/cloud_provider/file/fake.go
+++ b/pkg/cloud_provider/file/fake.go
@@ -18,22 +18,29 @@ package file
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 
+	"github.com/google/uuid"
 	filev1beta1 "google.golang.org/api/file/v1beta1"
 	"google.golang.org/api/googleapi"
+	"google.golang.org/genproto/googleapis/rpc/code"
+	"google.golang.org/grpc/codes"
 )
 
 const (
 	defaultProject    = "test-project"
 	defaultZone       = "us-central1-c"
+	defaultRegion     = "us-central1"
 	defaultTier       = "BASIC_HDD"
 	defaultCapacityGb = 1024
 )
 
 type fakeServiceManager struct {
-	createdInstances map[string]*ServiceInstance
-	backups          map[string]*BackupInfo
+	createdInstances          map[string]*ServiceInstance
+	backups                   map[string]*BackupInfo
+	createdMultishareInstance map[string]*MultishareInstance
+	createdMultishares        map[string]*Share
 }
 
 var _ Service = &fakeServiceManager{}
@@ -206,7 +213,8 @@ func notFoundError() *googleapi.Error {
 type fakeBlockingServiceManager struct {
 	*fakeServiceManager
 	// 'OperationUnblocker' channel is used to block the execution of the respective function using it. This is done by sending a channel of empty struct over 'OperationUnblocker' channel, and wait until the tester gives a go-ahead to proceed further in the execution of the function.
-	OperationUnblocker chan chan struct{}
+	OperationUnblocker  chan chan struct{}
+	MultishareUnblocker chan chan Signal
 }
 
 func NewFakeBlockingService(operationUnblocker chan chan struct{}) (Service, error) {
@@ -238,8 +246,12 @@ func (m *fakeBlockingServiceManager) HasOperations(ctx context.Context, obj *Ser
 }
 
 // Multishare fake functions defined here
-func (manager *fakeServiceManager) GetMultishareInstance(ctx context.Context, uri string) (*MultishareInstance, error) {
-	return nil, nil
+func (manager *fakeServiceManager) GetMultishareInstance(ctx context.Context, obj *MultishareInstance) (*MultishareInstance, error) {
+	instance, ok := manager.createdMultishareInstance[obj.Name]
+	if !ok {
+		return nil, fmt.Errorf("notFound")
+	}
+	return instance, nil
 }
 
 func (manager *fakeServiceManager) ListMultishareInstances(ctx context.Context, filter *ListFilter) ([]*MultishareInstance, error) {
@@ -247,7 +259,39 @@ func (manager *fakeServiceManager) ListMultishareInstances(ctx context.Context,
 }
 
 func (manager *fakeServiceManager) StartCreateMultishareInstanceOp(ctx context.Context, obj *MultishareInstance) (*filev1beta1.Operation, error) {
-	return nil, nil
+	instance := &MultishareInstance{
+		Project:       defaultProject,
+		Location:      defaultRegion,
+		Name:          obj.Name,
+		Tier:          obj.Tier,
+		CapacityBytes: obj.CapacityBytes,
+		Network: Network{
+			Name:            obj.Network.Name,
+			Ip:              "1.1.1.1",
+			ReservedIpRange: obj.Network.ReservedIpRange,
+		},
+		Labels: obj.Labels,
+		State:  "READY",
+	}
+	manager.createdMultishareInstance[instance.Name] = instance
+	manager.createdMultishareInstance[obj.Name] = instance
+	meta := &filev1beta1.OperationMetadata{
+		Target: fmt.Sprintf(instanceURIFmt, instance.Project, instance.Location, instance.Name),
+		Verb:   "CREATE",
+	}
+	metaBytes, _ := json.Marshal(meta)
+	op := &filev1beta1.Operation{
+		Name:     "operation-" + uuid.New().String(),
+		Metadata: metaBytes,
+	}
+	return op, nil
+}
+
+type Signal struct {
+	ReportError             bool
+	ReportNotFoundError     bool
+	ReportRunning           bool
+	ReportOpWithErrorStatus bool
 }
 
 func (manager *fakeServiceManager) StartDeleteMultishareInstanceOp(ctx context.Context, obj *MultishareInstance) (*filev1beta1.Operation, error) {
@@ -259,17 +303,160 @@ func (manager *fakeServiceManager) StartResizeMultishareInstanceOp(ctx context.C
 }
 
 func (manager *fakeServiceManager) StartCreateShareOp(ctx context.Context, obj *Share) (*filev1beta1.Operation, error) {
-	return nil, nil
+	if _, ok := manager.createdMultishareInstance[obj.Parent.Name]; !ok {
+		return nil, fmt.Errorf("host instance %s not found", obj.Parent.Name)
+	}
+
+	parent := &MultishareInstance{
+		Project:       obj.Parent.Project,
+		Location:      obj.Parent.Location,
+		Name:          obj.Parent.Name,
+		Tier:          obj.Parent.Tier,
+		CapacityBytes: obj.CapacityBytes,
+		Network: Network{
+			Name:            obj.Parent.Network.Name,
+			Ip:              "1.1.1.1",
+			ReservedIpRange: obj.Parent.Network.ReservedIpRange,
+		},
+		Labels: obj.Parent.Labels,
+		State:  "READY",
+	}
+	share := &Share{
+		Name:          obj.Name,
+		Parent:        parent,
+		CapacityBytes: obj.CapacityBytes,
+		Labels:        obj.Labels,
+	}
+	manager.createdMultishares[share.Name] = share
+
+	meta := &filev1beta1.OperationMetadata{
+		Target: fmt.Sprintf(shareURIFmt, share.Parent.Project, share.Parent.Location, share.Parent.Name, share.Name),
+		Verb:   "CREATE",
+	}
+	metaBytes, _ := json.Marshal(meta)
+	op := &filev1beta1.Operation{
+		Name:     "operation-" + uuid.New().String(),
+		Metadata: metaBytes,
+	}
+
+	return op, nil
 }
 
 func (manager *fakeServiceManager) StartDeleteShareOp(ctx context.Context, obj *Share) (*filev1beta1.Operation, error) {
-	return nil, nil
+	delete(manager.createdMultishares, obj.Name)
+
+	meta := &filev1beta1.OperationMetadata{
+		Target: fmt.Sprintf(shareURIFmt, obj.Parent.Project, obj.Parent.Location, obj.Parent.Name, obj.Name),
+		Verb:   "DELETE",
+	}
+	metaBytes, _ := json.Marshal(meta)
+	op := &filev1beta1.Operation{
+		Name:     "operation-" + uuid.New().String(),
+		Metadata: metaBytes,
+	}
+
+	return op, nil
 }
 
 func (manager *fakeServiceManager) StartResizeShareOp(ctx context.Context, obj *Share) (*filev1beta1.Operation, error) {
 	return nil, nil
 }
 
-func (manager *fakeServiceManager) WaitForOpWithOpts(ctx context.Context, opts PollOpts) error {
+func (manager *fakeServiceManager) WaitForOpWithOpts(ctx context.Context, op string, opts PollOpts) error {
 	return nil
 }
+
+func (manager *fakeServiceManager) GetOp(ctx context.Context, opName string) (*filev1beta1.Operation, error) {
+	op := &filev1beta1.Operation{
+		Name: opName,
+		Done: true,
+	}
+	return op, nil
+}
+
+func (manager *fakeServiceManager) IsOpDone(*filev1beta1.Operation) (bool, error) {
+	return true, nil
+}
+
+func (manager *fakeServiceManager) GetShare(ctx context.Context, obj *Share) (*Share, error) {
+	share, ok := manager.createdMultishares[obj.Name]
+	if !ok {
+		return nil, notFoundError()
+	}
+	return share, nil
+}
+
+func (manager *fakeServiceManager) ListShares(ctx context.Context, filter *ListFilter) ([]*Share, error) {
+	var s []*Share
+	for _, v := range manager.createdMultishares {
+		if v.Parent.Project == filter.Project && v.Parent.Location == filter.Location && v.Parent.Name == filter.InstanceName {
+			s = append(s, v)
+		}
+	}
+
+	return s, nil
+}
+
+func NewFakeBlockingServiceForMultishare(unblocker chan chan Signal) (Service, error) {
+	return &fakeBlockingServiceManager{
+		fakeServiceManager: &fakeServiceManager{
+			createdMultishareInstance: make(map[string]*MultishareInstance),
+			createdMultishares:        make(map[string]*Share),
+		},
+		MultishareUnblocker: unblocker,
+	}, nil
+}
+
+func (manager *fakeBlockingServiceManager) GetMultishareInstance(ctx context.Context, instance *MultishareInstance) (*MultishareInstance, error) {
+	execute := make(chan Signal)
+	manager.MultishareUnblocker <- execute
+	val := <-execute
+	if val.ReportError {
+		return nil, fmt.Errorf("mock error")
+	}
+	if val.ReportNotFoundError {
+		return nil, &googleapi.Error{
+			Code: int(code.Code_NOT_FOUND),
+			Errors: []googleapi.ErrorItem{
+				{
+					Reason: "notFound",
+				},
+			},
+		}
+	}
+	return manager.fakeServiceManager.GetMultishareInstance(ctx, instance)
+}
+
+func (manager *fakeBlockingServiceManager) GetOp(ctx context.Context, opName string) (*filev1beta1.Operation, error) {
+	execute := make(chan Signal)
+	manager.MultishareUnblocker <- execute
+	val := <-execute
+	if val.ReportError {
+		return nil, fmt.Errorf("mock error")
+	}
+
+	op := &filev1beta1.Operation{
+		Name: opName,
+		Done: true,
+	}
+	if val.ReportOpWithErrorStatus {
+		op.Error = &filev1beta1.Status{Code: int64(codes.Internal)}
+		return op, nil
+	}
+
+	if val.ReportRunning {
+		op.Done = false
+	}
+	return op, nil
+}
+
+func (manager *fakeBlockingServiceManager) IsOpDone(*filev1beta1.Operation) (bool, error) {
+	execute := make(chan Signal)
+	manager.MultishareUnblocker <- execute
+	val := <-execute
+	if val.ReportError {
+		return !val.ReportRunning, fmt.Errorf("mock error")
+	}
+
+	return !val.ReportRunning, nil
+}
diff --git a/pkg/cloud_provider/file/file.go b/pkg/cloud_provider/file/file.go
index 0b1df141..d93ebcf3 100644
--- a/pkg/cloud_provider/file/file.go
+++ b/pkg/cloud_provider/file/file.go
@@ -42,25 +42,32 @@ type PollOpts struct {
 }
 
 type Share struct {
-	Name      string
-	SizeBytes int64
+	Name           string              // only the share name
+	Parent         *MultishareInstance // parent captures the project, location details.
+	State          string
+	MountPointName string
+	Labels         map[string]string
+	CapacityBytes  int64
 }
 
 type MultishareInstance struct {
-	Project  string
-	Name     string
-	Location string
-	Tier     string
-	Network  Network
-	Shares   []Share
-	Labels   map[string]string
-	State    string
+	Project            string
+	Name               string
+	Location           string
+	Tier               string
+	Network            Network
+	CapacityBytes      int64
+	MaxCapacityBytes   int64
+	CapacityStepSizeGb int64
+	Labels             map[string]string
+	State              string
+	KmsKeyName         string
 }
 
 type ListFilter struct {
-	Project  string
-	Location string
-	Instance string
+	Project      string
+	Location     string
+	InstanceName string
 }
 
 type ServiceInstance struct {
@@ -104,15 +111,19 @@ type Service interface {
 	CreateInstanceFromBackupSource(ctx context.Context, obj *ServiceInstance, volumeSourceSnapshotId string) (*ServiceInstance, error)
 	HasOperations(ctx context.Context, obj *ServiceInstance, operationType string, done bool) (bool, error)
 	// Multishare ops
-	GetMultishareInstance(ctx context.Context, uri string) (*MultishareInstance, error)
+	GetMultishareInstance(ctx context.Context, obj *MultishareInstance) (*MultishareInstance, error)
 	ListMultishareInstances(ctx context.Context, filter *ListFilter) ([]*MultishareInstance, error)
 	StartCreateMultishareInstanceOp(ctx context.Context, obj *MultishareInstance) (*filev1beta1.Operation, error)
 	StartDeleteMultishareInstanceOp(ctx context.Context, obj *MultishareInstance) (*filev1beta1.Operation, error)
 	StartResizeMultishareInstanceOp(ctx context.Context, obj *MultishareInstance) (*filev1beta1.Operation, error)
+	ListShares(ctx context.Context, filter *ListFilter) ([]*Share, error)
+	GetShare(ctx context.Context, obj *Share) (*Share, error)
 	StartCreateShareOp(ctx context.Context, obj *Share) (*filev1beta1.Operation, error)
 	StartDeleteShareOp(ctx context.Context, obj *Share) (*filev1beta1.Operation, error)
 	StartResizeShareOp(ctx context.Context, obj *Share) (*filev1beta1.Operation, error)
-	WaitForOpWithOpts(ctx context.Context, opts PollOpts) error
+	WaitForOpWithOpts(ctx context.Context, op string, opts PollOpts) error
+	GetOp(ctx context.Context, op string) (*filev1beta1.Operation, error)
+	IsOpDone(op *filev1beta1.Operation) (bool, error)
 }
 
 type gcfsServiceManager struct {
@@ -491,6 +502,7 @@ func (manager *gcfsServiceManager) waitForOp(ctx context.Context, op *filev1beta
 	})
 }
 
+// TODO: unify this function behavior with IsOpDone
 func isOpDone(op *filev1beta1.Operation) (bool, error) {
 	if op == nil {
 		return false, nil
@@ -501,6 +513,17 @@ func isOpDone(op *filev1beta1.Operation) (bool, error) {
 	return op.Done, nil
 }
 
+func (manager *gcfsServiceManager) IsOpDone(op *filev1beta1.Operation) (bool, error) {
+	// TODO: verify this behavior with filestore
+	if op == nil {
+		return true, nil
+	}
+	if op.Error != nil {
+		return true, fmt.Errorf("operation %v failed (%v): %v", op.Name, op.Error.Code, op.Error.Message)
+	}
+	return op.Done, nil
+}
+
 func locationURI(project, location string) string {
 	return fmt.Sprintf(locationURIFmt, project, location)
 }
@@ -543,20 +566,17 @@ func IsNotFoundErr(err error) bool {
 }
 
 // This function returns true if the error is a googleapi error caused by users, such as
-// Error 429: Quota limit exceeded, Error 403: Permission Denied, Error 400: Bad Request,
-// Error 404: Not found.
+// Error 429: Quota limit exceeded, Error 403: Permission Denied, and Error 400: Bad Request
 func IsUserError(err error) bool {
 	apiErr, ok := err.(*googleapi.Error)
 	if !ok {
 		return false
 	}
-	userErrors := map[code.Code]bool{
-		code.Code_RESOURCE_EXHAUSTED: true,
-		code.Code_PERMISSION_DENIED:  true,
-		code.Code_INVALID_ARGUMENT:   true,
-		code.Code_NOT_FOUND:          true,
+
+	if apiErr.Code == int(code.Code_RESOURCE_EXHAUSTED) || apiErr.Code == int(code.Code_PERMISSION_DENIED) || apiErr.Code == int(code.Code_INVALID_ARGUMENT) {
+		return true
 	}
-	return userErrors[code.Code(apiErr.Code)]
+	return false
 }
 
 // This function returns the backup URI, the region that was picked to be the backup resource location and error.
@@ -617,38 +637,140 @@ func ApplyFilter(ops []*filev1beta1.Operation, uri string, opType string, done b
 }
 
 // Multishare functions defined here
-func (manager *gcfsServiceManager) GetMultishareInstance(ctx context.Context, uri string) (*MultishareInstance, error) {
+func (manager *gcfsServiceManager) GetMultishareInstance(ctx context.Context, obj *MultishareInstance) (*MultishareInstance, error) {
+	// TODO: do not silence not found error.
 	return nil, nil
 }
 
 func (manager *gcfsServiceManager) ListMultishareInstances(ctx context.Context, filter *ListFilter) ([]*MultishareInstance, error) {
+	// TODO
 	return nil, nil
 }
 
 func (manager *gcfsServiceManager) StartCreateMultishareInstanceOp(ctx context.Context, obj *MultishareInstance) (*filev1beta1.Operation, error) {
+	// TODO
 	return nil, nil
 }
 
 func (manager *gcfsServiceManager) StartDeleteMultishareInstanceOp(ctx context.Context, obj *MultishareInstance) (*filev1beta1.Operation, error) {
+	// TODO
 	return nil, nil
 }
 
 func (manager *gcfsServiceManager) StartResizeMultishareInstanceOp(ctx context.Context, obj *MultishareInstance) (*filev1beta1.Operation, error) {
+	// TODO
 	return nil, nil
 }
 
 func (manager *gcfsServiceManager) StartCreateShareOp(ctx context.Context, obj *Share) (*filev1beta1.Operation, error) {
+	// TODO
 	return nil, nil
 }
 
 func (manager *gcfsServiceManager) StartDeleteShareOp(ctx context.Context, obj *Share) (*filev1beta1.Operation, error) {
+	// TODO
 	return nil, nil
 }
 
 func (manager *gcfsServiceManager) StartResizeShareOp(ctx context.Context, obj *Share) (*filev1beta1.Operation, error) {
+	// TODO
+	return nil, nil
+}
+
+func (manager *gcfsServiceManager) WaitForOpWithOpts(ctx context.Context, op string, opts PollOpts) error {
+	// TODO
+	return nil
+}
+
+func (manager *gcfsServiceManager) GetOp(ctx context.Context, op string) (*filev1beta1.Operation, error) {
+	opInfo, err := manager.operationsService.Get(op).Context(ctx).Do()
+	if err != nil {
+		return nil, err
+	}
+	return opInfo, nil
+}
+
+func (manager *gcfsServiceManager) GetShare(ctx context.Context, obj *Share) (*Share, error) {
+	return &Share{}, nil
+}
+
+func (manager *gcfsServiceManager) ListShares(ctx context.Context, filter *ListFilter) ([]*Share, error) {
 	return nil, nil
 }
 
-func (manager *gcfsServiceManager) WaitForOpWithOpts(ctx context.Context, opts PollOpts) error {
+func ParseShare(s *Share) (string, string, string, string, error) {
+	if s == nil || s.Parent == nil {
+		return "", "", "", "", fmt.Errorf("Missing parent object for share %s", s.Name)
+	}
+
+	if s.Parent.Project == "" || s.Parent.Location == "" || s.Parent.Name == "" || s.Name == "" {
+		return "", "", "", "", fmt.Errorf("Missing identifier in share")
+	}
+
+	return s.Parent.Project, s.Parent.Location, s.Parent.Name, s.Name, nil
+}
+
+func GetMultishareInstanceHandle(m *MultishareInstance) (string, error) {
+	if m == nil {
+		return "", fmt.Errorf("empty multishare instance")
+	}
+	return fmt.Sprintf("%s/%s/%s", m.Project, m.Location, m.Name), nil
+}
+
+func CompareMultishareInstances(a, b *MultishareInstance) error {
+	if a == nil || b == nil {
+		return fmt.Errorf("empty instance object detected")
+	}
+
+	mismatches := []string{}
+	if a.Project != b.Project {
+		mismatches = append(mismatches, "project")
+	}
+
+	if a.Location != b.Location {
+		mismatches = append(mismatches, "location")
+	}
+
+	if a.Name != b.Name {
+		mismatches = append(mismatches, "name")
+	}
+
+	if strings.ToLower(a.Tier) != strings.ToLower(b.Tier) {
+		mismatches = append(mismatches, "tier")
+	}
+
+	if a.Network.Name != b.Network.Name {
+		mismatches = append(mismatches, "network name")
+	}
+
+	// Filestore API does not include key version info in the Instance object, simple string comparison will work
+	if a.KmsKeyName != b.KmsKeyName {
+		mismatches = append(mismatches, "kms key name")
+	}
+
+	if len(mismatches) > 0 {
+		return fmt.Errorf("instance %v already exists but doesn't match expected: %+v", a.Name, mismatches)
+	}
+	return nil
+}
+
+func CompareShares(a, b *Share) error {
+	if a == nil || b == nil {
+		return fmt.Errorf("empty share object detected")
+	}
+
+	mismatches := []string{}
+	if a.Name != b.Name {
+		mismatches = append(mismatches, "tier")
+	}
+	if util.RoundBytesToGb(a.CapacityBytes) != util.RoundBytesToGb(b.CapacityBytes) {
+		mismatches = append(mismatches, "share size")
+	}
+	if len(mismatches) > 0 {
+		return fmt.Errorf("share %v already exists but doesn't match expected: %+v", a.Name, mismatches)
+	}
+	if err := CompareMultishareInstances(a.Parent, b.Parent); err != nil {
+		return err
+	}
 	return nil
 }
diff --git a/pkg/csi_driver/controller.go b/pkg/csi_driver/controller.go
index 5413d077..81977847 100644
--- a/pkg/csi_driver/controller.go
+++ b/pkg/csi_driver/controller.go
@@ -55,13 +55,14 @@ const (
 
 // CreateVolume parameters
 const (
-	paramTier                     = "tier"
-	paramLocation                 = "location"
-	paramNetwork                  = "network"
-	paramReservedIPV4CIDR         = "reserved-ipv4-cidr"
-	paramReservedIPRange          = "reserved-ip-range"
-	paramConnectMode              = "connect-mode"
-	paramInstanceEncryptionKmsKey = "instance-encryption-kms-key"
+	paramTier                      = "tier"
+	paramLocation                  = "location"
+	paramNetwork                   = "network"
+	paramReservedIPV4CIDR          = "reserved-ipv4-cidr"
+	paramReservedIPRange           = "reserved-ip-range"
+	paramConnectMode               = "connect-mode"
+	paramInstanceEncryptionKmsKey  = "instance-encryption-kms-key"
+	paramMultishareInstanceScLabel = "instance-storageclass-label"
 
 	// Keys for PV and PVC parameters as reported by external-provisioner
 	ParameterKeyPVCName      = "csi.storage.k8s.io/pvc/name"
@@ -84,15 +85,20 @@ type controllerServer struct {
 }
 
 type controllerServerConfig struct {
-	driver      *GCFSDriver
-	fileService file.Service
-	cloud       *cloud.Cloud
-	ipAllocator *util.IPAllocator
-	volumeLocks *util.VolumeLocks
+	driver               *GCFSDriver
+	fileService          file.Service
+	cloud                *cloud.Cloud
+	ipAllocator          *util.IPAllocator
+	volumeLocks          *util.VolumeLocks
+	enableMultishare     bool
+	multiShareController *MultishareController
 }
 
 func newControllerServer(config *controllerServerConfig) csi.ControllerServer {
 	config.ipAllocator = util.NewIPAllocator(make(map[string]bool))
+	if config.enableMultishare {
+		config.multiShareController = NewMultishareController(config.driver, config.fileService, config.cloud, config.volumeLocks)
+	}
 	return &controllerServer{config: config}
 }
 
@@ -710,11 +716,7 @@ func (s *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateSn
 	backupObj, err := s.config.fileService.CreateBackup(ctx, filer, req.Name, util.GetBackupLocation(req.GetParameters()))
 	if err != nil {
 		glog.Errorf("Create snapshot for volume Id %s failed: %v", volumeID, err)
-		if file.IsUserError(err) {
-			return nil, status.Error(codes.InvalidArgument, err.Error())
-		} else {
-			return nil, status.Error(codes.Internal, err.Error())
-		}
+		return nil, status.Error(codes.Internal, err.Error())
 	}
 	tp, err := util.ParseTimestamp(backupObj.CreateTime)
 	if err != nil {
diff --git a/pkg/csi_driver/gcfs_driver.go b/pkg/csi_driver/gcfs_driver.go
index ca76fc36..6cd7b817 100644
--- a/pkg/csi_driver/gcfs_driver.go
+++ b/pkg/csi_driver/gcfs_driver.go
@@ -30,14 +30,15 @@ import (
 )
 
 type GCFSDriverConfig struct {
-	Name            string          // Driver name
-	Version         string          // Driver version
-	NodeID          string          // Node name
-	RunController   bool            // Run CSI controller service
-	RunNode         bool            // Run CSI node service
-	Mounter         mount.Interface // Mount library
-	Cloud           *cloud.Cloud    // Cloud provider
-	MetadataService metadataservice.Service
+	Name             string          // Driver name
+	Version          string          // Driver version
+	NodeID           string          // Node name
+	RunController    bool            // Run CSI controller service
+	RunNode          bool            // Run CSI node service
+	Mounter          mount.Interface // Mount library
+	Cloud            *cloud.Cloud    // Cloud provider
+	MetadataService  metadataservice.Service
+	EnableMultishare bool
 }
 
 type GCFSDriver struct {
@@ -98,10 +99,11 @@ func NewGCFSDriver(config *GCFSDriverConfig) (*GCFSDriver, error) {
 
 		// Configure controller server
 		driver.cs = newControllerServer(&controllerServerConfig{
-			driver:      driver,
-			fileService: config.Cloud.File,
-			cloud:       config.Cloud,
-			volumeLocks: util.NewVolumeLocks(),
+			driver:           driver,
+			fileService:      config.Cloud.File,
+			cloud:            config.Cloud,
+			volumeLocks:      util.NewVolumeLocks(),
+			enableMultishare: config.EnableMultishare,
 		})
 	}
 
diff --git a/pkg/csi_driver/multishare_controller.go b/pkg/csi_driver/multishare_controller.go
index 363ba969..f4dbe8c9 100644
--- a/pkg/csi_driver/multishare_controller.go
+++ b/pkg/csi_driver/multishare_controller.go
@@ -17,10 +17,21 @@ limitations under the License.
 package driver
 
 import (
+	"fmt"
+	"strings"
+
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"golang.org/x/net/context"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"k8s.io/apimachinery/pkg/util/uuid"
 	cloud "sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider"
 	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider/file"
+	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/util"
+)
+
+const (
+	modeMultishare = "modeMultishare"
 )
 
 // MultishareController handles CSI calls for volumes which use Filestore multishare instances.
@@ -29,14 +40,16 @@ type MultishareController struct {
 	fileService file.Service
 	cloud       *cloud.Cloud
 	opsManager  *MultishareOpsManager
+	volumeLocks *util.VolumeLocks
 }
 
-func NewMultishareController(driver *GCFSDriver, fileService file.Service, cloud *cloud.Cloud) *MultishareController {
+func NewMultishareController(driver *GCFSDriver, fileService file.Service, cloud *cloud.Cloud, volumeLock *util.VolumeLocks) *MultishareController {
 	return &MultishareController{
-		opsManager:  NewMultishareOpsManager(fileService, cloud),
+		opsManager:  NewMultishareOpsManager(cloud),
 		driver:      driver,
 		fileService: fileService,
 		cloud:       cloud,
+		volumeLocks: volumeLock,
 	}
 }
 
@@ -45,11 +58,90 @@ func (m *MultishareController) Run() {
 }
 
 func (m *MultishareController) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
-	// Handle higher level csi params validation, try locks
-	// Prepare instacne and initiate instance workflow by calling Multishare OpsManager functions
-	// Initiate share workflow by calling Multishare OpsManager functions
-	// Prepare and return csi response
-	return nil, nil
+	name := req.GetName()
+	if len(name) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "CreateVolume name must be provided")
+	}
+	if err := m.driver.validateVolumeCapabilities(req.GetVolumeCapabilities()); err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	instanceScPrefix, err := getInstanceSCPrefix(req)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	if acquired := m.volumeLocks.TryAcquire(name); !acquired {
+		return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, name)
+	}
+	defer m.volumeLocks.Release(name)
+
+	// If no eligible instance found, the ops manager may decide to create a new instance. Prepare a multishare instacne object for such a scenario.
+	instance, err := m.generateNewMultishareInstance(util.NewMultishareInstancePrefix+string(uuid.NewUUID()), req)
+	if err != nil {
+		return nil, err
+	}
+
+	workflow, share, err := m.opsManager.setupEligibleInstanceAndStartWorkflow(ctx, req, instance)
+	if err != nil {
+		return nil, err
+	}
+
+	if share != nil {
+		return m.getShareAndGenerateCSIResponse(ctx, share)
+	}
+
+	timeout, pollInterval, err := util.GetMultishareOpsTimeoutConfig(workflow.opType)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, err.Error())
+	}
+	// cache lock released. poll for op.
+	err = m.cloud.File.WaitForOpWithOpts(ctx, workflow.opName, file.PollOpts{Timeout: timeout, Interval: pollInterval})
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "Create Volume failed, operation %q poll error: %v", workflow.opName, err)
+	}
+
+	if workflow.opType == util.ShareCreate {
+		return m.getShareAndGenerateCSIResponse(ctx, workflow.share)
+	}
+
+	var shareCreateWorkflow *Workflow
+	var newShare *file.Share
+	switch workflow.opType {
+	case util.InstanceCreate, util.InstanceExpand:
+		newShare, err = generateNewShare(util.ConvertVolToShareName(req.Name), workflow.instance, req)
+		if err != nil {
+			return nil, err
+		}
+		shareCreateWorkflow, err = m.opsManager.startShareCreateWorkflowSafe(ctx, instanceScPrefix, newShare)
+		if err != nil {
+			return nil, err
+		}
+	default:
+		return nil, status.Errorf(codes.Internal, "Create Volume failed, unknown workflow %v detected", workflow.opType)
+	}
+
+	// cache lock released. poll for share create op.
+	shareCreatetimeout, shareCreatePollInterval, err := util.GetMultishareOpsTimeoutConfig(shareCreateWorkflow.opType)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, err.Error())
+	}
+	err = m.cloud.File.WaitForOpWithOpts(ctx, shareCreateWorkflow.opName, file.PollOpts{Timeout: shareCreatetimeout, Interval: shareCreatePollInterval})
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "operation %q poll error: %v", workflow.opName, err)
+	}
+
+	return m.getShareAndGenerateCSIResponse(ctx, newShare)
+}
+
+func (m *MultishareController) getShareAndGenerateCSIResponse(ctx context.Context, s *file.Share) (*csi.CreateVolumeResponse, error) {
+	share, err := m.fileService.GetShare(ctx, s)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, err.Error())
+	}
+
+	// TODO: do we need any further validation here?
+	return generateCSICreateVolumeResponse(share)
 }
 
 func (m *MultishareController) DeleteVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
@@ -65,3 +157,227 @@ func (m *MultishareController) ControllerExpandVolume(ctx context.Context, req *
 	// Prepare and return csi response
 	return nil, nil
 }
+
+func getInstanceSCPrefix(req *csi.CreateVolumeRequest) (string, error) {
+	params := req.GetParameters()
+	v, ok := params[paramMultishareInstanceScLabel]
+	if ok {
+		return "", fmt.Errorf("Failed to find instance prefix key")
+	}
+
+	if v == "" {
+		return "", fmt.Errorf("instance prefix is empty")
+	}
+
+	return v, nil
+}
+
+func (m *MultishareController) generateNewMultishareInstance(instanceName string, req *csi.CreateVolumeRequest) (*file.MultishareInstance, error) {
+	region, err := m.pickRegion(req.GetAccessibilityRequirements())
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	tier := enterpriseTier
+	network := defaultNetwork
+	connectMode := directPeering
+	kmsKeyName := ""
+	for k, v := range req.GetParameters() {
+		switch strings.ToLower(k) {
+		case paramTier:
+			tier = v
+		case paramNetwork:
+			network = v
+		case paramConnectMode:
+			connectMode = v
+			if connectMode != directPeering && connectMode != privateServiceAccess {
+				return nil, status.Errorf(codes.InvalidArgument, fmt.Sprintf("connect mode can only be one of %q or %q", directPeering, privateServiceAccess))
+			}
+		case paramInstanceEncryptionKmsKey:
+			kmsKeyName = v
+		// Ignore the cidr flag as it is not passed to the cloud provider
+		// It will be used to get unreserved IP in the reserveIPV4Range function
+		// ignore IPRange flag as it will be handled at the same place as cidr
+		case paramReservedIPV4CIDR, paramReservedIPRange:
+			continue
+		case strings.ToLower(util.ParamMultishareInstanceScLabel):
+			continue
+		case ParameterKeyLabels, ParameterKeyPVCName, ParameterKeyPVCNamespace, ParameterKeyPVName:
+		case "csiprovisionersecretname", "csiprovisionersecretnamespace":
+		default:
+			return nil, status.Errorf(codes.InvalidArgument, fmt.Sprintf("invalid parameter %q", k))
+		}
+	}
+
+	if tier != enterpriseTier {
+		return nil, status.Errorf(codes.InvalidArgument, fmt.Sprintf("tier %q not supported for multishare volumes", tier))
+	}
+
+	labels, err := extractInstanceLabels(req.GetParameters(), m.driver.config.Name)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, err.Error())
+	}
+
+	return &file.MultishareInstance{
+		Project:       m.cloud.Project,
+		Name:          instanceName,
+		CapacityBytes: util.MinMultishareInstanceSizeBytes,
+		Location:      region,
+		Tier:          tier,
+		Network: file.Network{
+			Name:        network,
+			ConnectMode: connectMode,
+		},
+		KmsKeyName: kmsKeyName,
+		Labels:     labels,
+	}, nil
+}
+
+func generateNewShare(name string, parent *file.MultishareInstance, req *csi.CreateVolumeRequest) (*file.Share, error) {
+	if parent == nil {
+		return nil, status.Error(codes.Internal, "parent mulishare instance is empty")
+	}
+	targetSizeBytes, err := getShareRequestCapacity(req.CapacityRange)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+	return &file.Share{
+		Name:          name,
+		Parent:        parent,
+		CapacityBytes: targetSizeBytes,
+		Labels:        extractShareLabels(req.Parameters),
+	}, nil
+}
+
+func (m *MultishareController) pickRegion(top *csi.TopologyRequirement) (string, error) {
+	if top == nil {
+		region, err := util.GetRegionFromZone(m.cloud.Zone)
+		if err != nil {
+			return "", err
+		}
+
+		return region, nil
+	}
+
+	zone, err := pickZoneFromTopology(top)
+	if err != nil {
+		return "", err
+	}
+	region, err := util.GetRegionFromZone(zone)
+	if err != nil {
+		return "", err
+	}
+	return region, nil
+}
+
+func extractInstanceLabels(parameters map[string]string, driverName string) (map[string]string, error) {
+	instanceLabels := make(map[string]string)
+	userProvidedLabels := make(map[string]string)
+	for k, v := range parameters {
+		switch strings.ToLower(k) {
+		case ParameterKeyLabels:
+			var err error
+			userProvidedLabels, err = util.ConvertLabelsStringToMap(v)
+			if err != nil {
+				return nil, status.Error(codes.InvalidArgument, err.Error())
+			}
+		case strings.ToLower(util.ParamMultishareInstanceScLabel):
+			err := util.CheckLabelValueRegex(v)
+			if err != nil {
+				return nil, status.Error(codes.InvalidArgument, err.Error())
+			}
+			instanceLabels[util.ParamMultishareInstanceScLabelKey] = v
+		}
+	}
+
+	instanceLabels[tagKeyCreatedBy] = strings.ReplaceAll(driverName, ".", "_")
+	finalInstanceLabels, err := mergeLabels(userProvidedLabels, instanceLabels)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+	return finalInstanceLabels, nil
+}
+
+func extractShareLabels(parameters map[string]string) map[string]string {
+	shareLabels := make(map[string]string)
+	for k, v := range parameters {
+		switch strings.ToLower(k) {
+		case ParameterKeyPVCName:
+			shareLabels[tagKeyCreatedForClaimName] = v
+		case ParameterKeyPVCNamespace:
+			shareLabels[tagKeyCreatedForClaimNamespace] = v
+		case ParameterKeyPVName:
+			shareLabels[tagKeyCreatedForVolumeName] = v
+		}
+	}
+	return shareLabels
+}
+
+func getShareRequestCapacity(capRange *csi.CapacityRange) (int64, error) {
+	if capRange == nil {
+		return util.MinShareSizeBytes, nil
+	}
+
+	rCap := capRange.GetRequiredBytes()
+	rSet := rCap > 0
+	lCap := capRange.GetLimitBytes()
+	lSet := lCap > 0
+
+	if !lSet && !rSet {
+		return 0, status.Errorf(codes.InvalidArgument, "Neither Limit bytes or Required bytes set")
+	}
+
+	if lSet && rSet && lCap < rCap {
+		return 0, status.Errorf(codes.InvalidArgument, "Limit bytes %v is less than required bytes %v", lCap, rCap)
+	}
+
+	// Check bounds of limit and request.
+	if lSet {
+		if lCap < util.MinShareSizeBytes {
+			return 0, status.Errorf(codes.InvalidArgument, "Limit bytes %v is less than minimum share size bytes %v", lCap, util.MinShareSizeBytes)
+		}
+
+		if lCap > util.MaxShareSizeBytes {
+			return 0, status.Errorf(codes.InvalidArgument, "Limit bytes %v is greater than maximum share size bytes %v", lCap, util.MaxShareSizeBytes)
+		}
+	}
+
+	if rSet {
+		if rCap < util.MinShareSizeBytes {
+			return 0, status.Errorf(codes.InvalidArgument, "Request bytes %v is less than minimum share size bytes %v", lCap, util.MinShareSizeBytes)
+		}
+
+		if rCap > util.MaxShareSizeBytes {
+			return 0, status.Errorf(codes.InvalidArgument, "Request bytes %v is greater than maximum share size bytes %v", lCap, util.MaxShareSizeBytes)
+		}
+	}
+
+	if lSet {
+		return lCap, nil
+	}
+
+	return rCap, nil
+}
+
+func generateCSICreateVolumeResponse(s *file.Share) (*csi.CreateVolumeResponse, error) {
+	volId, err := generateMultishareVolumeIdFromShare(s)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, err.Error())
+	}
+
+	resp := &csi.CreateVolumeResponse{
+		Volume: &csi.Volume{
+			VolumeId:      volId,
+			CapacityBytes: s.CapacityBytes,
+			VolumeContext: map[string]string{
+				attrIP: s.Parent.Network.Ip,
+			},
+		},
+	}
+	return resp, nil
+}
+
+func containsInstancePrefix(shareHandle string, project, location, instanceName string) bool {
+	targetInstance := fmt.Sprintf("%s/%s/%s", project, location, instanceName)
+	return strings.Contains(shareHandle, targetInstance)
+}
diff --git a/pkg/csi_driver/multishare_ops_manager.go b/pkg/csi_driver/multishare_ops_manager.go
index 1e718bcc..1544948d 100644
--- a/pkg/csi_driver/multishare_ops_manager.go
+++ b/pkg/csi_driver/multishare_ops_manager.go
@@ -17,23 +17,39 @@ limitations under the License.
 package driver
 
 import (
+	"context"
+	"fmt"
+	"math/rand"
+	"sync"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"k8s.io/klog"
 	cloud "sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider"
 	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider/file"
 	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/util"
 )
 
+// A workflow is defined as a sequence of steps to safely (by checking the storage class cache) initiate instance or share operations.
+type Workflow struct {
+	instance *file.MultishareInstance
+	share    *file.Share
+	opType   util.OperationType
+	opName   string
+}
+
 // MultishareOpsManager manages storage class cache, manages the lifecycle of all instance and share operations.
 type MultishareOpsManager struct {
-	cache       *util.StorageClassInfoMap
-	fileService file.Service
-	cloud       *cloud.Cloud
+	sync.Mutex // Lock to perform thread safe operations on the cache.
+	cache      *util.StorageClassInfoCache
+	cloud      *cloud.Cloud
 }
 
-func NewMultishareOpsManager(fileService file.Service, cloud *cloud.Cloud) *MultishareOpsManager {
+func NewMultishareOpsManager(cloud *cloud.Cloud) *MultishareOpsManager {
 	return &MultishareOpsManager{
-		cache:       util.NewStorageClassInfoMap(),
-		fileService: fileService,
-		cloud:       cloud,
+		cache: util.NewStorageClassInfoCache(),
+		cloud: cloud,
 	}
 }
 
@@ -41,3 +57,523 @@ func (m *MultishareOpsManager) Run() {
 	// TODO: Start periodic cache hydration
 	// TODO: Start periodic instance inspection for delete and shrink
 }
+
+// setupEligibleInstanceAndStartWorkflow returns a workflow object (to indicate an instance or share level workflow is started), or a share object (if existing share already found), or error.
+func (m *MultishareOpsManager) setupEligibleInstanceAndStartWorkflow(ctx context.Context, req *csi.CreateVolumeRequest, instance *file.MultishareInstance) (*Workflow, *file.Share, error) {
+	m.Lock()
+	defer m.Unlock()
+
+	// Check ShareCreateMap if a share create is already in progress.
+	shareName := util.ConvertVolToShareName(req.Name)
+	instanceScPrefix, err := getInstanceSCPrefix(req)
+	if err != nil {
+		return nil, nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	// Check if there is already a share create op in progress.
+	lastShareCreateOpInfo, opStatus, err := m.checkAndUpdateShareCreateOp(ctx, instanceScPrefix, shareName)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	if lastShareCreateOpInfo != nil {
+		switch opStatus {
+		case util.StatusRunning:
+			return nil, nil, status.Errorf(codes.Aborted, "Share create operation %q in progress", lastShareCreateOpInfo.OpName)
+		case util.StatusFailed:
+			return nil, nil, status.Errorf(codes.Internal, "Share create operation %q failed", lastShareCreateOpInfo.OpName)
+		case util.StatusSuccess:
+			share, err := m.validateShareExists(ctx, lastShareCreateOpInfo.InstanceHandle, shareName)
+			if err != nil {
+				return nil, nil, err
+			}
+			return nil, share, nil
+		default:
+			return nil, nil, status.Errorf(codes.Internal, "unknown op status %d for op %v", opStatus, lastShareCreateOpInfo.OpName)
+		}
+	}
+
+	// Check if share already part of an existing instance.
+	share, err := m.checkInstanceListForShare(ctx, instanceScPrefix, shareName)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	if share != nil {
+		return nil, share, nil
+	}
+
+	// No share or running share create op fouund. Proceed to eligible instance check.
+	eligible, numIneligible, err := m.runEligibleInstanceCheck(ctx, instanceScPrefix)
+	if err != nil {
+		return nil, nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	if len(eligible) > 0 {
+		// pick a random eligible instance
+		index := rand.Intn(len(eligible))
+		share, err := generateNewShare(shareName, eligible[index], req)
+		if err != nil {
+			return nil, nil, status.Error(codes.Internal, err.Error())
+		}
+
+		needExpand, targetBytes, err := m.instanceNeedsExpand(ctx, share)
+		if err != nil {
+			return nil, nil, status.Error(codes.Internal, err.Error())
+		}
+
+		if needExpand {
+			eligible[index].CapacityBytes = targetBytes
+			w, err := m.startInstanceWorkflow(ctx, instanceScPrefix, &Workflow{instance: eligible[index], opType: util.InstanceExpand})
+			return w, nil, err
+		}
+
+		w, err := m.startShareCreateWorkflow(ctx, instanceScPrefix, share)
+		return w, nil, err
+	}
+
+	if numIneligible > 0 {
+		// some instances not ready yet. wait for more instances to be ready.
+		return nil, nil, status.Errorf(codes.Aborted, " %d non-ready instances detected. No ready instance found", numIneligible)
+	}
+
+	w, err := m.startInstanceWorkflow(ctx, instanceScPrefix, &Workflow{instance: instance, opType: util.InstanceCreate})
+	return w, nil, err
+}
+
+// checkAndUpdateShareCreateOp checks the share create op map, and evaluates the running status of the ops. If ops are detected as complete, they are removed as part of the check.
+func (m *MultishareOpsManager) checkAndUpdateShareCreateOp(ctx context.Context, instanceScPrefix, shareName string) (*util.ShareCreateOpInfo, util.OperationStatus, error) {
+	opInfo := m.cache.GetShareCreateOp(instanceScPrefix, shareName)
+	if opInfo == nil {
+		return nil, util.StatusUnknown, nil
+	}
+
+	op, err := m.cloud.File.GetOp(ctx, opInfo.OpName)
+	if err != nil && !file.IsNotFoundErr(err) {
+		return opInfo, util.StatusUnknown, status.Errorf(codes.Internal, "Failed to get operation %q", opInfo.OpName)
+	}
+
+	done, err := m.cloud.File.IsOpDone(op)
+	if err != nil {
+		// op completed with error.
+		// clear cache and return retry error to the caller.
+		m.cache.DeleteShareCreateOp(instanceScPrefix, shareName, opInfo.OpName)
+		return opInfo, util.StatusFailed, nil
+	}
+
+	if !done {
+		return opInfo, util.StatusRunning, nil
+	}
+
+	// Clear cache.
+	m.cache.DeleteShareCreateOp(instanceScPrefix, shareName, opInfo.OpName)
+	return opInfo, util.StatusSuccess, nil
+}
+
+// checkAndUpdateShareOp checks the share ops map, and evaluates the running status of the ops. If ops are detected as complete, they are removed as part of the check.
+func (m *MultishareOpsManager) checkAndUpdateShareOp(ctx context.Context, instanceScPrefix string, share *file.Share) (*util.OpInfo, util.OperationStatus, error) {
+	shareKey := util.CreateShareKey(share.Parent.Project, share.Parent.Location, share.Parent.Name, share.Name)
+	opInfo := m.cache.GetShareOp(instanceScPrefix, shareKey)
+	if opInfo == nil {
+		return nil, util.StatusUnknown, nil
+	}
+
+	op, err := m.cloud.File.GetOp(ctx, opInfo.Name)
+	if err != nil && !file.IsNotFoundErr(err) {
+		return opInfo, util.StatusUnknown, status.Errorf(codes.Internal, "Failed to get operation %q", opInfo.Name)
+	}
+
+	done, err := m.cloud.File.IsOpDone(op)
+	if err != nil {
+		// op completed with error.
+		// clear cache and return retry error to the caller.
+		m.cache.DeleteShareOp(instanceScPrefix, shareKey, opInfo.Name)
+		return opInfo, util.StatusFailed, nil
+	}
+
+	if !done {
+		return opInfo, util.StatusRunning, nil
+	}
+
+	// Clear cache.
+	m.cache.DeleteShareOp(instanceScPrefix, shareKey, opInfo.Name)
+	return opInfo, util.StatusSuccess, nil
+}
+
+func (m *MultishareOpsManager) startShareCreateWorkflowSafe(ctx context.Context, instanceSCPrefix string, share *file.Share) (*Workflow, error) {
+	m.Lock()
+	defer m.Unlock()
+	return m.startShareCreateWorkflow(ctx, instanceSCPrefix, share)
+}
+
+func (m *MultishareOpsManager) startShareCreateWorkflow(ctx context.Context, instanceSCPrefix string, share *file.Share) (*Workflow, error) {
+	return m.startShareWorkflow(ctx, instanceSCPrefix, &Workflow{
+		share:  share,
+		opType: util.ShareCreate,
+	})
+}
+
+func (m *MultishareOpsManager) startInstanceWorkflow(ctx context.Context, instanceSCPrefix string, w *Workflow) (*Workflow, error) {
+	// This function has 3 distinct steps:
+	// 1. verify no instance ops running for the given instance.
+	// 2. verify no running share ops for the shares hosted on the given instance.
+	// 3. Start the instance op.
+	if w.instance == nil {
+		return nil, status.Errorf(codes.Internal, "instance not found in workflow object")
+	}
+
+	instanceReady, err := m.verifyNoRunningInstanceOps(ctx, instanceSCPrefix, w.instance)
+	if err != nil {
+		return nil, status.Errorf(codes.Aborted, "Instance %q check error: %v", w.instance.Name, err)
+	}
+	if !instanceReady {
+		return nil, status.Errorf(codes.Aborted, "Instance %q not ready", w.instance.Name)
+	}
+
+	// Verify no ongoing share ops for the instance. Also clear finised ops from cache, while doing verification.
+	instanceReady, err = m.verifyNoRunningShareOpsForInstance(ctx, instanceSCPrefix, w.instance)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "Failed to check share ops for Instance %q, error: %v", w.instance.Name, err)
+	}
+	if !instanceReady {
+		return nil, status.Errorf(codes.Aborted, "Instance %q not ready", w.instance.Name)
+	}
+
+	switch w.opType {
+	case util.InstanceCreate:
+		op, err := m.cloud.File.StartCreateMultishareInstanceOp(ctx, w.instance)
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, err.Error())
+		}
+		w.opName = op.Name
+	case util.InstanceExpand:
+		op, err := m.cloud.File.StartResizeMultishareInstanceOp(ctx, w.instance)
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, err.Error())
+		}
+		w.opName = op.Name
+	case util.InstanceDelete:
+		op, err := m.cloud.File.StartDeleteMultishareInstanceOp(ctx, w.instance)
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, err.Error())
+		}
+		w.opName = op.Name
+	default:
+		return nil, status.Errorf(codes.Internal, "for instance workflow, unknown op type %v", w.opType)
+	}
+
+	m.cache.AddInstanceOp(instanceSCPrefix, util.CreateInstanceKey(w.instance.Project, w.instance.Location, w.instance.Name), util.OpInfo{Name: w.opName, Type: w.opType})
+	return w, nil
+}
+
+func (m *MultishareOpsManager) startShareWorkflow(ctx context.Context, instanceSCPrefix string, w *Workflow) (*Workflow, error) {
+	// This function has 3 distinct steps:
+	// 1. verify no instance ops running for the instance hosting the given share.
+	// 2. verify no running ops for the given share.
+	// 3. Start the share op.
+	if w.share == nil {
+		return nil, status.Errorf(codes.Internal, "share not found in workflow object")
+	}
+
+	if w.share.Parent == nil {
+		return nil, status.Errorf(codes.Internal, "share parent not found in workflow object")
+	}
+
+	// verify instance is ready.
+	instanceReady, err := m.verifyNoRunningInstanceOps(ctx, instanceSCPrefix, w.share.Parent)
+	if err != nil {
+		return nil, status.Errorf(codes.Aborted, "Instance %q check error: %v", w.share.Parent.Name, err)
+	}
+	if !instanceReady {
+		return nil, status.Errorf(codes.Aborted, "Instance %q not ready", w.share.Parent.Name)
+	}
+
+	// Verify share is ready.
+	shareReady, err := m.verifyNoRunningShareOp(ctx, instanceSCPrefix, w.share)
+	if err != nil {
+		return nil, status.Errorf(codes.Aborted, "Share %q check error: %v", w.share.Name, err)
+	}
+	if !shareReady {
+		return nil, status.Errorf(codes.Aborted, "Share %q not ready", w.share.Name)
+	}
+
+	switch w.opType {
+	case util.ShareCreate:
+		// validate no entry in cache for the share
+		item := m.cache.GetShareCreateOp(instanceSCPrefix, w.share.Name)
+		if item != nil {
+			return nil, status.Errorf(codes.Aborted, "Share %q not ready, found op %q", w.share.Name, item.OpName)
+		}
+
+		op, err := m.cloud.File.StartCreateShareOp(ctx, w.share)
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, err.Error())
+		}
+
+		// Start op and update cache
+		instanceHandle, err := file.GetMultishareInstanceHandle(w.share.Parent)
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, err.Error())
+		}
+		m.cache.AddShareCreateOp(instanceSCPrefix, w.share.Name, util.ShareCreateOpInfo{
+			InstanceHandle: instanceHandle,
+			OpName:         op.Name,
+		})
+		w.opName = op.Name
+	case util.ShareExpand:
+		// validate no entry in cache for the share
+		item := m.cache.GetShareOp(instanceSCPrefix, util.CreateShareKey(w.share.Parent.Project, w.share.Parent.Location, w.share.Parent.Name, w.share.Name))
+		if item != nil {
+			return nil, status.Errorf(codes.Aborted, "Share %q not ready, found op %q", w.share.Name, item.Name)
+		}
+
+		// Start op and update cache
+		op, err := m.cloud.File.StartResizeShareOp(ctx, w.share)
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, err.Error())
+		}
+		m.cache.AddShareOp(instanceSCPrefix, util.CreateShareKey(w.share.Parent.Project, w.share.Parent.Location, w.share.Parent.Name, w.share.Name), util.OpInfo{Name: op.Name, Type: w.opType})
+		w.opName = op.Name
+	case util.ShareDelete:
+		// validate no entry in cache for the share
+		item := m.cache.GetShareOp(instanceSCPrefix, util.CreateShareKey(w.share.Parent.Project, w.share.Parent.Location, w.share.Parent.Name, w.share.Name))
+		if item != nil {
+			return nil, status.Errorf(codes.Aborted, "Share %q not ready, found op %q", w.share.Name, item.Name)
+		}
+
+		// Start op and update cache
+		op, err := m.cloud.File.StartDeleteShareOp(ctx, w.share)
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, err.Error())
+		}
+		m.cache.AddShareOp(instanceSCPrefix, util.CreateShareKey(w.share.Parent.Project, w.share.Parent.Location, w.share.Parent.Name, w.share.Name), util.OpInfo{Name: op.Name, Type: w.opType})
+		w.opName = op.Name
+	default:
+		return nil, status.Errorf(codes.Internal, "for share workflow, unknown op type %v", w.opType)
+	}
+	return w, nil
+}
+
+func (m *MultishareOpsManager) verifyNoRunningInstanceOps(ctx context.Context, instanceSCPrefix string, instance *file.MultishareInstance) (bool, error) {
+	opInfo := m.cache.GetInstanceOp(instanceSCPrefix, util.CreateInstanceKey(instance.Project, instance.Location, instance.Name))
+	if opInfo == nil {
+		return true, nil
+	}
+	if opInfo.Name == "" {
+		return true, nil
+	}
+
+	op, err := m.cloud.File.GetOp(ctx, opInfo.Name)
+	if err != nil && !file.IsNotFoundErr(err) {
+		return false, err
+	}
+
+	// This method returns error if op completed with error. Instance is still considered ready since op completed.
+	done, _ := m.cloud.File.IsOpDone(op)
+	if done {
+		m.cache.DeleteInstanceOp(instanceSCPrefix, util.CreateInstanceKey(instance.Project, instance.Location, instance.Name), opInfo.Name)
+	}
+	return done, nil
+}
+
+func (m *MultishareOpsManager) verifyNoRunningShareOp(ctx context.Context, instanceScPrefix string, share *file.Share) (bool, error) {
+	shareCreateOpInfo, shareCreateOpStatus, err := m.checkAndUpdateShareCreateOp(ctx, instanceScPrefix, share.Name)
+	if err != nil {
+		return false, err
+	}
+	shareCreateopRunning := (shareCreateOpInfo != nil && shareCreateOpStatus == util.StatusRunning)
+
+	shareOpInfo, shareOpStatus, err := m.checkAndUpdateShareOp(ctx, instanceScPrefix, share)
+	if err != nil {
+		return false, err
+	}
+
+	shareOpRunning := (shareOpInfo != nil && shareOpStatus == util.StatusRunning)
+	return (!shareCreateopRunning && !shareOpRunning), nil
+}
+
+func (m *MultishareOpsManager) verifyNoRunningShareOpsForInstance(ctx context.Context, instanceScPrefix string, instance *file.MultishareInstance) (bool, error) {
+	targetInstanceHandle, err := file.GetMultishareInstanceHandle(instance)
+	if err != nil {
+		return false, err
+	}
+	for _, item := range m.cache.GetShareCreateMap(instanceScPrefix).Items() {
+		if item.OpInfo.InstanceHandle != targetInstanceHandle {
+			continue
+		}
+		shareCreateOpInfo, shareCreateOpStatus, err := m.checkAndUpdateShareCreateOp(ctx, instanceScPrefix, item.Key)
+		if err != nil {
+			return false, err
+		}
+		if shareCreateOpInfo != nil && shareCreateOpStatus == util.StatusRunning {
+			return false, nil
+		}
+	}
+
+	for _, item := range m.cache.GetShareOpsMap(instanceScPrefix).Items() {
+		shareHandle := item.Key
+		if !containsInstancePrefix(string(shareHandle), instance.Project, instance.Location, instance.Name) {
+			continue
+		}
+
+		_, _, _, shareName, err := util.ParseShareHandle(string(item.Key))
+		if err != nil {
+			return false, err
+		}
+
+		opInfo, status, err := m.checkAndUpdateShareOp(ctx, instanceScPrefix, &file.Share{Parent: instance, Name: shareName})
+		if err != nil {
+			return false, err
+		}
+		if opInfo != nil && status == util.StatusRunning {
+			return false, nil
+		}
+	}
+
+	return true, nil
+}
+
+func (m *MultishareOpsManager) validateShareExists(ctx context.Context, instanceHandle, shareName string) (*file.Share, error) {
+	project, location, instanceName, err := util.ParseInstanceHandle(instanceHandle)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to parse instance handle %v", instanceHandle)
+	}
+
+	return m.cloud.File.GetShare(ctx, &file.Share{
+		Name: shareName,
+		Parent: &file.MultishareInstance{
+			Project:  project,
+			Location: location,
+			Name:     instanceName,
+		},
+	})
+}
+
+func (m *MultishareOpsManager) checkInstanceListForShare(ctx context.Context, instanceScPrefix string, targetShareName string) (*file.Share, error) {
+	for _, item := range m.cache.GetInstanceMap(instanceScPrefix).Items() {
+		project, location, instanceName, err := util.ParseInstanceHandle(string(item.Key))
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, "failed to parse instance key %q, err: %v", item.Key, err)
+		}
+
+		shares, err := m.cloud.File.ListShares(ctx, &file.ListFilter{Project: project, Location: location, InstanceName: instanceName})
+		// TODO: verify a list on a non existent instance would return not found error. or we need to check GET instance before listing shares.
+		if err != nil && !file.IsNotFoundErr(err) {
+			return nil, status.Errorf(codes.Internal, "failed to list shares for Instance %q, err: %v", item.Key, err)
+		}
+
+		for _, share := range shares {
+			_, _, _, shareName, err := file.ParseShare(share)
+			if err != nil {
+				return nil, status.Errorf(codes.Internal, "failed to parse share URI, err: %v", err)
+			}
+
+			if shareName == targetShareName {
+				return share, nil
+			}
+		}
+	}
+
+	return nil, nil
+}
+
+// runEligibleInstanceCheck returns a list of ready and non-ready instances.
+func (m *MultishareOpsManager) runEligibleInstanceCheck(ctx context.Context, instanceScPrefix string) ([]*file.MultishareInstance, int, error) {
+	var readyInstanceKeys []util.InstanceKey
+	nonReadyInstanceCount := 0
+	items := m.cache.GetInstanceMap(instanceScPrefix).Items()
+	for _, item := range items {
+		instanceKey := item.Key
+		opInfo := item.OpInfo
+		project, location, instanceName, err := util.ParseInstanceHandle(string(instanceKey))
+		if err != nil {
+			klog.Warningf("Failed to parse instance key %v", instanceKey)
+			continue
+		}
+		if opInfo.Name == "" {
+			readyInstanceKeys = append(readyInstanceKeys, instanceKey)
+			continue
+		}
+
+		// check and clear completed op
+		ready, err := m.verifyNoRunningInstanceOps(ctx, instanceScPrefix, &file.MultishareInstance{
+			Project:  project,
+			Location: location,
+			Name:     instanceName,
+		})
+		if err != nil {
+			klog.Warningf("Failed to check instance ready for instance key %v, err:%v", instanceKey, err)
+		}
+
+		if !ready && opInfo.Type != util.InstanceDelete {
+			nonReadyInstanceCount = nonReadyInstanceCount + 1
+		}
+
+		if ready {
+			readyInstanceKeys = append(readyInstanceKeys, instanceKey)
+		}
+	}
+
+	var eligibleReadyInstances []*file.MultishareInstance
+	for _, instanceKey := range readyInstanceKeys {
+		project, location, instanceName, err := util.ParseInstanceHandle(string(instanceKey))
+		if err != nil {
+			klog.Warningf("Failed to parse instance key %v", instanceKey)
+			continue
+		}
+
+		instance, err := m.cloud.File.GetMultishareInstance(ctx, &file.MultishareInstance{
+			Project:  project,
+			Location: location,
+			Name:     instanceName,
+		})
+		if err != nil {
+			if file.IsNotFoundErr(err) {
+				klog.Infof("Instance %v not found, clear from map", instanceKey)
+				m.cache.DeleteInstance(instanceScPrefix, instanceKey)
+			} else {
+				klog.Warningf("Failed to GET for instance key %v, err:%v", instanceKey, err)
+			}
+			continue
+		}
+
+		shares, err := m.cloud.File.ListShares(ctx, &file.ListFilter{Project: project, Location: location, InstanceName: instanceName})
+		// TODO: verify what is the behavior for 0 share instance? i.e. len(shares)=0, err = nil?
+		if err != nil {
+			klog.Warningf("failed to list shares for Instance %q, err: %v", instanceKey, err)
+			continue
+		}
+
+		if len(shares) < util.MaxSharesPerInstance {
+			eligibleReadyInstances = append(eligibleReadyInstances, instance)
+		}
+	}
+
+	return eligibleReadyInstances, nonReadyInstanceCount, nil
+}
+
+func (m *MultishareOpsManager) instanceNeedsExpand(ctx context.Context, share *file.Share) (bool, int64, error) {
+	if share == nil {
+		return false, 0, fmt.Errorf("empty share")
+	}
+	if share.Parent == nil {
+		return false, 0, fmt.Errorf("Parent missing from share %q", share.Name)
+	}
+
+	shares, err := m.cloud.File.ListShares(ctx, &file.ListFilter{Project: share.Parent.Project, Location: share.Parent.Location, InstanceName: share.Parent.Name})
+	if err != nil {
+		return false, 0, err
+	}
+
+	var sumShareBytes int64
+	for _, s := range shares {
+		sumShareBytes = sumShareBytes + s.CapacityBytes
+	}
+	// TODO: Check if we need to align the increment to step size.
+	var remainingBytes int64
+	remainingBytes = share.Parent.CapacityBytes - sumShareBytes
+	if remainingBytes < share.CapacityBytes {
+		return true, share.Parent.CapacityBytes + (share.CapacityBytes - remainingBytes), nil
+	}
+	return false, 0, nil
+}
diff --git a/pkg/csi_driver/volume_id.go b/pkg/csi_driver/volume_id.go
index 9b091eb2..9df6db48 100644
--- a/pkg/csi_driver/volume_id.go
+++ b/pkg/csi_driver/volume_id.go
@@ -58,3 +58,11 @@ func getFileInstanceFromID(id string) (*file.ServiceInstance, string, error) {
 		Volume:   file.Volume{Name: tokens[idVolume]},
 	}, tokens[idProvisioningMode], nil
 }
+
+func generateMultishareVolumeIdFromShare(s *file.Share) (string, error) {
+	if s == nil || s.Parent == nil {
+		return "", fmt.Errorf("invalid share object")
+	}
+
+	return fmt.Sprintf("%s/%s/%s/%s/%s", modeMultishare, s.Parent.Project, s.Parent.Location, s.Parent.Name, s.Name), nil
+}
diff --git a/pkg/util/cache.go b/pkg/util/cache.go
index a2818c70..6ab12ba8 100644
--- a/pkg/util/cache.go
+++ b/pkg/util/cache.go
@@ -18,19 +18,6 @@ package util
 
 import (
 	"fmt"
-	"sync"
-)
-
-type OperationType int
-
-const (
-	InstanceCreate OperationType = iota
-	InstanceDelete
-	InstanceExpand
-	InstanceShrink
-	ShareCreate
-	ShareDelete
-	ShareExpand
 )
 
 type OpInfo struct {
@@ -39,8 +26,23 @@ type OpInfo struct {
 }
 
 type ShareCreateOpInfo struct {
-	Instance string // Placeholder instance
-	OpName   string // share create op
+	InstanceHandle string // Placeholder instance of the form project/location/instanceName
+	OpName         string // share create op
+}
+
+type ShareCreateMapItem struct {
+	Key    string
+	OpInfo ShareCreateOpInfo
+}
+
+type ShareOpsMapItem struct {
+	Key    ShareKey
+	OpInfo OpInfo
+}
+
+type InstanceMapItem struct {
+	Key    InstanceKey
+	OpInfo OpInfo
 }
 
 type InstanceKey string
@@ -54,25 +56,27 @@ func CreateShareKey(project, location, instanceName, shareName string) ShareKey
 	return ShareKey(fmt.Sprintf("%s/%s/%s/%s", project, location, instanceName, shareName))
 }
 
-// Map of last known instance ops running for a given instance
-type InstanceMap struct {
-	m map[InstanceKey]OpInfo
-}
+// This map serves two purpose. To capture any ongoing instance ops, and to keep track of list of known filestore instances that belong to a given storage class.
+type InstanceMap map[InstanceKey]OpInfo
 
-func (s *InstanceMap) Get(key InstanceKey) *OpInfo {
-	k, ok := s.m[key]
+func (s InstanceMap) Get(key InstanceKey) *OpInfo {
+	k, ok := s[key]
 	if !ok {
 		return nil
 	}
 	return &k
 }
 
-func (s *InstanceMap) Add(key InstanceKey, o OpInfo) {
-	s.m[key] = o
+func (s InstanceMap) Add(key InstanceKey, o OpInfo) {
+	s[key] = o
+}
+
+func (s InstanceMap) DeleteKey(key InstanceKey) {
+	delete(s, key)
 }
 
-func (s *InstanceMap) Delete(key InstanceKey, opName string) error {
-	k, ok := s.m[key]
+func (s InstanceMap) DeleteValue(key InstanceKey, opName string) error {
+	k, ok := s[key]
 	if !ok {
 		return nil
 	}
@@ -81,35 +85,43 @@ func (s *InstanceMap) Delete(key InstanceKey, opName string) error {
 		return fmt.Errorf("For key %q, cannot clear op %q, cache already contains op %q", key, opName, k.Name)
 	}
 
-	delete(s.m, key)
+	s[key] = OpInfo{Name: "", Type: UnknownOp}
 	return nil
 }
 
-func NewInstanceMap() *InstanceMap {
-	return &InstanceMap{
-		m: make(map[InstanceKey]OpInfo),
+func (s InstanceMap) Keys() []InstanceKey {
+	keys := make([]InstanceKey, len(s))
+	for k := range s {
+		keys = append(keys, k)
 	}
+	return keys
 }
 
-// Map of share name to last known share creation operation
-type ShareCreateMap struct {
-	m map[string]ShareCreateOpInfo
+func (s InstanceMap) Items() []InstanceMapItem {
+	var items []InstanceMapItem
+	for k, v := range s {
+		items = append(items, InstanceMapItem{Key: k, OpInfo: v})
+	}
+	return items
 }
 
-func (s *ShareCreateMap) Get(key string) *ShareCreateOpInfo {
-	k, ok := s.m[key]
+// Map of share name to last known share creation operation
+type ShareCreateMap map[string]ShareCreateOpInfo
+
+func (s ShareCreateMap) Get(key string) *ShareCreateOpInfo {
+	k, ok := s[key]
 	if !ok {
 		return nil
 	}
 	return &k
 }
 
-func (s *ShareCreateMap) Add(key string, v ShareCreateOpInfo) {
-	s.m[key] = v
+func (s ShareCreateMap) Add(key string, v ShareCreateOpInfo) {
+	s[key] = v
 }
 
-func (s *ShareCreateMap) Delete(key string, opName string) error {
-	k, ok := s.m[key]
+func (s ShareCreateMap) Delete(key string, opName string) error {
+	k, ok := s[key]
 	if !ok {
 		return nil
 	}
@@ -117,37 +129,43 @@ func (s *ShareCreateMap) Delete(key string, opName string) error {
 	if k.OpName != opName {
 		return fmt.Errorf("For key %q, cannot clear op %q, cache already contains op %q", key, opName, k.OpName)
 	}
-	delete(s.m, key)
+	delete(s, key)
 	return nil
 }
 
-func NewShareCreateMap() *ShareCreateMap {
-	return &ShareCreateMap{
-		m: make(map[string]ShareCreateOpInfo),
+func (s ShareCreateMap) Keys() []string {
+	keys := make([]string, len(s))
+	for k := range s {
+		keys = append(keys, k)
 	}
+	return keys
 }
 
-// Map of share handle to share operation details
-type ShareOpsMap struct {
-	m map[ShareKey]OpInfo
-}
-
-func NewShareOpsMap() *ShareOpsMap {
-	return &ShareOpsMap{
-		m: make(map[ShareKey]OpInfo),
+func (s ShareCreateMap) Items() []ShareCreateMapItem {
+	var items []ShareCreateMapItem
+	for k, v := range s {
+		items = append(items, ShareCreateMapItem{Key: k, OpInfo: v})
 	}
+	return items
 }
 
-func (s *ShareOpsMap) Get(key ShareKey) *OpInfo {
-	k, ok := s.m[key]
+// Map of share handle to share operation details
+type ShareOpsMap map[ShareKey]OpInfo
+
+func (s ShareOpsMap) Get(key ShareKey) *OpInfo {
+	k, ok := s[key]
 	if !ok {
 		return nil
 	}
 	return &k
 }
 
-func (s *ShareOpsMap) Delete(key ShareKey, opName string) error {
-	k, ok := s.m[key]
+func (s ShareOpsMap) Add(key ShareKey, v OpInfo) {
+	s[key] = v
+}
+
+func (s ShareOpsMap) Delete(key ShareKey, opName string) error {
+	k, ok := s[key]
 	if !ok {
 		return nil
 	}
@@ -156,25 +174,148 @@ func (s *ShareOpsMap) Delete(key ShareKey, opName string) error {
 		return fmt.Errorf("For key %q, cannot clear op %q, cache already contains op %q", key, opName, k.Name)
 	}
 
-	delete(s.m, key)
+	delete(s, key)
 	return nil
 }
 
+func (s ShareOpsMap) Keys() []ShareKey {
+	var keys []ShareKey
+	for k := range s {
+		keys = append(keys, k)
+	}
+	return keys
+}
+
+func (s ShareOpsMap) Items() []ShareOpsMapItem {
+	var items []ShareOpsMapItem
+	for k, v := range s {
+		items = append(items, ShareOpsMapItem{Key: k, OpInfo: v})
+	}
+	return items
+}
+
 type StorageClassInfo struct {
-	InstanceMap    *InstanceMap    // Map of Filestore instance handle to last known Instance ops.
-	ShareCreateMap *ShareCreateMap // Map of share name to last known ongoing share creation operation on a given placeholder Filestore instance.
-	ShareOpsMap    *ShareOpsMap    // Map of share handle (project/location/instance/share) to share operation details
-	sync.Mutex
+	InstanceMap    InstanceMap    // Map of Filestore instance handle to last known Instance ops.
+	ShareCreateMap ShareCreateMap // Map of share name to last known ongoing share creation operation on a given placeholder Filestore instance.
+	ShareOpsMap    ShareOpsMap    // Map of share handle (project/location/instance/share) to share operation details
 }
 
 // Map of storageclass name to StorageClassInfo
-type StorageClassInfoMap struct {
+type StorageClassInfoCache struct {
 	ScInfoMap map[string]StorageClassInfo
 	Ready     bool
 }
 
-func NewStorageClassInfoMap() *StorageClassInfoMap {
-	return &StorageClassInfoMap{
+func NewStorageClassInfoCache() *StorageClassInfoCache {
+	return &StorageClassInfoCache{
 		ScInfoMap: make(map[string]StorageClassInfo),
 	}
 }
+
+func (m *StorageClassInfoCache) GetInstanceMap(scInfoMapKey string) InstanceMap {
+	v, ok := m.ScInfoMap[scInfoMapKey]
+	if !ok {
+		return nil
+	}
+	return v.InstanceMap
+}
+
+func (m *StorageClassInfoCache) AddInstanceOp(scInfoMapKey string, instanceKey InstanceKey, opInfo OpInfo) {
+	_, ok := m.ScInfoMap[scInfoMapKey]
+	if !ok {
+		m.ScInfoMap[scInfoMapKey] = StorageClassInfo{
+			InstanceMap:    make(InstanceMap),
+			ShareCreateMap: make(ShareCreateMap),
+			ShareOpsMap:    make(ShareOpsMap),
+		}
+	}
+	m.ScInfoMap[scInfoMapKey].InstanceMap.Add(instanceKey, opInfo)
+}
+
+func (m *StorageClassInfoCache) GetInstanceOp(scInfoMapKey string, instanceKey InstanceKey) *OpInfo {
+	v, ok := m.ScInfoMap[scInfoMapKey]
+	if !ok {
+		return nil
+	}
+	return v.InstanceMap.Get(instanceKey)
+}
+
+func (m *StorageClassInfoCache) DeleteInstanceOp(scInfoMapKey string, instanceKey InstanceKey, opName string) error {
+	v, ok := m.ScInfoMap[scInfoMapKey]
+	if !ok {
+		return nil
+	}
+	return v.InstanceMap.DeleteValue(instanceKey, opName)
+}
+
+func (m *StorageClassInfoCache) DeleteInstance(scInfoMapKey string, instanceKey InstanceKey) {
+	v, ok := m.ScInfoMap[scInfoMapKey]
+	if !ok {
+		return
+	}
+	v.InstanceMap.DeleteKey(instanceKey)
+}
+
+func (m *StorageClassInfoCache) GetShareCreateMap(scInfoMapKey string) ShareCreateMap {
+	v, ok := m.ScInfoMap[scInfoMapKey]
+	if !ok {
+		return nil
+	}
+	return v.ShareCreateMap
+}
+
+func (m *StorageClassInfoCache) GetShareOpsMap(scInfoMapKey string) ShareOpsMap {
+	v, ok := m.ScInfoMap[scInfoMapKey]
+	if !ok {
+		return nil
+	}
+	return v.ShareOpsMap
+}
+
+func (m *StorageClassInfoCache) AddShareCreateOp(scInfoMapKey string, shareName string, opInfo ShareCreateOpInfo) error {
+	if _, ok := m.ScInfoMap[scInfoMapKey]; !ok {
+		return fmt.Errorf("missing key %s", scInfoMapKey)
+	}
+	m.ScInfoMap[scInfoMapKey].ShareCreateMap.Add(shareName, opInfo)
+	return nil
+}
+
+func (m *StorageClassInfoCache) GetShareCreateOp(scInfoMapKey string, shareName string) *ShareCreateOpInfo {
+	v, ok := m.ScInfoMap[scInfoMapKey]
+	if !ok {
+		return nil
+	}
+	return v.ShareCreateMap.Get(shareName)
+}
+
+func (m *StorageClassInfoCache) DeleteShareCreateOp(scInfoMapKey string, shareName string, opName string) error {
+	v, ok := m.ScInfoMap[scInfoMapKey]
+	if !ok {
+		return nil
+	}
+	return v.ShareCreateMap.Delete(shareName, opName)
+}
+
+func (m *StorageClassInfoCache) AddShareOp(scInfoMapKey string, shareKey ShareKey, opInfo OpInfo) error {
+	if _, ok := m.ScInfoMap[scInfoMapKey]; !ok {
+		return fmt.Errorf("missing key %s", scInfoMapKey)
+	}
+	m.ScInfoMap[scInfoMapKey].ShareOpsMap.Add(shareKey, opInfo)
+	return nil
+}
+
+func (m *StorageClassInfoCache) GetShareOp(scInfoMapKey string, shareKey ShareKey) *OpInfo {
+	_, ok := m.ScInfoMap[scInfoMapKey]
+	if !ok {
+		return nil
+	}
+	return m.ScInfoMap[scInfoMapKey].ShareOpsMap.Get(shareKey)
+}
+
+func (m *StorageClassInfoCache) DeleteShareOp(scInfoMapKey string, shareKey ShareKey, opName string) error {
+	_, ok := m.ScInfoMap[scInfoMapKey]
+	if !ok {
+		return nil
+	}
+	return m.ScInfoMap[scInfoMapKey].ShareOpsMap.Delete(shareKey, opName)
+}
diff --git a/pkg/util/multishare_defs.go b/pkg/util/multishare_defs.go
new file mode 100644
index 00000000..94756fb2
--- /dev/null
+++ b/pkg/util/multishare_defs.go
@@ -0,0 +1,54 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package util
+
+const (
+	InstanceHandleSplitLen = 3
+	InstanceURISplitLen    = 6
+	ShareHandleSplitLen    = 4
+	ShareURISplitLen       = 8
+
+	MinMultishareInstanceSizeBytes    int64 = 1 * Tb
+	MaxShareSizeBytes                 int64 = 1 * Tb
+	MinShareSizeBytes                 int64 = 100 * Gb
+	MaxSharesPerInstance                    = 10
+	NewMultishareInstancePrefix             = "fs-"
+	ParamMultishareInstanceScLabel          = "instanceStorageClassLabel"
+	ParamMultishareInstanceScLabelKey       = "storage_gke_io_storage-class-id"
+)
+
+type OperationType int
+
+const (
+	InstanceCreate OperationType = iota
+	InstanceDelete
+	InstanceExpand
+	InstanceShrink
+	ShareCreate
+	ShareDelete
+	ShareExpand
+	UnknownOp
+)
+
+type OperationStatus int
+
+const (
+	StatusSuccess OperationStatus = iota
+	StatusRunning
+	StatusFailed
+	StatusUnknown
+)
diff --git a/pkg/util/util.go b/pkg/util/util.go
index c443adfa..9ce0f1bd 100644
--- a/pkg/util/util.go
+++ b/pkg/util/util.go
@@ -200,3 +200,102 @@ func BackupVolumeSourceToCSIVolumeHandle(backupVolumeSource string) (string, err
 	}
 	return fmt.Sprintf("modeInstance/%s/%s/vol1", splitId[3], splitId[5]), nil
 }
+
+// Multishare util functions.
+
+func ConvertVolToShareName(csiVolName string) string {
+	s := strings.ToLower(csiVolName)
+	return strings.ReplaceAll(s, "-", "_")
+	// TODO: verify regex
+}
+
+func CheckLabelValueRegex(value string) error {
+	// Values can be empty, and have a maximum length of 63 characters.
+	regexValue, _ := regexp.Compile(`^[\p{Ll}0-9_-]{0,63}$`)
+	if !regexValue.MatchString(value) {
+		return fmt.Errorf("value %q is invalid (lowercase letter, digit, _ and - chars are allowed / 0-63 characters", value)
+	}
+
+	return nil
+}
+
+func ParseInstanceHandle(instanceHandle string) (string, string, string, error) {
+	// Expected instance handle <project-name>/<location-name>/<instance-name>
+	splitStr := strings.Split(instanceHandle, "/")
+	if len(splitStr) != InstanceHandleSplitLen {
+		return "", "", "", fmt.Errorf("Unknown instance handle format %q", instanceHandle)
+	}
+
+	project := splitStr[0]
+	location := splitStr[1]
+	instanceName := splitStr[2]
+	if project == "" || location == "" || instanceName == "" {
+		return "", "", "", fmt.Errorf("Unknown instance handle format %q", instanceHandle)
+	}
+
+	return project, location, instanceName, nil
+}
+
+func ParseInstanceURI(instanceURI string) (string, string, string, error) {
+	// Expected instance URI projects/<project-name>/locations/<location-name>/instances/<instance-name>/shares/<share-name>
+	splitStr := strings.Split(instanceURI, "/")
+	if len(splitStr) != InstanceURISplitLen {
+		return "", "", "", fmt.Errorf("Unknown instance URI format %q", instanceURI)
+	}
+
+	project := splitStr[1]
+	location := splitStr[3]
+	instanceName := splitStr[5]
+	if project == "" || location == "" || instanceName == "" {
+		return "", "", "", fmt.Errorf("Unknown instance URI format %q", instanceURI)
+	}
+
+	return project, location, instanceName, nil
+}
+
+func ParseShareHandle(shareHandle string) (string, string, string, string, error) {
+	// Expected share handle <project-name>/<location-name>/<instance-name>/<share-name>
+	splitStr := strings.Split(shareHandle, "/")
+	if len(splitStr) != ShareHandleSplitLen {
+		return "", "", "", "", fmt.Errorf("Unknown share handle format %q", shareHandle)
+	}
+
+	project := splitStr[0]
+	location := splitStr[1]
+	instanceName := splitStr[2]
+	shareName := splitStr[3]
+	if project == "" || location == "" || instanceName == "" || shareName == "" {
+		return "", "", "", "", fmt.Errorf("Unknown share handle format %q", shareHandle)
+	}
+
+	return project, location, instanceName, shareName, nil
+}
+
+func ParseShareURI(shareURI string) (string, string, string, string, error) {
+	// Expected share URI projects/<project-name>/locations/<location-name>/instances/<instance-name>/shares/<share-name>
+	splitStr := strings.Split(shareURI, "/")
+	if len(splitStr) != ShareURISplitLen {
+		return "", "", "", "", fmt.Errorf("Unknown share URI format %q", shareURI)
+	}
+
+	project := splitStr[1]
+	location := splitStr[3]
+	instanceName := splitStr[5]
+	shareName := splitStr[7]
+	if project == "" || location == "" || instanceName == "" || shareName == "" {
+		return "", "", "", "", fmt.Errorf("Unknown share URI format %q", shareURI)
+	}
+
+	return project, location, instanceName, shareName, nil
+}
+
+func GetMultishareOpsTimeoutConfig(opType OperationType) (time.Duration, time.Duration, error) {
+	switch opType {
+	case InstanceCreate, InstanceDelete, ShareDelete:
+		return 1 * time.Hour, 5 * time.Second, nil
+	case InstanceExpand, InstanceShrink, ShareCreate, ShareExpand:
+		return 10 * time.Minute, 5 * time.Second, nil
+	default:
+		return 0, 0, fmt.Errorf("unknown op type %v", opType)
+	}
+}


Test patch:
diff --git a/pkg/csi_driver/multishare_controller_test.go b/pkg/csi_driver/multishare_controller_test.go
new file mode 100644
index 00000000..29d31063
--- /dev/null
+++ b/pkg/csi_driver/multishare_controller_test.go
@@ -0,0 +1,468 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package driver
+
+import (
+	"reflect"
+	"testing"
+
+	csi "github.com/container-storage-interface/spec/lib/go/csi"
+	cloud "sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider"
+	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider/file"
+	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/util"
+)
+
+const (
+	testDriverName           = "test.filestore"
+	testDrivernameLabelValue = "test_filestore"
+	testPVCName              = "testPVC"
+	testPVCNamespace         = "testNamespace"
+	testPVName               = "testPV"
+)
+
+func initTestMultishareController(t *testing.T) *MultishareController {
+	fileService, err := file.NewFakeService()
+	if err != nil {
+		t.Fatalf("failed to initialize GCFS service: %v", err)
+	}
+
+	cloudProvider, err := cloud.NewFakeCloud()
+	if err != nil {
+		t.Fatalf("Failed to get cloud provider: %v", err)
+	}
+	return NewMultishareController(initTestDriver(t), fileService, cloudProvider, util.NewVolumeLocks())
+}
+
+func TestPickRegion(t *testing.T) {
+	tests := []struct {
+		name           string
+		toporeq        *csi.TopologyRequirement
+		expectErr      bool
+		expectedRegion string
+	}{
+		{
+			name:           "empty toppo",
+			expectedRegion: "us-central1",
+		},
+		{
+			name: "empty toppo req and preferred segments",
+			toporeq: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{},
+					},
+				},
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{},
+					},
+				},
+			},
+			expectErr: true,
+		},
+		{
+			name: "empty toppo req segments",
+			toporeq: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{},
+					},
+				},
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "us-central1-c",
+						},
+					},
+				},
+			},
+			expectErr: true,
+		},
+		{
+			name: "empty toppo preferred segments",
+			toporeq: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "us-central1-c",
+						},
+					},
+				},
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{},
+					},
+				},
+			},
+			expectErr: true,
+		},
+		{
+			name: "empty preferred list",
+			toporeq: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "us-central1-c",
+						},
+					},
+				},
+			},
+			expectedRegion: "us-central1",
+		},
+		{
+			name: "empty req list",
+			toporeq: &csi.TopologyRequirement{
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "us-central1-c",
+						},
+					},
+				},
+			},
+			expectedRegion: "us-central1",
+		},
+		{
+			name: "non-empty preferred and req list",
+			toporeq: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "us-central1-a",
+						},
+					},
+				},
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "us-east1-a",
+						},
+					},
+				},
+			},
+			expectedRegion: "us-east1",
+		},
+		{
+			name: "malformed zone name in preferred list",
+			toporeq: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "us-central1-a",
+						},
+					},
+				},
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "us-east1a",
+						},
+					},
+				},
+			},
+			expectErr: true,
+		},
+		{
+			name: "malformed zone name in req list",
+			toporeq: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "us-central1a",
+						},
+					},
+				},
+			},
+			expectErr: true,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			m := initTestMultishareController(t)
+			region, err := m.pickRegion(tc.toporeq)
+			if tc.expectErr && err == nil {
+				t.Error("expected error, got none")
+			}
+			if !tc.expectErr && err != nil {
+				t.Error("unexpected error")
+			}
+
+			if tc.expectedRegion != region {
+				t.Errorf("got %v, want %v", region, tc.expectedRegion)
+			}
+		})
+	}
+}
+
+func TestGetShareRequestCapacity(t *testing.T) {
+	tests := []struct {
+		name             string
+		cap              *csi.CapacityRange
+		expectErr        bool
+		expectedCapacity int64
+	}{
+		{
+			name:      "req and limit not set",
+			cap:       &csi.CapacityRange{},
+			expectErr: true,
+		},
+		{
+			name:             "empty cap range",
+			expectedCapacity: 100 * util.Gb,
+		},
+		{
+			name:      "cap range limit less than minimum supported size",
+			expectErr: true,
+			cap: &csi.CapacityRange{
+				LimitBytes: 99 * util.Gb,
+			},
+		},
+		{
+			name:      "cap range limit more than maximum supported size",
+			expectErr: true,
+			cap: &csi.CapacityRange{
+				LimitBytes: 2 * util.Tb,
+			},
+		},
+		{
+			name:      "cap range req less than minimum supported size",
+			expectErr: true,
+			cap: &csi.CapacityRange{
+				RequiredBytes: 99 * util.Gb,
+			},
+		},
+		{
+			name:      "cap range req more than maximum supported size",
+			expectErr: true,
+			cap: &csi.CapacityRange{
+				RequiredBytes: 2 * util.Tb,
+			},
+		},
+		{
+			name: "cap range req and limit set, limit less than req",
+			cap: &csi.CapacityRange{
+				RequiredBytes: 1 * util.Tb,
+				LimitBytes:    100 * util.Gb,
+			},
+			expectErr: true,
+		},
+		{
+			name: "cap range req and limit set",
+			cap: &csi.CapacityRange{
+				RequiredBytes: 100 * util.Gb,
+				LimitBytes:    1 * util.Tb,
+			},
+			expectedCapacity: 1 * util.Tb,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			capacity, err := getShareRequestCapacity(tc.cap)
+			if tc.expectErr && err == nil {
+				t.Error("expected error, got none")
+			}
+			if !tc.expectErr && err != nil {
+				t.Error("unexpected error")
+			}
+
+			if tc.expectedCapacity != capacity {
+				t.Errorf("got %v, want %v", capacity, tc.expectedCapacity)
+			}
+		})
+	}
+}
+
+func TestExtractInstanceLabels(t *testing.T) {
+	tests := []struct {
+		name          string
+		params        map[string]string
+		driver        string
+		expectedLabel map[string]string
+		expectErr     bool
+	}{
+		{
+			name:   "empty params",
+			driver: testDriverName,
+			expectedLabel: map[string]string{
+				tagKeyCreatedBy: testDrivernameLabelValue,
+			},
+		},
+		{
+			name:   "user labels",
+			driver: testDriverName,
+			params: map[string]string{
+				ParameterKeyLabels:                  "a=b,c=d",
+				util.ParamMultishareInstanceScLabel: "testsc",
+			},
+			expectedLabel: map[string]string{
+				tagKeyCreatedBy:                        testDrivernameLabelValue,
+				util.ParamMultishareInstanceScLabelKey: "testsc",
+				"a":                                    "b",
+				"c":                                    "d",
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			label, err := extractInstanceLabels(tc.params, tc.driver)
+			if tc.expectErr && err == nil {
+				t.Error("expected error, got none")
+			}
+			if !tc.expectErr && err != nil {
+				t.Error("unexpected error")
+			}
+			if len(label) != len(tc.expectedLabel) {
+				t.Errorf("got len %v, want %v", len(label), len(tc.expectedLabel))
+			}
+			for k, v := range tc.expectedLabel {
+				vgot, ok := label[k]
+				if !ok {
+					t.Errorf("key %v missing", k)
+				}
+				if vgot != v {
+					t.Errorf("got %v, want %v", vgot, v)
+				}
+			}
+		})
+	}
+}
+
+func TestExtractShareLabels(t *testing.T) {
+	tests := []struct {
+		name          string
+		params        map[string]string
+		expectedLabel map[string]string
+	}{
+		{
+			name: "empty params",
+		},
+		{
+			name: "user labels ignored",
+			params: map[string]string{
+				ParameterKeyLabels:                  "a=b,c=d",
+				util.ParamMultishareInstanceScLabel: "testsc",
+			},
+		},
+		{
+			name: "driver labels",
+			params: map[string]string{
+				ParameterKeyLabels:                  "a=b,c=d",
+				util.ParamMultishareInstanceScLabel: "testsc",
+				ParameterKeyPVCName:                 testPVCName,
+				ParameterKeyPVCNamespace:            testPVCNamespace,
+				ParameterKeyPVName:                  testPVName,
+			},
+			expectedLabel: map[string]string{
+				tagKeyCreatedForClaimName:      testPVCName,
+				tagKeyCreatedForClaimNamespace: testPVCNamespace,
+				tagKeyCreatedForVolumeName:     testPVName,
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			label := extractShareLabels(tc.params)
+			if len(label) != len(tc.expectedLabel) {
+				t.Errorf("got len %v, want %v", len(label), len(tc.expectedLabel))
+			}
+			for k, v := range tc.expectedLabel {
+				vgot, ok := label[k]
+				if !ok {
+					t.Errorf("key %v missing", k)
+				}
+				if vgot != v {
+					t.Errorf("got %v, want %v", vgot, v)
+				}
+			}
+		})
+	}
+}
+
+func TestGenerateNewMultishareInstance(t *testing.T) {
+	tests := []struct {
+		name             string
+		instanceName     string
+		req              *csi.CreateVolumeRequest
+		expectedInstance *file.MultishareInstance
+		expectErr        bool
+	}{
+		{
+			name:         "non enterprise tier",
+			instanceName: testInstanceName,
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramTier: "standard",
+				},
+			},
+			expectErr: true,
+		},
+		{
+			name:         "invalid connect mode",
+			instanceName: testInstanceName,
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramConnectMode: "blah",
+				},
+			},
+			expectErr: true,
+		},
+		{
+			name:         "valid params",
+			instanceName: testInstanceName,
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramConnectMode:                    directPeering,
+					ParameterKeyLabels:                  "a=b,c=d",
+					util.ParamMultishareInstanceScLabel: testInstanceScPrefix,
+				},
+			},
+			expectedInstance: &file.MultishareInstance{
+				Project:       "test-project",
+				Location:      "us-central1",
+				Name:          testInstanceName,
+				CapacityBytes: util.MinMultishareInstanceSizeBytes,
+				Network: file.Network{
+					Name:        "default",
+					ConnectMode: directPeering,
+				},
+				Tier:       enterpriseTier,
+				KmsKeyName: "",
+				Labels: map[string]string{
+					"a":                                    "b",
+					"c":                                    "d",
+					tagKeyCreatedBy:                        "test-driver",
+					util.ParamMultishareInstanceScLabelKey: testInstanceScPrefix,
+				},
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			m := initTestMultishareController(t)
+			filer, err := m.generateNewMultishareInstance(tc.instanceName, tc.req)
+			if tc.expectErr && err == nil {
+				t.Error("expected error, got none")
+			}
+			if !tc.expectErr && err != nil {
+				t.Error("unexpected error")
+			}
+			if !reflect.DeepEqual(filer, tc.expectedInstance) {
+				t.Errorf("got filer %+v, want %+v", filer, tc.expectedInstance)
+			}
+		})
+	}
+}
diff --git a/pkg/csi_driver/multishare_ops_manager_test.go b/pkg/csi_driver/multishare_ops_manager_test.go
new file mode 100644
index 00000000..b8626fad
--- /dev/null
+++ b/pkg/csi_driver/multishare_ops_manager_test.go
@@ -0,0 +1,1732 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package driver
+
+import (
+	"fmt"
+	"testing"
+
+	"golang.org/x/net/context"
+	cloud "sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider"
+	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider/file"
+	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/util"
+)
+
+const (
+	testInstanceScPrefix = "testinstancescprefix"
+	testInstanceName     = "testInstanceName"
+	testShareName        = "testShareName"
+)
+
+var (
+	testInstanceHandle = fmt.Sprintf("%s/%s/%s", testProject, testRegion, testInstanceName)
+)
+
+type Item struct {
+	scKey          string
+	instanceKey    util.InstanceKey
+	shareKey       util.ShareKey
+	shareCreateKey string
+	op             util.OpInfo
+	createOp       util.ShareCreateOpInfo
+}
+
+func initCloudProviderWithBlockingFileService(t *testing.T, opUnblocker chan chan file.Signal) *cloud.Cloud {
+	fbs, err := file.NewFakeBlockingServiceForMultishare(opUnblocker)
+	if err != nil {
+		t.Errorf("failed to initialize blocking GCFS service: %v", err)
+	}
+
+	cloudProvider, err := cloud.NewFakeCloudWithFiler(fbs, testProject, testLocation)
+	if err != nil {
+		t.Errorf("failed to initialize blocking GCFS service: %v", err)
+	}
+	return cloudProvider
+}
+
+type MockOpStatus struct {
+	reportRunning           bool
+	reportError             bool
+	reportNotFoundError     bool
+	reportOpWithErrorStatus bool
+}
+
+type Response struct {
+	opStatus             util.OperationStatus
+	createOp             *util.ShareCreateOpInfo
+	shareOp              *util.OpInfo
+	verified             bool
+	readyInstances       []*file.MultishareInstance
+	numNonReadyInstances int
+	instanceNeedsExpand  bool
+	targetBytes          int64
+	err                  error
+}
+
+func TestCheckAndUpdateShareCreateOp(t *testing.T) {
+	tests := []struct {
+		name                      string
+		scKey                     string
+		shareName                 string
+		initShareCreateOpMap      []Item
+		signalGetOp               bool
+		signalIsOpDone            bool
+		getOpStatus               *MockOpStatus
+		isOpDoneStatus            *MockOpStatus
+		expectedShareCreateOpInfo *util.ShareCreateOpInfo
+		expectedOpStaus           util.OperationStatus
+		shareKeyExpectedInCache   bool
+	}{
+		{
+			name:            "tc1 - no share create op in cache, unknown op status",
+			scKey:           testInstanceScPrefix,
+			shareName:       testShareName,
+			expectedOpStaus: util.StatusUnknown,
+		},
+		{
+			name:      "tc2 - get share create op returns error, unknown op status, cache entry not cleared",
+			scKey:     testInstanceScPrefix,
+			shareName: testShareName,
+			initShareCreateOpMap: []Item{
+				{
+					scKey:          testInstanceScPrefix,
+					shareCreateKey: testShareName,
+					createOp: util.ShareCreateOpInfo{
+						InstanceHandle: testInstanceHandle,
+						OpName:         "op-1",
+					},
+				},
+			},
+			signalGetOp: true,
+			getOpStatus: &MockOpStatus{
+				reportError: true,
+			},
+			expectedOpStaus: util.StatusUnknown,
+			expectedShareCreateOpInfo: &util.ShareCreateOpInfo{
+				InstanceHandle: testInstanceHandle,
+				OpName:         "op-1",
+			},
+			shareKeyExpectedInCache: true,
+		},
+		{
+			name:      "tc3 - IsOpDone returns error, return failed op status, cache entry cleared",
+			scKey:     testInstanceScPrefix,
+			shareName: testShareName,
+			initShareCreateOpMap: []Item{
+				{
+					scKey:          testInstanceScPrefix,
+					shareCreateKey: testShareName,
+					createOp: util.ShareCreateOpInfo{
+						InstanceHandle: testInstanceHandle,
+						OpName:         "op-1",
+					},
+				},
+			},
+			signalGetOp:    true,
+			signalIsOpDone: true,
+			isOpDoneStatus: &MockOpStatus{
+				reportError: true,
+			},
+			expectedOpStaus: util.StatusFailed,
+			expectedShareCreateOpInfo: &util.ShareCreateOpInfo{
+				InstanceHandle: testInstanceHandle,
+				OpName:         "op-1",
+			},
+		},
+		{
+			name:      "tc4 - IsOpDone false, return running op status, cache entry not cleared",
+			scKey:     testInstanceScPrefix,
+			shareName: testShareName,
+			initShareCreateOpMap: []Item{
+				{
+					scKey:          testInstanceScPrefix,
+					shareCreateKey: testShareName,
+					createOp: util.ShareCreateOpInfo{
+						InstanceHandle: testInstanceHandle,
+						OpName:         "op-1",
+					},
+				},
+			},
+			signalGetOp:    true,
+			signalIsOpDone: true,
+			isOpDoneStatus: &MockOpStatus{
+				reportRunning: true,
+			},
+			expectedOpStaus: util.StatusRunning,
+			expectedShareCreateOpInfo: &util.ShareCreateOpInfo{
+				InstanceHandle: testInstanceHandle,
+				OpName:         "op-1",
+			},
+			shareKeyExpectedInCache: true,
+		},
+		{
+			name:      "tc4 - IsOpDone true, return success op status, cache entry cleared",
+			scKey:     testInstanceScPrefix,
+			shareName: testShareName,
+			initShareCreateOpMap: []Item{
+				{
+					scKey:          testInstanceScPrefix,
+					shareCreateKey: testShareName,
+					createOp: util.ShareCreateOpInfo{
+						InstanceHandle: testInstanceHandle,
+						OpName:         "op-1",
+					},
+				},
+			},
+			signalGetOp:     true,
+			signalIsOpDone:  true,
+			expectedOpStaus: util.StatusSuccess,
+			expectedShareCreateOpInfo: &util.ShareCreateOpInfo{
+				InstanceHandle: testInstanceHandle,
+				OpName:         "op-1",
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			opUnblocker := make(chan chan file.Signal)
+			cloudProvider := initCloudProviderWithBlockingFileService(t, opUnblocker)
+			manager := NewMultishareOpsManager(cloudProvider)
+
+			runRequest := func(ctx context.Context, instanceSCPrefix, shareName string) <-chan Response {
+				responseChannel := make(chan Response)
+				go func() {
+					op, status, err := manager.checkAndUpdateShareCreateOp(ctx, instanceSCPrefix, shareName)
+					responseChannel <- Response{
+						opStatus: status,
+						createOp: op,
+						err:      err,
+					}
+				}()
+				return responseChannel
+			}
+
+			// Pre-populate the cache.
+			for _, v := range tc.initShareCreateOpMap {
+				if _, ok := manager.cache.ScInfoMap[v.scKey]; !ok {
+					manager.cache.ScInfoMap[v.scKey] = util.StorageClassInfo{
+						InstanceMap:    make(util.InstanceMap),
+						ShareCreateMap: make(util.ShareCreateMap),
+						ShareOpsMap:    make(util.ShareOpsMap),
+					}
+				}
+				err := manager.cache.AddShareCreateOp(v.scKey, v.shareCreateKey, v.createOp)
+				if err != nil {
+					t.Errorf("failed to add share create op")
+				}
+			}
+
+			respChannel := runRequest(context.Background(), tc.scKey, tc.shareName)
+
+			// Inject mock response
+			if tc.signalGetOp {
+				s := file.Signal{}
+				if tc.getOpStatus != nil {
+					s.ReportError = tc.getOpStatus.reportError
+					s.ReportOpWithErrorStatus = tc.getOpStatus.reportOpWithErrorStatus
+					s.ReportRunning = tc.getOpStatus.reportRunning
+				}
+				execute := <-opUnblocker
+				execute <- s
+			}
+
+			// Inject mock response
+			if tc.signalIsOpDone {
+				s := file.Signal{}
+				if tc.isOpDoneStatus != nil {
+					s.ReportError = tc.isOpDoneStatus.reportError
+					s.ReportOpWithErrorStatus = tc.isOpDoneStatus.reportOpWithErrorStatus
+					s.ReportRunning = tc.isOpDoneStatus.reportRunning
+				}
+				execute := <-opUnblocker
+				execute <- s
+			}
+
+			// Verify response
+			response := <-respChannel
+			if response.opStatus != tc.expectedOpStaus {
+				t.Errorf("op status want %v, got %v", tc.expectedOpStaus, response.opStatus)
+			}
+			if tc.expectedShareCreateOpInfo == nil && response.createOp != nil {
+				t.Errorf("unexpected share create op found")
+			}
+
+			if tc.expectedShareCreateOpInfo != nil && response.createOp == nil {
+				t.Errorf("expected share create op not found")
+			}
+
+			if tc.expectedShareCreateOpInfo != nil && response.createOp != nil {
+				if tc.expectedShareCreateOpInfo.InstanceHandle != response.createOp.InstanceHandle {
+					t.Errorf("want %v, got %v", tc.expectedShareCreateOpInfo.InstanceHandle, response.createOp.InstanceHandle)
+				}
+				if tc.expectedShareCreateOpInfo.OpName != response.createOp.OpName {
+					t.Errorf("want %v, got %v", tc.expectedShareCreateOpInfo.OpName, response.createOp.OpName)
+				}
+			}
+
+			// Verify cache content
+			shareCreateOp := manager.cache.GetShareCreateOp(tc.scKey, tc.shareName)
+			if tc.shareKeyExpectedInCache && shareCreateOp == nil {
+				t.Errorf("expcted share key not found")
+			}
+			if !tc.shareKeyExpectedInCache && shareCreateOp != nil {
+				t.Errorf("unexpcted share key found")
+			}
+		})
+	}
+}
+
+func TestCheckAndUpdateShareOp(t *testing.T) {
+	testShare := &file.Share{
+		Name: testShareName,
+		Parent: &file.MultishareInstance{
+			Project:  testProject,
+			Location: testRegion,
+			Name:     testInstanceName,
+		},
+	}
+	tests := []struct {
+		name                    string
+		scKey                   string
+		shareKey                util.ShareKey
+		share                   *file.Share
+		initShareOpMap          []Item
+		signalGetOp             bool
+		signalIsOpDone          bool
+		getOpStatus             *MockOpStatus
+		isOpDoneStatus          *MockOpStatus
+		expectedShareOpInfo     *util.OpInfo
+		expectedOpStaus         util.OperationStatus
+		shareKeyExpectedInCache bool
+	}{
+		{
+			name:            "tc1 - no share op in cache, unknown op status",
+			scKey:           testInstanceScPrefix,
+			share:           testShare,
+			expectedOpStaus: util.StatusUnknown,
+		},
+		{
+			name:     "tc2 - get share op returns error, unknown op status, cache entry not cleared",
+			scKey:    testInstanceScPrefix,
+			share:    testShare,
+			shareKey: util.CreateShareKey(testProject, testRegion, testInstanceName, testShareName),
+			initShareOpMap: []Item{
+				{
+					scKey:    testInstanceScPrefix,
+					shareKey: util.CreateShareKey(testProject, testRegion, testInstanceName, testShareName),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.ShareDelete,
+					},
+				},
+			},
+			signalGetOp: true,
+			getOpStatus: &MockOpStatus{
+				reportError: true,
+			},
+			expectedOpStaus: util.StatusUnknown,
+			expectedShareOpInfo: &util.OpInfo{
+				Name: "op-1",
+				Type: util.ShareDelete,
+			},
+			shareKeyExpectedInCache: true,
+		},
+		{
+			name:     "tc3 - IsOpDone returns error, return failed op status, cache entry cleared",
+			scKey:    testInstanceScPrefix,
+			share:    testShare,
+			shareKey: util.CreateShareKey(testProject, testRegion, testInstanceName, testShareName),
+			initShareOpMap: []Item{
+				{
+					scKey:    testInstanceScPrefix,
+					shareKey: util.CreateShareKey(testProject, testRegion, testInstanceName, testShareName),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.ShareDelete,
+					},
+				},
+			},
+			signalGetOp:    true,
+			signalIsOpDone: true,
+			isOpDoneStatus: &MockOpStatus{
+				reportError: true,
+			},
+			expectedOpStaus: util.StatusFailed,
+			expectedShareOpInfo: &util.OpInfo{
+				Name: "op-1",
+				Type: util.ShareDelete,
+			},
+		},
+		{
+			name:     "tc4 - IsOpDone false, return running op status, cache entry not cleared",
+			scKey:    testInstanceScPrefix,
+			share:    testShare,
+			shareKey: util.CreateShareKey(testProject, testRegion, testInstanceName, testShareName),
+			initShareOpMap: []Item{
+				{
+					scKey:    testInstanceScPrefix,
+					shareKey: util.CreateShareKey(testProject, testRegion, testInstanceName, testShareName),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.ShareDelete,
+					},
+				},
+			},
+			signalGetOp:    true,
+			signalIsOpDone: true,
+			isOpDoneStatus: &MockOpStatus{
+				reportRunning: true,
+			},
+			expectedOpStaus: util.StatusRunning,
+			expectedShareOpInfo: &util.OpInfo{
+				Name: "op-1",
+				Type: util.ShareDelete,
+			},
+			shareKeyExpectedInCache: true,
+		},
+		{
+			name:     "tc4 - IsOpDone true, return success op status, cache entry cleared",
+			scKey:    testInstanceScPrefix,
+			share:    testShare,
+			shareKey: util.CreateShareKey(testProject, testRegion, testInstanceName, testShareName),
+			initShareOpMap: []Item{
+				{
+					scKey:    testInstanceScPrefix,
+					shareKey: util.CreateShareKey(testProject, testRegion, testInstanceName, testShareName),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.ShareDelete,
+					},
+				},
+			},
+			signalGetOp:     true,
+			signalIsOpDone:  true,
+			expectedOpStaus: util.StatusSuccess,
+			expectedShareOpInfo: &util.OpInfo{
+				Name: "op-1",
+				Type: util.ShareDelete,
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			opUnblocker := make(chan chan file.Signal)
+			cloudProvider := initCloudProviderWithBlockingFileService(t, opUnblocker)
+			manager := NewMultishareOpsManager(cloudProvider)
+
+			runRequest := func(ctx context.Context, instanceSCPrefix string, share *file.Share) <-chan Response {
+				responseChannel := make(chan Response)
+				go func() {
+					op, status, err := manager.checkAndUpdateShareOp(ctx, instanceSCPrefix, share)
+					responseChannel <- Response{
+						opStatus: status,
+						shareOp:  op,
+						err:      err,
+					}
+				}()
+				return responseChannel
+			}
+
+			// Pre-populate the cache.
+			for _, v := range tc.initShareOpMap {
+				if _, ok := manager.cache.ScInfoMap[v.scKey]; !ok {
+					manager.cache.ScInfoMap[v.scKey] = util.StorageClassInfo{
+						InstanceMap:    make(util.InstanceMap),
+						ShareCreateMap: make(util.ShareCreateMap),
+						ShareOpsMap:    make(util.ShareOpsMap),
+					}
+				}
+				err := manager.cache.AddShareOp(v.scKey, v.shareKey, v.op)
+				if err != nil {
+					t.Errorf("failed to add share op to map")
+				}
+			}
+
+			respChannel := runRequest(context.Background(), tc.scKey, tc.share)
+
+			// Inject mock response
+			if tc.signalGetOp {
+				s := file.Signal{}
+				if tc.getOpStatus != nil {
+					s.ReportError = tc.getOpStatus.reportError
+					s.ReportOpWithErrorStatus = tc.getOpStatus.reportOpWithErrorStatus
+					s.ReportRunning = tc.getOpStatus.reportRunning
+				}
+				execute := <-opUnblocker
+				execute <- s
+			}
+
+			// Inject mock response
+			if tc.signalIsOpDone {
+				s := file.Signal{}
+				if tc.isOpDoneStatus != nil {
+					s.ReportError = tc.isOpDoneStatus.reportError
+					s.ReportOpWithErrorStatus = tc.isOpDoneStatus.reportOpWithErrorStatus
+					s.ReportRunning = tc.isOpDoneStatus.reportRunning
+				}
+				execute := <-opUnblocker
+				execute <- s
+			}
+
+			// Verify response
+			response := <-respChannel
+			if response.opStatus != tc.expectedOpStaus {
+				t.Errorf("op status want %v, got %v", tc.expectedOpStaus, response.opStatus)
+			}
+			if tc.expectedShareOpInfo == nil && response.shareOp != nil {
+				t.Errorf("unexpected share create op found")
+			}
+
+			if tc.expectedShareOpInfo != nil && response.shareOp == nil {
+				t.Errorf("expected share create op not found")
+			}
+
+			if tc.expectedShareOpInfo != nil && response.shareOp != nil {
+				if tc.expectedShareOpInfo.Name != response.shareOp.Name {
+					t.Errorf("want %v, got %v", tc.expectedShareOpInfo.Name, response.shareOp.Name)
+				}
+				if tc.expectedShareOpInfo.Type != response.shareOp.Type {
+					t.Errorf("want %v, got %v", tc.expectedShareOpInfo.Type, response.shareOp.Type)
+				}
+			}
+
+			// Verify cache content
+			shareOp := manager.cache.GetShareOp(tc.scKey, tc.shareKey)
+			if tc.shareKeyExpectedInCache && shareOp == nil {
+				t.Errorf("expcted share key not found")
+			}
+			if !tc.shareKeyExpectedInCache && shareOp != nil {
+				t.Errorf("unexpcted share key found")
+			}
+		})
+	}
+}
+
+func TestVerifyNoRunningInstanceOps(t *testing.T) {
+	tests := []struct {
+		name                       string
+		scKey                      string
+		instanceKey                util.InstanceKey
+		instance                   *file.MultishareInstance
+		initInstanceOpMap          []Item
+		signalGetOp                bool
+		signalIsOpDone             bool
+		getOpStatus                *MockOpStatus
+		isOpDoneStatus             *MockOpStatus
+		errorExpected              bool
+		expectedVerificationStaus  bool
+		instanceKeyExpectedInCache bool
+		emptyOpExpected            bool
+	}{
+		{
+			name:        "tc1 - no instance op in cache, verified",
+			scKey:       testInstanceScPrefix,
+			instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName),
+			instance: &file.MultishareInstance{
+				Project:  testProject,
+				Location: testRegion,
+				Name:     testInstanceName,
+			},
+			expectedVerificationStaus: true,
+		},
+		{
+			name:        "tc2 - get instance op returns empty op, verified",
+			scKey:       testInstanceScPrefix,
+			instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName),
+			instance: &file.MultishareInstance{
+				Project:  testProject,
+				Location: testRegion,
+				Name:     testInstanceName,
+			},
+			initInstanceOpMap: []Item{
+				{
+					scKey:       testInstanceScPrefix,
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName),
+					op:          util.OpInfo{},
+				},
+			},
+			instanceKeyExpectedInCache: true,
+			emptyOpExpected:            true,
+			expectedVerificationStaus:  true,
+		},
+
+		{
+			name:        "tc3 - get instance op returns error, status not verified",
+			scKey:       testInstanceScPrefix,
+			instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName),
+			instance: &file.MultishareInstance{
+				Project:  testProject,
+				Location: testRegion,
+				Name:     testInstanceName,
+			},
+			initInstanceOpMap: []Item{
+				{
+					scKey:       testInstanceScPrefix,
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.InstanceCreate,
+					},
+				},
+			},
+			signalGetOp: true,
+			getOpStatus: &MockOpStatus{
+				reportError: true,
+			},
+			errorExpected:              true,
+			instanceKeyExpectedInCache: true,
+		},
+		{
+			name:        "tc4 - IsOpDone returns error for completeed op, status verified",
+			scKey:       testInstanceScPrefix,
+			instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName),
+			instance: &file.MultishareInstance{
+				Project:  testProject,
+				Location: testRegion,
+				Name:     testInstanceName,
+			},
+			initInstanceOpMap: []Item{
+				{
+					scKey:       testInstanceScPrefix,
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.InstanceCreate,
+					},
+				},
+			},
+			signalGetOp:    true,
+			signalIsOpDone: true,
+			isOpDoneStatus: &MockOpStatus{
+				reportError: true,
+			},
+			instanceKeyExpectedInCache: true,
+			expectedVerificationStaus:  true,
+		},
+		{
+			name:        "tc5 - IsOpDone returns false, status not verified",
+			scKey:       testInstanceScPrefix,
+			instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName),
+			instance: &file.MultishareInstance{
+				Project:  testProject,
+				Location: testRegion,
+				Name:     testInstanceName,
+			},
+			initInstanceOpMap: []Item{
+				{
+					scKey:       testInstanceScPrefix,
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.InstanceCreate,
+					},
+				},
+			},
+			signalGetOp:    true,
+			signalIsOpDone: true,
+			isOpDoneStatus: &MockOpStatus{
+				reportRunning: true,
+			},
+			instanceKeyExpectedInCache: true,
+			expectedVerificationStaus:  false,
+		},
+		{
+			name:        "tc5 - IsOpDone true, status verified",
+			scKey:       testInstanceScPrefix,
+			instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName),
+			instance: &file.MultishareInstance{
+				Project:  testProject,
+				Location: testRegion,
+				Name:     testInstanceName,
+			},
+			initInstanceOpMap: []Item{
+				{
+					scKey:       testInstanceScPrefix,
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.InstanceCreate,
+					},
+				},
+			},
+			signalGetOp:    true,
+			signalIsOpDone: true,
+			isOpDoneStatus: &MockOpStatus{
+				reportRunning: true,
+			},
+			instanceKeyExpectedInCache: true,
+			expectedVerificationStaus:  false,
+		}}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			opUnblocker := make(chan chan file.Signal)
+			cloudProvider := initCloudProviderWithBlockingFileService(t, opUnblocker)
+			manager := NewMultishareOpsManager(cloudProvider)
+
+			runRequest := func(ctx context.Context, instanceSCPrefix string, instance *file.MultishareInstance) <-chan Response {
+				responseChannel := make(chan Response)
+				go func() {
+					status, err := manager.verifyNoRunningInstanceOps(ctx, instanceSCPrefix, instance)
+					responseChannel <- Response{
+						verified: status,
+						err:      err,
+					}
+				}()
+				return responseChannel
+			}
+
+			// Pre-populate the cache.
+			for _, v := range tc.initInstanceOpMap {
+				manager.cache.AddInstanceOp(v.scKey, v.instanceKey, v.op)
+			}
+
+			respChannel := runRequest(context.Background(), tc.scKey, tc.instance)
+
+			// Inject mock response
+			if tc.signalGetOp {
+				s := file.Signal{}
+				if tc.getOpStatus != nil {
+					s.ReportError = tc.getOpStatus.reportError
+					s.ReportOpWithErrorStatus = tc.getOpStatus.reportOpWithErrorStatus
+					s.ReportRunning = tc.getOpStatus.reportRunning
+				}
+				execute := <-opUnblocker
+				execute <- s
+			}
+
+			// Inject mock response
+			if tc.signalIsOpDone {
+				s := file.Signal{}
+				if tc.isOpDoneStatus != nil {
+					s.ReportError = tc.isOpDoneStatus.reportError
+					s.ReportOpWithErrorStatus = tc.isOpDoneStatus.reportOpWithErrorStatus
+					s.ReportRunning = tc.isOpDoneStatus.reportRunning
+				}
+				execute := <-opUnblocker
+				execute <- s
+			}
+
+			// Verify response
+			response := <-respChannel
+			if response.verified != tc.expectedVerificationStaus {
+				t.Errorf("verify status want %v, got %v", tc.expectedVerificationStaus, response.verified)
+			}
+
+			if tc.errorExpected && response.err == nil {
+				t.Errorf("expected error not found")
+			}
+			if !tc.errorExpected && response.err != nil {
+				t.Errorf("unexpected error found")
+			}
+
+			// Verify cache content
+			instanceOp := manager.cache.GetInstanceOp(tc.scKey, tc.instanceKey)
+			if tc.instanceKeyExpectedInCache && instanceOp == nil {
+				t.Errorf("expcted instance key not found")
+			}
+			if !tc.instanceKeyExpectedInCache && instanceOp != nil {
+				t.Errorf("unexpcted instance key found")
+			}
+			if tc.emptyOpExpected && instanceOp.Name != "" {
+				t.Errorf("expcted empty op")
+			}
+		})
+	}
+}
+
+func TestVerifyNoRunningShareOp(t *testing.T) {
+	tests := []struct {
+		name                         string
+		scKey                        string
+		share                        *file.Share
+		initShareCreateOpMap         []Item
+		initShareOpMap               []Item
+		signalGetOpForShareCreate    bool
+		signalIsOpDoneForShareCreate bool
+		getOpStatusforShareCreate    *MockOpStatus
+		isOpDoneStatusForShareCreate *MockOpStatus
+		signalGetOpForShareOp        bool
+		signalIsOpDoneForShareOp     bool
+		getOpStatusforShareOp        *MockOpStatus
+		isOpDoneStatusForShareOp     *MockOpStatus
+
+		errorExpected             bool
+		expectedVerificationStaus bool
+	}{
+		{
+			name:  "tc1 - no share create op in cache, no share op in cache",
+			scKey: testInstanceScPrefix,
+			share: &file.Share{
+				Name: testShareName,
+				Parent: &file.MultishareInstance{
+					Project:  testProject,
+					Location: testLocation,
+					Name:     testInstanceName,
+				},
+			},
+			expectedVerificationStaus: true,
+		},
+		{
+			name:  "tc2 - check for share create op returns error, status not verified",
+			scKey: testInstanceScPrefix,
+			share: &file.Share{
+				Name: testShareName,
+				Parent: &file.MultishareInstance{
+					Project:  testProject,
+					Location: testLocation,
+					Name:     testInstanceName,
+				},
+			},
+			initShareCreateOpMap: []Item{
+				{
+					scKey:          testInstanceScPrefix,
+					shareCreateKey: testShareName,
+					createOp: util.ShareCreateOpInfo{
+						InstanceHandle: testInstanceHandle,
+						OpName:         "op-1",
+					},
+				},
+			},
+			signalGetOpForShareCreate: true,
+			getOpStatusforShareCreate: &MockOpStatus{
+				reportError: true,
+			},
+			expectedVerificationStaus: false,
+			errorExpected:             true,
+		},
+		{
+			name:  "tc3 - check for share create op returns a stale op status complete in isOpDone, check for share op returns error in get op, status not verified",
+			scKey: testInstanceScPrefix,
+			share: &file.Share{
+				Name: testShareName,
+				Parent: &file.MultishareInstance{
+					Project:  testProject,
+					Location: testRegion,
+					Name:     testInstanceName,
+				},
+			},
+			initShareCreateOpMap: []Item{
+				{
+					scKey:          testInstanceScPrefix,
+					shareCreateKey: testShareName,
+					createOp: util.ShareCreateOpInfo{
+						InstanceHandle: testInstanceHandle,
+						OpName:         "op-1",
+					},
+				},
+			},
+			initShareOpMap: []Item{
+				{
+					scKey:    testInstanceScPrefix,
+					shareKey: util.CreateShareKey(testProject, testRegion, testInstanceName, testShareName),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.ShareDelete,
+					},
+				},
+			},
+			signalGetOpForShareCreate:    true,
+			signalIsOpDoneForShareCreate: true,
+			signalGetOpForShareOp:        true,
+			getOpStatusforShareOp: &MockOpStatus{
+				reportError: true,
+			},
+			expectedVerificationStaus: false,
+			errorExpected:             true,
+		},
+		{
+			name:  "tc4 - check for share create op returns nil, check for share op returns op fialed, status verified",
+			scKey: testInstanceScPrefix,
+			share: &file.Share{
+				Name: testShareName,
+				Parent: &file.MultishareInstance{
+					Project:  testProject,
+					Location: testRegion,
+					Name:     testInstanceName,
+				},
+			},
+			initShareOpMap: []Item{
+				{
+					scKey:    testInstanceScPrefix,
+					shareKey: util.CreateShareKey(testProject, testRegion, testInstanceName, testShareName),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.ShareDelete,
+					},
+				},
+			},
+			signalGetOpForShareOp:    true,
+			signalIsOpDoneForShareOp: true,
+			isOpDoneStatusForShareOp: &MockOpStatus{
+				reportError: true,
+			},
+			expectedVerificationStaus: true,
+			errorExpected:             false,
+		},
+		{
+			name:  "tc4 - check for share create op returns nil, check for share op returns op success, status verified",
+			scKey: testInstanceScPrefix,
+			share: &file.Share{
+				Name: testShareName,
+				Parent: &file.MultishareInstance{
+					Project:  testProject,
+					Location: testRegion,
+					Name:     testInstanceName,
+				},
+			},
+			initShareOpMap: []Item{
+				{
+					scKey:    testInstanceScPrefix,
+					shareKey: util.CreateShareKey(testProject, testRegion, testInstanceName, testShareName),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.ShareDelete,
+					},
+				},
+			},
+			signalGetOpForShareOp:     true,
+			signalIsOpDoneForShareOp:  true,
+			expectedVerificationStaus: true,
+			errorExpected:             false,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			opUnblocker := make(chan chan file.Signal)
+			cloudProvider := initCloudProviderWithBlockingFileService(t, opUnblocker)
+			manager := NewMultishareOpsManager(cloudProvider)
+
+			runRequest := func(ctx context.Context, instanceSCPrefix string, share *file.Share) <-chan Response {
+				responseChannel := make(chan Response)
+				go func() {
+					status, err := manager.verifyNoRunningShareOp(ctx, instanceSCPrefix, share)
+					responseChannel <- Response{
+						verified: status,
+						err:      err,
+					}
+				}()
+				return responseChannel
+			}
+			// Pre-populate the share create op map
+			for _, v := range tc.initShareCreateOpMap {
+				if _, ok := manager.cache.ScInfoMap[v.scKey]; !ok {
+					manager.cache.ScInfoMap[v.scKey] = util.StorageClassInfo{
+						InstanceMap:    make(util.InstanceMap),
+						ShareCreateMap: make(util.ShareCreateMap),
+						ShareOpsMap:    make(util.ShareOpsMap),
+					}
+				}
+				manager.cache.AddShareCreateOp(v.scKey, v.shareCreateKey, v.createOp)
+			}
+
+			// Pre-populate the share op map
+			for _, v := range tc.initShareOpMap {
+				if _, ok := manager.cache.ScInfoMap[v.scKey]; !ok {
+					manager.cache.ScInfoMap[v.scKey] = util.StorageClassInfo{
+						InstanceMap:    make(util.InstanceMap),
+						ShareCreateMap: make(util.ShareCreateMap),
+						ShareOpsMap:    make(util.ShareOpsMap),
+					}
+				}
+				manager.cache.AddShareOp(v.scKey, v.shareKey, v.op)
+			}
+
+			respChannel := runRequest(context.Background(), tc.scKey, tc.share)
+
+			// Inject mock response
+			if tc.signalGetOpForShareCreate {
+				s := file.Signal{}
+				if tc.getOpStatusforShareCreate != nil {
+					s.ReportError = tc.getOpStatusforShareCreate.reportError
+					s.ReportOpWithErrorStatus = tc.getOpStatusforShareCreate.reportOpWithErrorStatus
+					s.ReportRunning = tc.getOpStatusforShareCreate.reportRunning
+				}
+				execute := <-opUnblocker
+				execute <- s
+			}
+
+			// Inject mock response
+			if tc.signalIsOpDoneForShareCreate {
+				s := file.Signal{}
+				if tc.isOpDoneStatusForShareCreate != nil {
+					s.ReportError = tc.isOpDoneStatusForShareCreate.reportError
+					s.ReportOpWithErrorStatus = tc.isOpDoneStatusForShareCreate.reportOpWithErrorStatus
+					s.ReportRunning = tc.isOpDoneStatusForShareCreate.reportRunning
+				}
+				execute := <-opUnblocker
+				execute <- s
+			}
+
+			// Inject mock response
+			if tc.signalGetOpForShareOp {
+				s := file.Signal{}
+				if tc.getOpStatusforShareOp != nil {
+					s.ReportError = tc.getOpStatusforShareOp.reportError
+					s.ReportOpWithErrorStatus = tc.getOpStatusforShareOp.reportOpWithErrorStatus
+					s.ReportRunning = tc.getOpStatusforShareOp.reportRunning
+				}
+				execute := <-opUnblocker
+				execute <- s
+			}
+
+			// Inject mock response
+			if tc.signalIsOpDoneForShareOp {
+				s := file.Signal{}
+				if tc.isOpDoneStatusForShareOp != nil {
+					s.ReportError = tc.isOpDoneStatusForShareOp.reportError
+					s.ReportOpWithErrorStatus = tc.isOpDoneStatusForShareOp.reportOpWithErrorStatus
+					s.ReportRunning = tc.isOpDoneStatusForShareOp.reportRunning
+				}
+				execute := <-opUnblocker
+				execute <- s
+			}
+			// Verify response
+			response := <-respChannel
+			if response.verified != tc.expectedVerificationStaus {
+				t.Errorf("op status want %v, got %v", tc.expectedVerificationStaus, response.verified)
+			}
+			if response.err != nil && !tc.errorExpected {
+				t.Errorf("got unexpected error")
+			}
+			if response.err == nil && tc.errorExpected {
+				t.Errorf("expected error")
+			}
+		})
+	}
+}
+
+func TestRunEligibleInstanceCheck(t *testing.T) {
+	tests := []struct {
+		name                              string
+		scKey                             string
+		initInstance                      []file.MultishareInstance
+		initShares                        []file.Share
+		initInstanceOpMap                 []Item
+		numSignalGetOpForInstance         int
+		numsignalIsOpDoneForInstance      int
+		isOpDoneStatusForInstance         []MockOpStatus
+		numSignalGetInstance              int
+		reportErrorForGetInstance         []bool
+		reportNotFoundErrorForGetInstance []bool
+		expectedNumReadyInstances         int
+		expectedNumNonReadyInstances      int
+		expectedError                     bool
+	}{
+		{
+			name:  "tc1-empty instance map",
+			scKey: testInstanceScPrefix,
+		},
+		{
+			name:  "single instance in map, failed to parse handle",
+			scKey: testInstanceScPrefix,
+			initInstanceOpMap: []Item{
+				{
+					instanceKey: util.InstanceKey("blah"),
+					op:          util.OpInfo{},
+				},
+			},
+		},
+		{
+			name:  "tc2-single ready instance in map, GET failed",
+			scKey: testInstanceScPrefix,
+			initInstanceOpMap: []Item{
+				{
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName),
+					op:          util.OpInfo{},
+				},
+			},
+			numSignalGetInstance:              1,
+			reportErrorForGetInstance:         []bool{false},
+			reportNotFoundErrorForGetInstance: []bool{false},
+		},
+		{
+			name:  "tc3-single ready instance with 0 shares in map, GET success",
+			scKey: testInstanceScPrefix,
+			initInstanceOpMap: []Item{
+				{
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName),
+					op:          util.OpInfo{},
+				},
+			},
+			initInstance: []file.MultishareInstance{
+				{
+					Name:     testInstanceName,
+					Project:  testProject,
+					Location: testRegion,
+				},
+			},
+			numSignalGetInstance:              1,
+			reportErrorForGetInstance:         []bool{false},
+			reportNotFoundErrorForGetInstance: []bool{false},
+			expectedNumReadyInstances:         1,
+		},
+		{
+			name:  "tc4-two ready instance (no op in map) with 0 shares in map, GET success",
+			scKey: testInstanceScPrefix,
+			initInstanceOpMap: []Item{
+				{
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName+"1"),
+					op:          util.OpInfo{},
+				},
+				{
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName+"2"),
+					op:          util.OpInfo{},
+				},
+			},
+			initInstance: []file.MultishareInstance{
+				{
+					Name:     testInstanceName + "1",
+					Project:  testProject,
+					Location: testRegion,
+				},
+				{
+					Name:     testInstanceName + "2",
+					Project:  testProject,
+					Location: testRegion,
+				},
+			},
+			numSignalGetInstance:              2,
+			reportErrorForGetInstance:         []bool{false, false},
+			reportNotFoundErrorForGetInstance: []bool{false, false},
+			expectedNumReadyInstances:         2,
+		},
+		{
+			name:  "tc5-two ready instance (op in map, that is complete) with 0 shares in map, GET success",
+			scKey: testInstanceScPrefix,
+			initInstanceOpMap: []Item{
+				{
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName+"1"),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.InstanceCreate,
+					},
+				},
+				{
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName+"2"),
+					op: util.OpInfo{
+						Name: "op-2",
+						Type: util.InstanceCreate,
+					},
+				},
+			},
+			initInstance: []file.MultishareInstance{
+				{
+					Name:     testInstanceName + "1",
+					Project:  testProject,
+					Location: testRegion,
+				},
+				{
+					Name:     testInstanceName + "2",
+					Project:  testProject,
+					Location: testRegion,
+				},
+			},
+			numSignalGetOpForInstance:    2,
+			numsignalIsOpDoneForInstance: 2,
+			isOpDoneStatusForInstance: []MockOpStatus{
+				{
+					reportRunning:           false,
+					reportNotFoundError:     false,
+					reportOpWithErrorStatus: false,
+				},
+				{
+					reportRunning:           false,
+					reportNotFoundError:     false,
+					reportOpWithErrorStatus: false,
+				},
+			},
+			numSignalGetInstance:              2,
+			reportErrorForGetInstance:         []bool{false, false},
+			reportNotFoundErrorForGetInstance: []bool{false, false},
+			expectedNumReadyInstances:         2,
+		},
+		{
+			name:  "tc6-two instance (delete op in map, that is complete) with 0 shares in map",
+			scKey: testInstanceScPrefix,
+			initInstanceOpMap: []Item{
+				{
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName+"1"),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.InstanceDelete,
+					},
+				},
+				{
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName+"2"),
+					op: util.OpInfo{
+						Name: "op-2",
+						Type: util.InstanceDelete,
+					},
+				},
+			},
+			initInstance: []file.MultishareInstance{
+				{
+					Name:     testInstanceName + "1",
+					Project:  testProject,
+					Location: testRegion,
+				},
+				{
+					Name:     testInstanceName + "2",
+					Project:  testProject,
+					Location: testRegion,
+				},
+			},
+			numSignalGetOpForInstance:    2,
+			numsignalIsOpDoneForInstance: 2,
+			isOpDoneStatusForInstance: []MockOpStatus{
+				{
+					reportRunning:           false,
+					reportNotFoundError:     false,
+					reportOpWithErrorStatus: false,
+				},
+				{
+					reportRunning:           false,
+					reportNotFoundError:     false,
+					reportOpWithErrorStatus: false,
+				},
+			},
+			numSignalGetInstance:              2,
+			reportErrorForGetInstance:         []bool{false, false},
+			reportNotFoundErrorForGetInstance: []bool{true, true},
+		},
+		{
+			name:  "tc7-two instance (one ready, one not ready) with 0 shares in map",
+			scKey: testInstanceScPrefix,
+			initInstanceOpMap: []Item{
+				{
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName+"1"),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.InstanceCreate,
+					},
+				},
+				{
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName+"2"),
+					op: util.OpInfo{
+						Name: "op-2",
+						Type: util.InstanceExpand,
+					},
+				},
+			},
+			initInstance: []file.MultishareInstance{
+				{
+					Name:     testInstanceName + "1",
+					Project:  testProject,
+					Location: testRegion,
+				},
+				{
+					Name:     testInstanceName + "2",
+					Project:  testProject,
+					Location: testRegion,
+				},
+			},
+			numSignalGetOpForInstance:    2,
+			numsignalIsOpDoneForInstance: 2,
+			isOpDoneStatusForInstance: []MockOpStatus{
+				{
+					reportRunning:           false,
+					reportNotFoundError:     false,
+					reportOpWithErrorStatus: false,
+				},
+				{
+					reportRunning:           true,
+					reportNotFoundError:     false,
+					reportOpWithErrorStatus: false,
+				},
+			},
+			numSignalGetInstance:              1,
+			reportErrorForGetInstance:         []bool{false},
+			reportNotFoundErrorForGetInstance: []bool{false},
+			expectedNumNonReadyInstances:      1,
+			expectedNumReadyInstances:         1,
+		},
+		{
+			name:  "tc8-one instance (delete op in progress) with 0 shares in map",
+			scKey: testInstanceScPrefix,
+			initInstanceOpMap: []Item{
+				{
+					instanceKey: util.CreateInstanceKey(testProject, testRegion, testInstanceName+"1"),
+					op: util.OpInfo{
+						Name: "op-1",
+						Type: util.InstanceDelete,
+					},
+				},
+			},
+			initInstance: []file.MultishareInstance{
+				{
+					Name:     testInstanceName + "1",
+					Project:  testProject,
+					Location: testRegion,
+				},
+			},
+			numSignalGetOpForInstance:    1,
+			numsignalIsOpDoneForInstance: 1,
+			isOpDoneStatusForInstance: []MockOpStatus{
+				{
+					reportRunning:           true,
+					reportNotFoundError:     false,
+					reportOpWithErrorStatus: false,
+				},
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			opUnblocker := make(chan chan file.Signal, 1)
+			cloudProvider := initCloudProviderWithBlockingFileService(t, opUnblocker)
+			manager := NewMultishareOpsManager(cloudProvider)
+
+			runRequest := func(ctx context.Context, instanceSCPrefix string) <-chan Response {
+				responseChannel := make(chan Response)
+				go func() {
+					ready, numNonReady, err := manager.runEligibleInstanceCheck(ctx, instanceSCPrefix)
+					responseChannel <- Response{
+						readyInstances:       ready,
+						numNonReadyInstances: numNonReady,
+						err:                  err,
+					}
+				}()
+				return responseChannel
+			}
+			// Prepopulate known instances and shares
+			for _, instance := range tc.initInstance {
+				manager.cloud.File.StartCreateMultishareInstanceOp(context.Background(), &instance)
+			}
+			for _, share := range tc.initShares {
+				manager.cloud.File.StartCreateShareOp(context.Background(), &share)
+			}
+			for _, item := range tc.initInstanceOpMap {
+				manager.cache.AddInstanceOp(tc.scKey, item.instanceKey, item.op)
+			}
+
+			respChannel := runRequest(context.Background(), tc.scKey)
+			// Inject mock response for GetOp
+			for i := 0; i < tc.numSignalGetOpForInstance; i++ {
+				s := file.Signal{}
+				execute := <-opUnblocker
+				execute <- s
+			}
+
+			// Inject mock response for IsOpDone
+			for i := 0; i < tc.numsignalIsOpDoneForInstance; i++ {
+				s := file.Signal{}
+				s.ReportError = tc.isOpDoneStatusForInstance[i].reportError
+				s.ReportOpWithErrorStatus = tc.isOpDoneStatusForInstance[i].reportOpWithErrorStatus
+				s.ReportRunning = tc.isOpDoneStatusForInstance[i].reportRunning
+				execute := <-opUnblocker
+				execute <- s
+			}
+
+			for i := 0; i < tc.numSignalGetInstance; i++ {
+				s := file.Signal{}
+				s.ReportError = tc.reportErrorForGetInstance[i]
+				s.ReportNotFoundError = tc.reportNotFoundErrorForGetInstance[i]
+				execute := <-opUnblocker
+				execute <- s
+			}
+			// Verify response
+			response := <-respChannel
+			if response.numNonReadyInstances != tc.expectedNumNonReadyInstances {
+				t.Errorf("want %v, got %v", tc.expectedNumNonReadyInstances, response.numNonReadyInstances)
+			}
+			if len(response.readyInstances) != tc.expectedNumReadyInstances {
+				t.Errorf("unexpected instance %v", response.readyInstances)
+			}
+			if !tc.expectedError && response.err != nil {
+				t.Errorf("unexpected error")
+			}
+			if tc.expectedError && response.err == nil {
+				t.Errorf("expecteded error")
+			}
+		})
+	}
+}
+
+func TestInstanceNeedsExpand(t *testing.T) {
+	tests := []struct {
+		name                    string
+		scKey                   string
+		initShares              []file.Share
+		targetShareToAccomodate *file.Share
+		expectedNeedsExpand     bool
+		targetBytes             int64
+		expectError             bool
+	}{
+		{
+			name:  "0 shares in 1 T instance,  new 100G share",
+			scKey: testInstanceScPrefix,
+			targetShareToAccomodate: &file.Share{
+				Name:          testShareName,
+				CapacityBytes: 100 * util.Gb,
+				Parent: &file.MultishareInstance{
+					Project:       testProject,
+					Location:      testRegion,
+					Name:          testInstanceName,
+					CapacityBytes: 1 * util.Tb,
+				},
+			},
+		},
+		{
+			name:  "1 existing 100G share in 1 T instance,  new 100G share",
+			scKey: testInstanceScPrefix,
+			initShares: []file.Share{
+				{
+					Name:          testShareName + "1",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+			},
+			targetShareToAccomodate: &file.Share{
+				Name:          testShareName + "2",
+				CapacityBytes: 100 * util.Gb,
+				Parent: &file.MultishareInstance{
+					Project:       testProject,
+					Location:      testRegion,
+					Name:          testInstanceName,
+					CapacityBytes: 1 * util.Tb,
+				},
+			},
+		},
+		{
+			name:  "9 existing 100G share in 1 T instance, new 100G share",
+			scKey: testInstanceScPrefix,
+			initShares: []file.Share{
+				{
+					Name:          testShareName + "1",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "2",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "3",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "4",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "5",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "6",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "7",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "8",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "9",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+			},
+			targetShareToAccomodate: &file.Share{
+				Name:          testShareName + "10",
+				CapacityBytes: 100 * util.Gb,
+				Parent: &file.MultishareInstance{
+					Project:       testProject,
+					Location:      testRegion,
+					Name:          testInstanceName,
+					CapacityBytes: 1 * util.Tb,
+				},
+			},
+		},
+		{
+			name:  "1 existing 100G share in 1 T instance,  new 1T share",
+			scKey: testInstanceScPrefix,
+			initShares: []file.Share{
+				{
+					Name:          testShareName + "1",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+			},
+			targetShareToAccomodate: &file.Share{
+				Name:          testShareName + "2",
+				CapacityBytes: 1 * util.Tb,
+				Parent: &file.MultishareInstance{
+					Project:       testProject,
+					Location:      testRegion,
+					Name:          testInstanceName,
+					CapacityBytes: 1 * util.Tb,
+				},
+			},
+			expectedNeedsExpand: true,
+			targetBytes:         1*util.Tb + (1*util.Tb - (1*util.Tb - 100*util.Gb)),
+		},
+		{
+			name:  "2 existing 100G share in 1 T instance,  new 900G share",
+			scKey: testInstanceScPrefix,
+			initShares: []file.Share{
+				{
+					Name:          testShareName + "1",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "2",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "3",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "4",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "5",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "6",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "7",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "8",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "9",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+			},
+			targetShareToAccomodate: &file.Share{
+				Name:          testShareName + "10",
+				CapacityBytes: 1 * util.Tb,
+				Parent: &file.MultishareInstance{
+					Project:       testProject,
+					Location:      testRegion,
+					Name:          testInstanceName,
+					CapacityBytes: 1 * util.Tb,
+				},
+			},
+			expectedNeedsExpand: true,
+			targetBytes:         1*util.Tb + (1*util.Tb - (1*util.Tb - 9*100*util.Gb)),
+		},
+		{
+			name:  "9 existing 100G share in 1 T instance,  new 1T share",
+			scKey: testInstanceScPrefix,
+			initShares: []file.Share{
+				{
+					Name:          testShareName + "1",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+				{
+					Name:          testShareName + "2",
+					CapacityBytes: 100 * util.Gb,
+					Parent: &file.MultishareInstance{
+						Project:       testProject,
+						Location:      testRegion,
+						Name:          testInstanceName,
+						CapacityBytes: 1 * util.Tb,
+					},
+				},
+			},
+			targetShareToAccomodate: &file.Share{
+				Name:          testShareName + "3",
+				CapacityBytes: 900 * util.Gb,
+				Parent: &file.MultishareInstance{
+					Project:       testProject,
+					Location:      testRegion,
+					Name:          testInstanceName,
+					CapacityBytes: 1 * util.Tb,
+				},
+			},
+			expectedNeedsExpand: true,
+			targetBytes:         1*util.Tb + (900*util.Gb - (1*util.Tb - 2*100*util.Gb)),
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			opUnblocker := make(chan chan file.Signal, 1)
+			cloudProvider := initCloudProviderWithBlockingFileService(t, opUnblocker)
+			manager := NewMultishareOpsManager(cloudProvider)
+
+			runRequest := func(ctx context.Context, share *file.Share) <-chan Response {
+				responseChannel := make(chan Response)
+				go func() {
+					needsExpand, targetBytes, err := manager.instanceNeedsExpand(context.Background(), share)
+					responseChannel <- Response{
+						instanceNeedsExpand: needsExpand,
+						targetBytes:         targetBytes,
+						err:                 err,
+					}
+				}()
+				return responseChannel
+			}
+
+			for _, share := range tc.initShares {
+				if share.Parent != nil {
+					manager.cloud.File.StartCreateMultishareInstanceOp(context.Background(), share.Parent)
+				}
+				manager.cloud.File.StartCreateShareOp(context.Background(), &share)
+			}
+
+			respChannel := runRequest(context.Background(), tc.targetShareToAccomodate)
+			response := <-respChannel
+			if tc.expectError && response.err == nil {
+				t.Errorf("expected error")
+			}
+			if !tc.expectError && response.err != nil {
+				t.Errorf("unexpectded error")
+			}
+			if tc.expectedNeedsExpand != response.instanceNeedsExpand {
+				t.Errorf("want %v, got %v", tc.expectedNeedsExpand, response.instanceNeedsExpand)
+			}
+			if tc.targetBytes != response.targetBytes {
+				t.Errorf("want %v, got %v", tc.targetBytes, response.targetBytes)
+			}
+		})
+	}
+}
diff --git a/pkg/util/cache_test.go b/pkg/util/cache_test.go
new file mode 100644
index 00000000..568cde96
--- /dev/null
+++ b/pkg/util/cache_test.go
@@ -0,0 +1,1750 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package util
+
+import (
+	"fmt"
+	"testing"
+)
+
+const (
+	testProject  = "test-project"
+	testLocation = "us-central1"
+	testInstance = "testInstance"
+	testShare    = "testShane"
+)
+
+type Item struct {
+	scKey          string
+	instanceKey    InstanceKey
+	shareKey       ShareKey
+	shareCreateKey string
+	op             OpInfo
+	createOp       ShareCreateOpInfo
+}
+
+type AddItem struct {
+	item          Item
+	skipInitSCKey bool
+	expectError   bool
+}
+
+type ClearItem struct {
+	item        Item
+	clearValue  bool
+	clearKey    bool
+	expectError bool
+}
+type OutputItem struct {
+	item             Item
+	expectEmpty      bool
+	expectEmptyValue bool
+}
+
+func TestInstanceMap(t *testing.T) {
+	tests := []struct {
+		name   string
+		add    []Item
+		clear  []ClearItem
+		output []OutputItem
+	}{
+		{
+			name: "tc1 - add ops for 1 storage class",
+			add: []Item{
+				{
+					scKey:       "sc-1",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+					op: OpInfo{
+						Name: "op-1",
+						Type: InstanceCreate,
+					},
+				},
+				{
+					scKey:       "sc-1",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+					op: OpInfo{
+						Name: "op-2",
+						Type: InstanceExpand,
+					},
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:       "sc-1",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+						op: OpInfo{
+							Name: "op-1",
+							Type: InstanceCreate,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:       "sc-1",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc2 - add ops for 2 storage classes",
+			add: []Item{
+				{
+					scKey:       "sc-1",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+					op: OpInfo{
+						Name: "op-1",
+						Type: InstanceCreate,
+					},
+				},
+				{
+					scKey:       "sc-2",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+					op: OpInfo{
+						Name: "op-2",
+						Type: InstanceExpand,
+					},
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:       "sc-1",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+						op: OpInfo{
+							Name: "op-1",
+							Type: InstanceCreate,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc3 - add 2 ops, clear the ops for the instances",
+			add: []Item{
+				{
+					scKey:       "sc-1",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+					op: OpInfo{
+						Name: "op-1",
+						Type: InstanceCreate,
+					},
+				},
+				{
+					scKey:       "sc-2",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+					op: OpInfo{
+						Name: "op-2",
+						Type: InstanceExpand,
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:       "sc-1",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+						op: OpInfo{
+							Name: "op-1",
+							Type: InstanceCreate,
+						},
+					},
+					clearValue: true,
+				},
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+					clearValue: true,
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:       "sc-1",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+					},
+					expectEmptyValue: true,
+				},
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+					},
+					expectEmptyValue: true,
+				},
+			},
+		},
+		{
+			name: "tc4 - add 2 ops, clear op value of one instance",
+			add: []Item{
+				{
+					scKey:       "sc-1",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+					op: OpInfo{
+						Name: "op-1",
+						Type: InstanceCreate,
+					},
+				},
+				{
+					scKey:       "sc-2",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+					op: OpInfo{
+						Name: "op-2",
+						Type: InstanceExpand,
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+					clearValue: true,
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:       "sc-1",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+						op: OpInfo{
+							Name: "op-1",
+							Type: InstanceCreate,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+					expectEmptyValue: true,
+				},
+			},
+		},
+		{
+			name: "tc5 - clear op attempt for non-existent sc",
+			add: []Item{
+				{
+					scKey:       "sc-1",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+					op: OpInfo{
+						Name: "op-1",
+						Type: InstanceCreate,
+					},
+				},
+				{
+					scKey:       "sc-2",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+					op: OpInfo{
+						Name: "op-2",
+						Type: InstanceExpand,
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:       "sc-3",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+					clearValue: true,
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:       "sc-1",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+						op: OpInfo{
+							Name: "op-1",
+							Type: InstanceCreate,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc6 - clear op attempt for existing sc, error expected in clear",
+			add: []Item{
+				{
+					scKey:       "sc-1",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+					op: OpInfo{
+						Name: "op-1",
+						Type: InstanceCreate,
+					},
+				},
+				{
+					scKey:       "sc-2",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+					op: OpInfo{
+						Name: "op-2",
+						Type: InstanceExpand,
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-3",
+							Type: InstanceExpand,
+						},
+					},
+					expectError: true,
+					clearValue:  true,
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:       "sc-1",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+						op: OpInfo{
+							Name: "op-1",
+							Type: InstanceCreate,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc7 - clear non-existent instance key for existing sc",
+			add: []Item{
+				{
+					scKey:       "sc-1",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+					op: OpInfo{
+						Name: "op-1",
+						Type: InstanceCreate,
+					},
+				},
+				{
+					scKey:       "sc-2",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+					op: OpInfo{
+						Name: "op-2",
+						Type: InstanceExpand,
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"3")),
+						op: OpInfo{
+							Name: "op-3",
+							Type: InstanceExpand,
+						},
+					},
+					clearValue: true,
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:       "sc-1",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+						op: OpInfo{
+							Name: "op-1",
+							Type: InstanceCreate,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"3")),
+					},
+					expectEmpty: true,
+				},
+			},
+		},
+		{
+			name: "tc8 - add 2 ops, clear the keys for the instances",
+			add: []Item{
+				{
+					scKey:       "sc-1",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+					op: OpInfo{
+						Name: "op-1",
+						Type: InstanceCreate,
+					},
+				},
+				{
+					scKey:       "sc-2",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+					op: OpInfo{
+						Name: "op-2",
+						Type: InstanceExpand,
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:       "sc-1",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+						op: OpInfo{
+							Name: "op-1",
+							Type: InstanceCreate,
+						},
+					},
+					clearKey: true,
+				},
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+					clearKey: true,
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:       "sc-1",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+					},
+					expectEmpty: true,
+				},
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+					},
+					expectEmpty: true,
+				},
+			},
+		},
+		{
+			name: "tc9 - add 2 ops, clear key of one instance",
+			add: []Item{
+				{
+					scKey:       "sc-1",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+					op: OpInfo{
+						Name: "op-1",
+						Type: InstanceCreate,
+					},
+				},
+				{
+					scKey:       "sc-2",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+					op: OpInfo{
+						Name: "op-2",
+						Type: InstanceExpand,
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+					clearKey: true,
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:       "sc-1",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+						op: OpInfo{
+							Name: "op-1",
+							Type: InstanceCreate,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+					expectEmpty: true,
+				},
+			},
+		},
+		{
+			name: "tc10 - clear key attempt for non-existent sc",
+			add: []Item{
+				{
+					scKey:       "sc-1",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+					op: OpInfo{
+						Name: "op-1",
+						Type: InstanceCreate,
+					},
+				},
+				{
+					scKey:       "sc-2",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+					op: OpInfo{
+						Name: "op-2",
+						Type: InstanceExpand,
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:       "sc-3",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+					clearKey: true,
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:       "sc-1",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+						op: OpInfo{
+							Name: "op-1",
+							Type: InstanceCreate,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc11 - clear non-existent instance key for existing sc",
+			add: []Item{
+				{
+					scKey:       "sc-1",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+					op: OpInfo{
+						Name: "op-1",
+						Type: InstanceCreate,
+					},
+				},
+				{
+					scKey:       "sc-2",
+					instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+					op: OpInfo{
+						Name: "op-2",
+						Type: InstanceExpand,
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"3")),
+						op: OpInfo{
+							Name: "op-3",
+							Type: InstanceExpand,
+						},
+					},
+					clearKey: true,
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:       "sc-1",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"1")),
+						op: OpInfo{
+							Name: "op-1",
+							Type: InstanceCreate,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"2")),
+						op: OpInfo{
+							Name: "op-2",
+							Type: InstanceExpand,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:       "sc-2",
+						instanceKey: InstanceKey(CreateInstanceKey(testProject, testLocation, testInstance+"3")),
+					},
+					expectEmpty: true,
+				},
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			cache := NewStorageClassInfoCache()
+			for _, item := range tc.add {
+				cache.AddInstanceOp(item.scKey, item.instanceKey, item.op)
+			}
+			for _, item := range tc.clear {
+				if item.clearValue {
+					err := cache.DeleteInstanceOp(item.item.scKey, item.item.instanceKey, item.item.op.Name)
+					if item.expectError && err == nil {
+						t.Errorf("expected error")
+					}
+					if !item.expectError && err != nil {
+						t.Errorf("unexpected error")
+					}
+				}
+
+				if item.clearKey {
+					cache.DeleteInstance(item.item.scKey, item.item.instanceKey)
+				}
+			}
+
+			for _, item := range tc.output {
+				op := cache.GetInstanceOp(item.item.scKey, item.item.instanceKey)
+				if item.expectEmpty {
+					if op != nil {
+						t.Errorf("want nil, got %+v", op)
+					}
+					continue
+				}
+
+				if item.expectEmptyValue {
+					if op.Name != "" || op.Type != UnknownOp {
+						t.Errorf("want empty op, got %+v", op)
+					}
+					continue
+				}
+
+				if op.Name != item.item.op.Name || op.Type != item.item.op.Type {
+					t.Errorf("want %+v, got : %+v", item.item.op, op)
+				}
+			}
+		})
+	}
+}
+
+func TestShareOpsMap(t *testing.T) {
+	tests := []struct {
+		name   string
+		add    []AddItem
+		clear  []ClearItem
+		output []OutputItem
+	}{
+		{
+			name: "tc1 - add ops for 2 shares on same instance, for 1 storage class",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance, testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareExpand,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance, testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareDelete,
+						},
+					},
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance, testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareExpand,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance, testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareDelete,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc2 - add 2 share ops on different instances, for 2 storage classes",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareDelete,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareExpand,
+						},
+					},
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareDelete,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareExpand,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc3 - add 2 ops, clear 2 ops",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareDelete,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareExpand,
+						},
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc4 - add 2 ops, clear 1 op",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareDelete,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareExpand,
+						},
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+						},
+					},
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareDelete,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+						},
+					},
+					expectEmpty: true,
+				},
+			},
+		},
+		{
+			name: "tc5 - clear non-existent op for non-existent sc",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareDelete,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareExpand,
+						},
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:    "sc-3",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-3",
+						},
+					},
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareDelete,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareExpand,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc6 - clear non-existent op for existing sc, error expected in clear",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareDelete,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareExpand,
+						},
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-3",
+						},
+					},
+					expectError: true,
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareDelete,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareExpand,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc7 - clear non-existent share key (different share) for existing sc",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareDelete,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareExpand,
+						},
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"3"),
+						op: OpInfo{
+							Name: "op-3",
+						},
+					},
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareDelete,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareExpand,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc7 - clear non-existent share key (different instance) for existing sc",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareDelete,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareExpand,
+						},
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"3", testShare+"2"),
+						op: OpInfo{
+							Name: "op-3",
+						},
+					},
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareDelete,
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareExpand,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc8 - non-existent sc key, add op failure",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:    "sc-1",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"1", testShare+"1"),
+						op: OpInfo{
+							Name: "op-1",
+							Type: ShareDelete,
+						},
+					},
+					skipInitSCKey: true,
+					expectError:   true,
+				},
+				{
+					item: Item{
+						scKey:    "sc-2",
+						shareKey: CreateShareKey(testProject, testLocation, testInstance+"2", testShare+"2"),
+						op: OpInfo{
+							Name: "op-2",
+							Type: ShareExpand,
+						},
+					},
+					skipInitSCKey: true,
+					expectError:   true,
+				},
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			cache := NewStorageClassInfoCache()
+			for _, item := range tc.add {
+				if !item.skipInitSCKey {
+					if _, ok := cache.ScInfoMap[item.item.scKey]; !ok {
+						cache.ScInfoMap[item.item.scKey] = StorageClassInfo{
+							InstanceMap:    make(InstanceMap),
+							ShareCreateMap: make(ShareCreateMap),
+							ShareOpsMap:    make(ShareOpsMap),
+						}
+					}
+				}
+				err := cache.AddShareOp(item.item.scKey, item.item.shareKey, item.item.op)
+				if !item.expectError && err != nil {
+					t.Errorf("unexpected error")
+				}
+				if item.expectError && err == nil {
+					t.Errorf("expected error not found")
+				}
+			}
+			for _, item := range tc.clear {
+				err := cache.DeleteShareOp(item.item.scKey, item.item.shareKey, item.item.op.Name)
+				if item.expectError && err == nil {
+					t.Errorf("expected error")
+				}
+				if !item.expectError && err != nil {
+					t.Errorf("unexpected error")
+				}
+			}
+
+			for _, item := range tc.output {
+				op := cache.GetShareOp(item.item.scKey, item.item.shareKey)
+				if item.expectEmpty {
+					if op != nil {
+						t.Errorf("want nil, got %+v", op)
+					}
+					return
+				}
+
+				if op.Name != item.item.op.Name || op.Type != item.item.op.Type {
+					t.Errorf("want %+v, got : %+v", item.item.op, op)
+				}
+			}
+		})
+	}
+
+}
+
+func createInstanceHandle(project, location, instance string) string {
+	return fmt.Sprintf("%s/%s/%s", project, location, instance)
+}
+
+func TestShareCreateOpMap(t *testing.T) {
+	tests := []struct {
+		name   string
+		add    []AddItem
+		clear  []ClearItem
+		output []OutputItem
+	}{
+		{
+			name: "tc1 - add ops for 2 shares on same instance, for 1 storage class",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance),
+							OpName:         "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance),
+							OpName:         "op-2",
+						},
+					},
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance),
+							OpName:         "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance),
+							OpName:         "op-2",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc2 - add 2 share ops on different instances, for 2 storage classes",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"1"),
+							OpName:         "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:          "sc-2",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"2"),
+							OpName:         "op-2",
+						},
+					},
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"1"),
+							OpName:         "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:          "sc-2",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"2"),
+							OpName:         "op-2",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc3 - add 2 share ops and 2 clear ops, for 2 storage classes",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"1"),
+							OpName:         "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:          "sc-2",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"2"),
+							OpName:         "op-2",
+						},
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							OpName: "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:          "sc-2",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							OpName: "op-2",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc4 - add 2 share ops and 1 clear op, for 2 storage classes",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"1"),
+							OpName:         "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:          "sc-2",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"2"),
+							OpName:         "op-2",
+						},
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:          "sc-2",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							OpName: "op-2",
+						},
+					},
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"1"),
+							OpName:         "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:          "sc-2",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"2"),
+							OpName:         "op-2",
+						},
+					},
+					expectEmpty: true,
+				},
+			},
+		},
+		{
+			name: "tc5 - clear non-existent sc",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"1"),
+							OpName:         "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:          "sc-2",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"2"),
+							OpName:         "op-2",
+						},
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:          "sc-3",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							OpName: "op-2",
+						},
+					},
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"1"),
+							OpName:         "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:          "sc-2",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"2"),
+							OpName:         "op-2",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc6 - clear non-existent share key",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"1"),
+							OpName:         "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:          "sc-2",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"2"),
+							OpName:         "op-2",
+						},
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "3",
+						createOp: ShareCreateOpInfo{
+							OpName: "op-3",
+						},
+					},
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"1"),
+							OpName:         "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:          "sc-2",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"2"),
+							OpName:         "op-2",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc7 - clear mistched op for share key, error expected in clear",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"1"),
+							OpName:         "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:          "sc-2",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"2"),
+							OpName:         "op-2",
+						},
+					},
+				},
+			},
+			clear: []ClearItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							OpName: "op-3",
+						},
+					},
+					expectError: true,
+				},
+			},
+			output: []OutputItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"1"),
+							OpName:         "op-1",
+						},
+					},
+				},
+				{
+					item: Item{
+						scKey:          "sc-2",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"2"),
+							OpName:         "op-2",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "tc8 - missing sc key, add op failure",
+			add: []AddItem{
+				{
+					item: Item{
+						scKey:          "sc-1",
+						shareCreateKey: testShare + "1",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"1"),
+							OpName:         "op-1",
+						},
+					},
+					skipInitSCKey: true,
+					expectError:   true,
+				},
+				{
+					item: Item{
+						scKey:          "sc-2",
+						shareCreateKey: testShare + "2",
+						createOp: ShareCreateOpInfo{
+							InstanceHandle: createInstanceHandle(testProject, testLocation, testInstance+"2"),
+							OpName:         "op-2",
+						},
+					},
+					skipInitSCKey: true,
+					expectError:   true,
+				},
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			cache := NewStorageClassInfoCache()
+			for _, item := range tc.add {
+				if !item.skipInitSCKey {
+					if _, ok := cache.ScInfoMap[item.item.scKey]; !ok {
+						cache.ScInfoMap[item.item.scKey] = StorageClassInfo{
+							InstanceMap:    make(InstanceMap),
+							ShareCreateMap: make(ShareCreateMap),
+							ShareOpsMap:    make(ShareOpsMap),
+						}
+					}
+				}
+				err := cache.AddShareCreateOp(item.item.scKey, item.item.shareCreateKey, item.item.createOp)
+				if !item.expectError && err != nil {
+					t.Errorf("unexpected error")
+				}
+				if item.expectError && err == nil {
+					t.Errorf("expected error not found")
+				}
+			}
+			for _, item := range tc.clear {
+				err := cache.DeleteShareCreateOp(item.item.scKey, item.item.shareCreateKey, item.item.createOp.OpName)
+				if item.expectError && err == nil {
+					t.Errorf("expected error")
+				}
+				if !item.expectError && err != nil {
+					t.Errorf("unexpected error")
+				}
+			}
+
+			for _, item := range tc.output {
+				op := cache.GetShareCreateOp(item.item.scKey, item.item.shareCreateKey)
+				if item.expectEmpty {
+					if op != nil {
+						t.Errorf("want nil, got %+v", op)
+					}
+					return
+				}
+
+				if op.OpName != item.item.createOp.OpName || op.InstanceHandle != item.item.createOp.InstanceHandle {
+					t.Errorf("want %+v, got : %+v", item.item.createOp, op)
+				}
+			}
+		})
+	}
+
+}
+
+func TestCreateKey(t *testing.T) {
+	tests := []struct {
+		name                string
+		project             string
+		location            string
+		instance            string
+		share               string
+		expectedShareKey    string
+		expectedInstanceKey string
+	}{
+		{
+			name:             "tc1",
+			project:          testProject,
+			location:         testLocation,
+			instance:         testInstance,
+			share:            testShare,
+			expectedShareKey: fmt.Sprintf("%s/%s/%s/%s", testProject, testLocation, testInstance, testShare),
+		},
+		{
+			name:                "tc2",
+			project:             testProject,
+			location:            testLocation,
+			instance:            testInstance,
+			expectedInstanceKey: fmt.Sprintf("%s/%s/%s", testProject, testLocation, testInstance),
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if tc.expectedShareKey != "" {
+				k := CreateShareKey(tc.project, tc.location, tc.instance, tc.share)
+				if string(k) != tc.expectedShareKey {
+					t.Errorf("want %v, got %v", tc.expectedShareKey, k)
+				}
+			}
+			if tc.expectedInstanceKey != "" {
+				k := CreateInstanceKey(tc.project, tc.location, tc.instance)
+				if string(k) != tc.expectedInstanceKey {
+					t.Errorf("want %v, got %v", tc.expectedInstanceKey, k)
+				}
+			}
+		})
+	}
+}
diff --git a/pkg/util/util_test.go b/pkg/util/util_test.go
index 9a993a92..cc558d22 100644
--- a/pkg/util/util_test.go
+++ b/pkg/util/util_test.go
@@ -18,7 +18,16 @@ package util
 
 import (
 	"reflect"
+	"strings"
 	"testing"
+
+	"github.com/google/uuid"
+)
+
+const (
+	testRegion       = "us-central1"
+	testInstanceName = "testInstance"
+	testShareName    = "testShare"
 )
 
 func TestRoundBytesToGb(t *testing.T) {
@@ -266,3 +275,221 @@ func TestConvertLabelsStringToMap(t *testing.T) {
 	})
 
 }
+
+func TestConvertVolToShareName(t *testing.T) {
+	testuuid := uuid.New().String()
+	tests := []struct {
+		name      string
+		volName   string
+		shareName string
+	}{
+		{
+			name:      "tc1 - all caps",
+			volName:   "PVC",
+			shareName: "pvc",
+		},
+		{
+			name:      "tc1 - caps and UUID",
+			volName:   "PVC-" + testuuid,
+			shareName: "pvc_" + strings.ReplaceAll(testuuid, "-", "_"),
+		},
+		{
+			name:      "tc1 - lower and UUID",
+			volName:   "pvc-" + testuuid,
+			shareName: "pvc_" + strings.ReplaceAll(testuuid, "-", "_"),
+		},
+		{
+			name:      "tc1 - caps and number",
+			volName:   "pvc-" + "123",
+			shareName: "pvc_" + "123",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			s := ConvertVolToShareName(tc.volName)
+			if s != tc.shareName {
+				t.Errorf("got %v, want %v", s, tc.shareName)
+			}
+		})
+	}
+
+}
+
+func TestParseInstanceHandle(t *testing.T) {
+	tests := []struct {
+		name           string
+		instanceHandle string
+		expectErr      bool
+		project        string
+		location       string
+		instancename   string
+	}{
+		{
+			name:           "invalid handle",
+			instanceHandle: "a/b/c/d",
+			expectErr:      true,
+		},
+		{
+			name:      "empty handle",
+			expectErr: true,
+		},
+		{
+			name:           "valid handle",
+			instanceHandle: testProject + "/" + testRegion + "/" + testInstanceName,
+			project:        testProject,
+			location:       testRegion,
+			instancename:   testInstanceName,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			p, l, n, err := ParseInstanceHandle(tc.instanceHandle)
+			if !tc.expectErr && err != nil {
+				t.Error("unexpected error")
+			}
+			if tc.expectErr && err == nil {
+				t.Error("expected error, got none")
+			}
+			if p != tc.project || l != tc.location || n != tc.instancename {
+				t.Errorf("mismatch")
+			}
+		})
+	}
+
+}
+
+func TestParseInstanceURI(t *testing.T) {
+	tests := []struct {
+		name         string
+		instanceuri  string
+		expectErr    bool
+		project      string
+		location     string
+		instancename string
+	}{
+		{
+			name:      "empty uri",
+			expectErr: true,
+		},
+		{
+			name:        "invalid uri 1",
+			instanceuri: "a/b/c/d",
+			expectErr:   true,
+		},
+		{
+			name:        "invalid uri",
+			instanceuri: testProject + "/" + testRegion + "/" + testInstanceName,
+			expectErr:   true,
+		},
+		{
+			name:         "invalid uri",
+			instanceuri:  "projects/" + testProject + "/locations/" + testRegion + "/instances/" + testInstanceName,
+			project:      testProject,
+			location:     testRegion,
+			instancename: testInstanceName,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			p, l, n, err := ParseInstanceURI(tc.instanceuri)
+			if !tc.expectErr && err != nil {
+				t.Error("unexpected error")
+			}
+			if tc.expectErr && err == nil {
+				t.Error("expected error, got none")
+			}
+			if p != tc.project || l != tc.location || n != tc.instancename {
+				t.Errorf("mismatch")
+			}
+		})
+	}
+}
+
+func TestParseShareHandle(t *testing.T) {
+	tests := []struct {
+		name         string
+		sharehandle  string
+		expectErr    bool
+		project      string
+		location     string
+		instancename string
+		sharename    string
+	}{
+		{
+			name:        "invalid handle",
+			sharehandle: "a/b/c/d/e",
+			expectErr:   true,
+		},
+		{
+			name:      "empty handle",
+			expectErr: true,
+		},
+		{
+			name:         "valid handle",
+			sharehandle:  testProject + "/" + testRegion + "/" + testInstanceName + "/" + testShareName,
+			project:      testProject,
+			location:     testRegion,
+			instancename: testInstanceName,
+			sharename:    testShareName,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			p, l, n, s, err := ParseShareHandle(tc.sharehandle)
+			if !tc.expectErr && err != nil {
+				t.Error("unexpected error")
+			}
+			if tc.expectErr && err == nil {
+				t.Error("expected error, got none")
+			}
+			if p != tc.project || l != tc.location || n != tc.instancename || s != tc.sharename {
+				t.Errorf("mismatch")
+			}
+		})
+	}
+}
+
+func TestParseShareURI(t *testing.T) {
+	tests := []struct {
+		name         string
+		shareuri     string
+		expectErr    bool
+		project      string
+		location     string
+		instancename string
+		sharename    string
+	}{
+		{
+			name:      "invalid uri",
+			shareuri:  "a/b/c/d/e",
+			expectErr: true,
+		},
+		{
+			name:      "empty uri",
+			expectErr: true,
+		},
+		{
+			name:         "valid uri",
+			shareuri:     "projects/" + testProject + "/locations/" + testRegion + "/instances/" + testInstanceName + "/shares/" + testShareName,
+			project:      testProject,
+			location:     testRegion,
+			instancename: testInstanceName,
+			sharename:    testShareName,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			p, l, n, s, err := ParseShareURI(tc.shareuri)
+			if !tc.expectErr && err != nil {
+				t.Error("unexpected error")
+			}
+			if tc.expectErr && err == nil {
+				t.Error("expected error, got none")
+			}
+			if p != tc.project || l != tc.location || n != tc.instancename || s != tc.sharename {
+				t.Errorf("mismatch")
+			}
+		})
+	}
+
+}