This enhancement proposes an improvement to the current behavior of Window Evictor, by providing more control on how the elements are to be evicted from the Window. Original Design Document of this proposal can be found here

Status

Current state: Under Discussion

Discussion thread: http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-Enhance-Window-Evictor-in-Flink-td12406.html

JIRA: https://issues.apache.org/jira/browse/FLINK-4174

...

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

• Changes to the Evictor interface :  addition of two new methods evictBefore and evictAfter, removal of the existing evict method
• Corresponding changes to CountEvictor, DeltaEvictor
• Renaming of the existing TimeEvictor as ProcessingTimeEvictor(since the current behavior of TimeEvictor does not consider the EventTime of the records) and addition of new class EventTimeEvictorNew class TimestampedValue to store records with a timestamp. This class is exposed to the users in the evictBefore and evictAfter methods.
• New overloaded method CountEvictor.of() that takes an additional parameter doEvictAfter, which decides whether to do eviction after the WindowFunction.
• Similar overloaded methods in DeltaEvictor , ProcessingTimeEvictor and EventTimeEvictor. TimeEvictor

Proposed Changes

We propose to change the interface of Evictor to this

public interface Evictor<T, W extends Window> extends Serializable {

   /**
    * Optionally evicts elements. Called before windowing function.
    * @param elements The elements currently in the pane.
    * @param size The current number of elements in the pane.
    * @param window The {@link Window}
    * @param evictorContext The context for the Evictor
     */
   void evictBefore(Iterable<StreamRecord<T>>Iterable<TimestampedValue<T>> elements, int size, W window, EvictorContext evictorContext);

   /**
    * Optionally evicts elements. Called after windowing function.
    * @param elements The elements currently in the pane.
    * @param size The current number of elements in the pane.
    * @param window The {@link Window}
    * @param evictorContext The context for the Evictor
    */
   void evictAfter(Iterable<StreamRecord<T>>Iterable<TimestampedValue<T>> elements, int size, W window, EvictorContext evictorContext);
 
  /**
   * A context object that is given to {@link Evictor} methods
   */
   interface EvictorContext {}
}

...

CountEvictor, DeltaEvictor, ProcessingTimeEvictor and EventTimeEvictor TimeEvictor by default will evict elements before the window function. 

DeltaEvictor will always iterate through the entire elements and evict if the element has higher delta than the threshold. EventTimeEvictor TimeEvictor will iterate through evict all the elements and remove the elements with timestamp <= evictCutoff. ProcessingTimeEvictor will evict elements from the window till it finds the first element that has that have a timestamp <= the evictCutoff (Since we are not concerned about out-of-order events). CountEvictor will only iterate through the elements till the count of elements in the window is reduced to specified size. CountEvictor will do the iteration only till the count of elements in the window is reduced to specified size. 

Overloaded method in DeltaEvictor to make the Evictor evict after the windowFunction:

...

Similar overloaded methods are added to CountEvictor , EventTimeEvictor and ProcessingTimeEvictorTimeEvictor.

Compatibility, Deprecation, and Migration Plan

...

• Users currently using DeltaEvictor will have to be aware that the new implementation will apply the delta function to all the elements in the window(not till it finds the first element in the pane)
• Users who have currently implemented a custom Evictor will have to adapt to the new interface of the Evictor.
• CountEvictor will behave the same even after these changes.
• By default, all Evictors will evict before the WindowFunction, which matches the existing behavior.