Module: Karafka::Pro::Processing::Strategies::Lrj::Ftr
- Includes:
- Ftr::Default, Default
- Included in:
- Dlq::FtrLrj, FtrMom, FtrVp
- Defined in:
- lib/karafka/pro/processing/strategies/lrj/ftr.rb
Overview
Filtering enabled Long-Running Job enabled
In general aside from throttling this one will behave the same way as the Lrj
Constant Summary collapse
- FEATURES =
Features for this strategy
%i[ filtering long_running_job ].freeze
Constants included from Default
Instance Method Summary collapse
-
#handle_after_consume ⇒ Object
LRJ standard flow after consumption with potential throttling on success.
Methods included from Default
#handle_before_enqueue, #handle_revoked
Methods included from Default
#handle_before_consume, #handle_before_enqueue, #handle_consume, #handle_revoked
Methods included from Karafka::Processing::Strategies::Default
#commit_offsets, #commit_offsets!, #handle_before_consume, #handle_before_enqueue, #handle_consume, #handle_idle, #handle_revoked, #handle_shutdown, #mark_as_consumed, #mark_as_consumed!
Methods included from Karafka::Processing::Strategies::Base
#handle_before_consume, #handle_before_enqueue, #handle_consume, #handle_idle, #handle_revoked, #handle_shutdown
Methods included from Ftr::Default
#handle_idle, #handle_post_filtering
Instance Method Details
#handle_after_consume ⇒ Object
LRJ standard flow after consumption with potential throttling on success
34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 |
# File 'lib/karafka/pro/processing/strategies/lrj/ftr.rb', line 34 def handle_after_consume coordinator.on_finished do || if coordinator.success? coordinator.pause_tracker.reset # Manual pausing has the highest priority return if coordinator.manual_pause? # It's not a MoM, so for successful we need to mark as consumed mark_as_consumed() unless revoked? # If still not revoked and was throttled, we need to apply throttling logic if coordinator.filtered? && !revoked? handle_post_filtering elsif !revoked? && !coordinator.manual_seek? # If not revoked and not throttled, we move to where we were suppose to and # resume seek(coordinator.seek_offset, false) resume else resume end else # If processing failed, we need to pause # For long running job this will overwrite the default never-ending pause and # will cause the processing to keep going after the error backoff retry_after_pause end end end |