Persisting closed sessions to external store inside updateStateByKey

[Anthony Brew]

Hi,
     I'm working on storing effectively what is a session that receives its
close event using spark streaming, by using updateStateByKey.

private static Function2<List<SessionUpdate>, Optional<SessionUpdate>,
Optional<SessionUpdate>>

 COLLECTED_SESSION = (newItems, current) -> {

        SessionUpdate returnValue = current.orNull();

        for (SessionUpdate i : newItems) {
             returnValue = (returnValue == null)?returnValue:returnValue.
combine(i);
        }

       if(returnValue.isClosed()){
           *// the question is should long term persistence happen here?*
*           // ie. someDataStore.store(ans);*
           return null;
       }else{
          return Optional.of(returnValue);
       }
};

For some reason this feels wrong, but it could just be I haven't
encountered this way of thinking before. Any pointers/confirmation of what
the best approach is to this would be great.

Thanks a Million,
Anthony
Phone: 087 - 9179799
Quidquid latine dictum sit, altum sonatur