Bigtable: 'test_create_instance_w_two_clusters' flakes with '504 Deadline Exceeded' #5928

___________ TestInstanceAdminAPI.test_create_instance_w_two_clusters ___________

self = <tests.system.TestInstanceAdminAPI testMethod=test_create_instance_w_two_clusters>

    def test_create_instance_w_two_clusters(self):
        from google.cloud.bigtable import enums
        from google.cloud.bigtable.table import ClusterState
        _PRODUCTION = enums.Instance.Type.PRODUCTION
        ALT_INSTANCE_ID = 'dif' + unique_resource_id('-')
        instance = Config.CLIENT.instance(ALT_INSTANCE_ID,
                                          instance_type=_PRODUCTION,
                                          labels=LABELS)
    
        ALT_CLUSTER_ID_1 = ALT_INSTANCE_ID + '-c1'
        ALT_CLUSTER_ID_2 = ALT_INSTANCE_ID + '-c2'
        LOCATION_ID_2 = 'us-central1-f'
        STORAGE_TYPE = enums.StorageType.HDD
        cluster_1 = instance.cluster(
            ALT_CLUSTER_ID_1, location_id=LOCATION_ID, serve_nodes=SERVE_NODES,
            default_storage_type=STORAGE_TYPE)
        cluster_2 = instance.cluster(
            ALT_CLUSTER_ID_2, location_id=LOCATION_ID_2,
            serve_nodes=SERVE_NODES, default_storage_type=STORAGE_TYPE)
        operation = instance.create(clusters=[cluster_1, cluster_2])
        # We want to make sure the operation completes.
        operation.result(timeout=10)
    
        # Make sure this instance gets deleted after the test case.
        self.instances_to_delete.append(instance)
    
        # Create a new instance instance and make sure it is the same.
        instance_alt = Config.CLIENT.instance(ALT_INSTANCE_ID)
        instance_alt.reload()
    
        self.assertEqual(instance, instance_alt)
        self.assertEqual(instance.display_name, instance_alt.display_name)
        self.assertEqual(instance.type_, instance_alt.type_)
    
        clusters, failed_locations = instance_alt.list_clusters()
        self.assertEqual(failed_locations, [])
    
        clusters.sort(key=lambda x: x.name)
        alt_cluster_1, alt_cluster_2 = clusters
    
        self.assertEqual(cluster_1.location_id, alt_cluster_1.location_id)
        self.assertEqual(alt_cluster_1.state, enums.Cluster.State.READY)
        self.assertEqual(cluster_1.serve_nodes, alt_cluster_1.serve_nodes)
        self.assertEqual(cluster_1.default_storage_type,
                         alt_cluster_1.default_storage_type)
        self.assertEqual(cluster_2.location_id, alt_cluster_2.location_id)
        self.assertEqual(alt_cluster_2.state, enums.Cluster.State.READY)
        self.assertEqual(cluster_2.serve_nodes, alt_cluster_2.serve_nodes)
        self.assertEqual(cluster_2.default_storage_type,
                         alt_cluster_2.default_storage_type)
    
        # Test list clusters in project via 'client.list_clusters'
        clusters, failed_locations = Config.CLIENT.list_clusters()
        self.assertFalse(failed_locations)
        found = set([cluster.name for cluster in clusters])
        self.assertTrue({alt_cluster_1.name,
                         alt_cluster_2.name,
                         Config.CLUSTER.name}.issubset(found))
    
        temp_table_id = 'test-get-cluster-states'
        temp_table = instance.table(temp_table_id)
>       temp_table.create()

tests/system.py:280: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
google/cloud/bigtable/table.py:218: in create
    table=table, initial_splits=splits)
google/cloud/bigtable_admin_v2/gapic/bigtable_table_admin_client.py:327: in create_table
    request, retry=retry, timeout=timeout, metadata=metadata)
../api_core/google/api_core/gapic_v1/method.py:139: in __call__
    return wrapped_func(*args, **kwargs)
../api_core/google/api_core/retry.py:260: in retry_wrapped_func
    on_error=on_error,
../api_core/google/api_core/retry.py:177: in retry_target
    return target()
../api_core/google/api_core/timeout.py:206: in func_with_timeout
    return func(*args, **kwargs)
../api_core/google/api_core/grpc_helpers.py:61: in error_remapped_callable
    six.raise_from(exceptions.from_grpc_error(exc), exc)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

value = DeadlineExceeded('Deadline Exceeded',)
from_value = <_Rendezvous of RPC that terminated with:
	status = StatusCode.DEADLINE_EXCEED...all.cc","file_line":1099,"grpc_message":"Deadline Exceeded","grpc_status":4}"
>

    def raise_from(value, from_value):
>       raise value
E       DeadlineExceeded: 504 Deadline Exceeded

../.nox/sys-2-7/lib/python2.7/site-packages/six.py:737: DeadlineExceeded