add timeout code for redoer

Author: spompea-cfa

README.md

@@ -202,7 +202,13 @@ Similar to the consumer, the redoer is also a continually-running process.
 docker compose run --env AWS_PROFILE=localstack --env LOG_LEVEL=DEBUG redoer
 ```
 
-`LOG_LEVEL` is optional; defaults to `INFO`.
+Environment variables:
+
+- `LOG_LEVEL` is optional; defaults to `INFO`.
+- `SZ_CALL_TIMEOUT_SECONDS`
+  - Optional; defaults to 420 seconds (7 min.)
+  - Sets the maximum amount of time the Exporter will wait for a Senzing
+    `process_redo_record` to complete before bailing and moving on.
 
 ### Exporter
 

middleware/consumer.py

@@ -187,7 +187,7 @@ def clean_up(signum, frm):
                 start_alarm_timer(SZ_CALL_TIMEOUT_SECONDS)
                 resp = sz_eng.add_record(rcd['DATA_SOURCE'], rcd['RECORD_ID'], body)
                 cancel_alarm_timer()
-                log.info(SZ_TAG + 'Successful add_record having ReceiptHandle: '
+                log.debug(SZ_TAG + 'Successful add_record having ReceiptHandle: '
                          + receipt_handle)
             except sz.SzUnknownDataSourceError as sz_uds_err:
                 try:

middleware/redoer.py

@@ -8,6 +8,8 @@
 from loglib import *
 log = retrieve_logger()
 
+from timeout_handling import *
+
 try:
     log.info('Importing senzing_core library . . .')
     import senzing_core as sz_core
@@ -17,6 +19,7 @@
     log.error(e)
     sys.exit(1)
 
+SZ_CALL_TIMEOUT_SECONDS = int(os.environ.get('SZ_CALL_TIMEOUT_SECONDS', 420))
 SZ_CONFIG = json.loads(os.environ['SENZING_ENGINE_CONFIGURATION_JSON'])
 
 # How long to wait before attempting next Senzing op.
@@ -62,7 +65,9 @@ def go():
 
             if have_rcd:
                 try:
+                    start_alarm_timer(SZ_CALL_TIMEOUT_SECONDS)
                     sz_eng.process_redo_record(rcd)
+                    cancel_alarm_timer()
                     have_rcd = 0
                     log.debug(SZ_TAG + 'Successfully redid one record via process_redo_record().')
                     continue
@@ -77,6 +82,12 @@ def go():
                                   + ' for this record; dropping on the floor and moving on.')
                     time.sleep(WAIT_SECONDS)
                     continue
+                except LongRunningCallTimeoutEx as lrex:
+                    # Abandon and move on.
+                    have_rcd = 0
+                    log.error(f'{SZ_TAG} {type(lrex).__module__}.{type(lrex).__qualname__} :: '
+                              + f'Long-running Senzing add_record call exceeded {SZ_CALL_TIMEOUT_SECONDS} sec.; '
+                              + f'skipping and moving on; receipt_handle was: {receipt_handle}')
                 except sz.SzError as sz_err:
                     log.error(SZ_TAG + str(sz_err))
                     sys.exit(1)