Fix new flaky postcommit python tests (#34638)

* Fix new flaky postcommit python test

* Formatting

* Change streaming wordcount pipeline's wait duration to 1200 seconds

Author: akashorabek

sdks/python/apache_beam/examples/streaming_wordcount_it_test.py

@@ -39,7 +39,7 @@
 OUTPUT_SUB = 'wc_subscription_output'
 
 DEFAULT_INPUT_NUMBERS = 500
-WAIT_UNTIL_FINISH_DURATION = 10 * 60 * 1000  # in milliseconds
+WAIT_UNTIL_FINISH_DURATION = 20 * 60 * 1000  # in milliseconds
 
 
 class StreamingWordCountIT(unittest.TestCase):

sdks/python/apache_beam/io/gcp/bigquery_read_it_test.py

@@ -729,6 +729,8 @@ def test_iobase_source(self):
 
 
 class ReadAllBQTests(BigQueryReadIntegrationTests):
+  TABLE_DATA_AVAILABILITY_WAIT_SECONDS = 30
+
   TABLE_DATA_1 = [{
       'number': 1, 'str': 'abc'
   }, {
@@ -789,6 +791,7 @@ def create_table(cls, table_name, data, table_schema):
     _ = cls.bigquery_client.get_table(cls.project, cls.dataset_id, table_name)
     cls.bigquery_client.insert_rows(
         cls.project, cls.dataset_id, table_name, data)
+    time.sleep(cls.TABLE_DATA_AVAILABILITY_WAIT_SECONDS)
     return table_schema
 
   @classmethod

sdks/python/apache_beam/io/gcp/bigquery_write_it_test.py

@@ -180,7 +180,7 @@ def test_big_query_write_schema_autodetect(self):
     if self.runner_name == 'TestDataflowRunner':
       self.skipTest('DataflowRunner does not support schema autodetection')
 
-    table_name = 'python_write_table'
+    table_name = 'python_write_table_schema_autodetect'
     table_id = '{}.{}'.format(self.dataset_id, table_name)
 
     input_data = [
@@ -383,7 +383,7 @@ def test_big_query_write_insert_errors_reporting(self):
     Test that errors returned by beam.io.WriteToBigQuery
     contain both the failed rows and the reason for it failing.
     """
-    table_name = 'python_write_table'
+    table_name = 'python_write_table_insert_errors'
     table_id = '{}.{}'.format(self.dataset_id, table_name)
 
     input_data = [{

sdks/python/apache_beam/io/gcp/gcsio_integration_test.py

@@ -59,8 +59,8 @@
 except ImportError:
   NotFound = None
 
-# Number of seconds to wait for bucket deletion to propagate.
-WAIT_DELETE_BUCKET_PROPAGATION_SECONDS = 10
+# Number of seconds to wait for bucket deletion or creation to propagate.
+WAIT_BUCKET_PROPAGATION_SECONDS = 60
 
 
 @unittest.skipIf(gcsio is None, 'GCP dependencies are not installed')
@@ -208,15 +208,17 @@ def test_create_default_bucket(self, mock_default_gcs_bucket_name):
     google_cloud_options.dataflow_kms_key = None
 
     import random
-    from hashlib import md5
+    from hashlib import blake2b
     # Add a random number to avoid collision if multiple test instances
     # are run at the same time. To avoid too many dangling buckets if bucket
     # removal fails, we limit the max number of possible bucket names in this
     # test to 1000.
-    overridden_bucket_name = 'gcsio-it-%d-%s-%s' % (
+    overridden_bucket_name = 'gcsio-it-%d-%s-%s-%d' % (
         random.randint(0, 999),
         google_cloud_options.region,
-        md5(google_cloud_options.project.encode('utf8')).hexdigest())
+        blake2b(google_cloud_options.project.encode('utf8'),
+                digest_size=4).hexdigest(),
+        int(time.time()))
 
     mock_default_gcs_bucket_name.return_value = overridden_bucket_name
 
@@ -225,13 +227,14 @@ def test_create_default_bucket(self, mock_default_gcs_bucket_name):
     if existing_bucket:
       try:
         existing_bucket.delete()
-        time.sleep(WAIT_DELETE_BUCKET_PROPAGATION_SECONDS)
+        time.sleep(WAIT_BUCKET_PROPAGATION_SECONDS)
       except NotFound:
         # Bucket existence check from get_bucket may be inaccurate due to gcs
         # cache or delay
         pass
 
     bucket = gcsio.get_or_create_default_gcs_bucket(google_cloud_options)
+    time.sleep(WAIT_BUCKET_PROPAGATION_SECONDS)
     self.assertIsNotNone(bucket)
     self.assertEqual(bucket.name, overridden_bucket_name)