Softens the GBEK determinism requirement (#36495)

* Softens the GBEK determinism requirement

* fmt

* Fix lint/test

Author: damccorm

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

@@ -75,7 +75,7 @@
   from google.api_core.exceptions import ClientError, GoogleAPICallError
   from google.api_core.client_info import ClientInfo
   from google.cloud import bigquery as gcp_bigquery
-except ImportError:
+except Exception:
   gcp_bigquery = None
   pass
 

sdks/python/apache_beam/transforms/util.py

@@ -546,13 +546,18 @@ def expand(self, pcoll):
         pcoll.element_type).tuple_types)
     kv_type_hint = typehints.KV[key_type, value_type]
     if kv_type_hint and kv_type_hint != typehints.Any:
-      coder = coders.registry.get_coder(kv_type_hint).as_deterministic_coder(
-          f'GroupByEncryptedKey {self.label}'
-          'The key coder is not deterministic. This may result in incorrect '
-          'pipeline output. This can be fixed by adding a type hint to the '
-          'operation preceding the GroupByKey step, and for custom key '
-          'classes, by writing a deterministic custom Coder. Please see the '
-          'documentation for more details.')
+      coder = coders.registry.get_coder(kv_type_hint)
+      try:
+        coder = coder.as_deterministic_coder(self.label)
+      except ValueError:
+        logging.warning(
+            'GroupByEncryptedKey %s: '
+            'The key coder is not deterministic. This may result in incorrect '
+            'pipeline output. This can be fixed by adding a type hint to the '
+            'operation preceding the GroupByKey step, and for custom key '
+            'classes, by writing a deterministic custom Coder. Please see the '
+            'documentation for more details.',
+            self.label)
       if not coder.is_kv_coder():
         raise ValueError(
             'Input elements to the transform %s with stateful DoFn must be '