Re: [I] [Task]: Add Python AfterSynchronizedProcessingTime trigger and add an Iceberg CDC streaming read test [beam]

["marianobrc (via GitHub)" <gi...@apache.org>]

marianobrc commented on issue #34212:
URL: https://github.com/apache/beam/issues/34212#issuecomment-3334874920

   I'm having the same issue, tried all the beam sdk versions from` v2.57` to `v2.68` but I still get the same error.
   As a workaround I monkey-patched the code like this:
   ```python
   # main.py
   
   # Apply Apache Beam trigger patch before importing beam modules
   def _apply_beam_trigger_patch():
       """Apply monkey patch to fix after_synchronized_processing_time KeyError."""
       import apache_beam as beam
       from apache_beam.transforms.trigger import TriggerFn, AfterProcessingTime
       
       original_from_runner_api = TriggerFn.from_runner_api
       
       @staticmethod
       def patched_from_runner_api(proto, context):
           """Patched version that handles after_synchronized_processing_time."""
           from apache_beam.transforms.trigger import (
               AfterAll, AfterAny, AfterEach, AfterWatermark, AfterProcessingTime,
               Always, DefaultTrigger, AfterCount, _Never, OrFinally, Repeatedly
           )
           
           trigger_type = proto.WhichOneof('trigger')
           
           # Handle the special case of after_synchronized_processing_time
           if trigger_type == 'after_synchronized_processing_time':
               return AfterProcessingTime(delay=0)
           
           # Handle all other trigger types
           trigger_mapping = {
               'after_all': AfterAll,
               'after_any': AfterAny,
               'after_each': AfterEach,
               'after_end_of_window': AfterWatermark,
               'after_processing_time': AfterProcessingTime,
               'always': Always,
               'default': DefaultTrigger,
               'element_count': AfterCount,
               'never': _Never,
               'or_finally': OrFinally,
               'repeat': Repeatedly,
           }
           
           if trigger_type not in trigger_mapping:
               raise ValueError(f"Unknown trigger type: {trigger_type}")
           
           return trigger_mapping[trigger_type].from_runner_api(proto, context)
       
       TriggerFn.from_runner_api = patched_from_runner_api
   
   # Apply the patch before importing beam
   _apply_beam_trigger_patch()
   
   import apache_beam as beam
   
   # ... the pipeline definition and the rest of my main file goes here after patching
   
   ```


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org