fix: check if Bigtable cluster zones are affected by partial unavaila…

…bility

google/resource_bigtable_instance.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"strings"
 
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/customdiff"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
@@ -259,7 +260,17 @@ func resourceBigtableInstanceRead(d *schema.ResourceData, meta interface{}) erro
 
 	clusters, err := c.Clusters(ctx, instance.Name)
 	if err != nil {
-		return fmt.Errorf("Error retrieving instance clusters. %s", err)
+		partiallyUnavailableErr, ok := err.(bigtable.ErrPartiallyUnavailable)
+
+		if !ok {
+			return fmt.Errorf("Error retrieving instance clusters. %s", err)
+		}
+
+		unavailableClusterZones := getUnavailableClusterZones(d.Get("cluster").([]interface{}), partiallyUnavailableErr.Locations)
+
+		if len(unavailableClusterZones) > 0 {
+			return fmt.Errorf("Error retrieving instance clusters. The following zones are unavailable: %s", strings.Join(unavailableClusterZones, ", "))
+		}
 	}
 
 	clustersNewState := []map[string]interface{}{}
@@ -404,6 +415,23 @@ func flattenBigtableCluster(c *bigtable.ClusterInfo) map[string]interface{} {
 	return cluster
 }
 
+func getUnavailableClusterZones(clusters []interface{}, unavailableZones []string) []string {
+	var zones []string
+
+	for _, c := range clusters {
+		cluster := c.(map[string]interface{})
+		zone := cluster["zone"].(string)
+
+		for _, unavailableZone := range unavailableZones {
+			if zone == unavailableZone {
+				zones = append(zones, zone)
+				break
+			}
+		}
+	}
+	return zones
+}
+
 func expandBigtableClusters(clusters []interface{}, instanceID string, config *Config) ([]bigtable.ClusterConfig, error) {
 	results := make([]bigtable.ClusterConfig, 0, len(clusters))
 	for _, c := range clusters {