Enrichment Transform with BigTable handler

sdks/python/apache_beam/io/requestresponse.py

@@ -150,7 +150,7 @@ def _execute_request(
     try:
       return future.result(timeout=timeout)
     except TooManyRequests as e:
-      _LOGGER.warning(
+      _LOGGER.info(
           'request could not be completed. got code %i from the service.',
           e.code)
       raise e
@@ -264,10 +264,10 @@ def __init__(
       caller: Caller[RequestT, ResponseT],
       timeout: Optional[float] = DEFAULT_TIMEOUT_SECS,
       should_backoff: Optional[ShouldBackOff] = None,
-      repeater: Optional[Repeater] = ExponentialBackOffRepeater(),
+      repeater: Repeater = ExponentialBackOffRepeater(),
       cache_reader: Optional[CacheReader] = None,
       cache_writer: Optional[CacheWriter] = None,
-      throttler: Optional[PreCallThrottler] = DefaultThrottler(),
+      throttler: PreCallThrottler = DefaultThrottler(),
   ):
     """
     Instantiates a RequestResponseIO transform.
@@ -343,8 +343,8 @@ def __init__(
       caller: Caller[RequestT, ResponseT],
       timeout: Optional[float] = DEFAULT_TIMEOUT_SECS,
       should_backoff: Optional[ShouldBackOff] = None,
-      repeater: Optional[Repeater] = None,
-      throttler: Optional[PreCallThrottler] = None,
+      repeater: Repeater = None,
+      throttler: PreCallThrottler = None,
   ):
     self._caller = caller
     self._timeout = timeout

sdks/python/apache_beam/transforms/enrichment.py

@@ -14,7 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-
+import logging
 from typing import Any
 from typing import Callable
 from typing import Dict
@@ -37,6 +37,8 @@
 
 JoinFn = Callable[[Dict[str, Any], Dict[str, Any]], beam.Row]
 
+_LOGGER = logging.getLogger(__name__)
+
 
 def cross_join(left: Dict[str, Any], right: Dict[str, Any]) -> beam.Row:
   """cross_join performs a cross join on two `dict` objects.
@@ -54,6 +56,12 @@ def cross_join(left: Dict[str, Any], right: Dict[str, Any]) -> beam.Row:
     if k not in left:
       # Don't override the values in left.
       left[k] = v
+    elif left[k] != v:
+      _LOGGER.warning(
+          '%s exists in the input row as well the row fetched '
+          'from API but have different values. Using the input '
+          'value, you can override this behavior by passing a '
+          'custom `join_fn`.' % k)
   return beam.Row(**left)