code: 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);
  	}
docs: Processes element to deduplicate on keys, sends current element as last row, retracts previous element if
  needed.
  
  @param currentRow latest row received by deduplicate function
  @param generateRetraction whether need to send retract message to downstream
  @param state state of function
  @param out underlying collector
  @throws Exception