static void processLastRow ( BaseRow currentRow , boolean generateRetraction , ValueState < BaseRow > state , Collector < BaseRow > out ) throws Exception { // Check message should be accumulate Preconditions . checkArgument ( BaseRowUtil . isAccumulateMsg ( currentRow ) ) ; if ( generateRetraction ) { // state stores complete row if generateRetraction is true BaseRow preRow = state . value ( ) ; state . update ( currentRow ) ; if ( preRow != null ) { preRow . setHeader ( BaseRowUtil . RETRACT_MSG ) ; out . collect ( preRow ) ; } } out . collect ( currentRow ) ; }
Processes element to deduplicate on keys sends current element as last row retracts previous element if needed .