One issue with the introduction of streaming is how to deal with state that accumulates over the course of a long-running job. In this case, it seems like this particular HashMap only gets larger, which will pose a problem. I'm going to defer to TD or Matei here to figure out how to best address this particular case. TD has a handful of MetaDataCleaner implementations which go through and delete things based on a timestamp (checkout DAGScheduler), that may be appropriate to use here.