private boolean updateStreamBlacklist(Set<Tuple2<String, String>> processingBlacklist,
                                          Message message,
                                          Set<String> initialStreamIds) {
        boolean addedStreams = false;
        for (Stream stream : message.getStreams()) {
            if (!initialStreamIds.remove(stream.getId())) {
                addedStreams = true;
            } else {
                // only add pre-existing streams to blacklist, this has the effect of only adding already processed streams,
                // not newly added ones.
                processingBlacklist.add(tuple(message.getId(), stream.getId()));
            }
        }
        return addedStreams;
    }
private boolean updateStreamBlacklist ( Set < Tuple2 < String , String > > processingBlacklist , Message message , Set < String > initialStreamIds ) { boolean addedStreams = false ; for ( Stream stream : message . getStreams ( ) ) { if ( ! initialStreamIds . remove ( stream . getId ( ) ) ) { addedStreams = true ; } else { processingBlacklist . add ( tuple ( message . getId ( ) , stream . getId ( ) ) ) ; } } return addedStreams ; }
<msgid, stream> that should not be run again (which prevents re-running pipelines over and over again)