...

• The config directly determines when the source's backlog status should be True. However, it does not directly determine when the backlog should be false. The source's backlog is set to false if there is NO strategy (e.g. FLIP-309, FLIP-328) that says the source's backlog should be set to True. In other words, if the configured value is null, or if the watermark lag is less than or equal to the configured value, source's isProcessingBacklog status will not be affected by this config.
• The watermarkLag metric is proposed in FLIP-33.

Proposed Changes

1. Update the SourceOperator to emit RecordAttribute
   1. When the SourceOperator emits a watermark to downstream operators, it checks the watermark lag and emitsRecordAttribute(backlog=true) event to the downstream operator. The watermark lag is calculated the same way as defined in FLIP-33, WatermarkLag = CurrentTime - Watermark. When the current watermark lags behind the current time by a value greater than the threshold. A RecordAttribute(backlog=false) is sent when the watermark lag is within the threshold.
   2. When the watermark status of the SourceOperator changes to idle, it emits the RecordAttribute(backlog=false) event to downstream operators. When the watermark status changes from idle to active, it will check the current watermark and emit RecordAttributeevent according to the watermark threshold.

...