code: @SuppressWarnings({ "unchecked", "rawtypes" })
      public synchronized void closeStreamingSessions(Callable resetCheckpointCallback) {
  
          closeStreamingSessions();
  
          Future future = streamingresults.submit(resetCheckpointCallback);
          try {
              future.get();
  
          } catch(InterruptedException e1) {
              logger.warn("Reset check point interrupted" + e1);
          } catch(ExecutionException e1) {
              logger.warn("Reset check point interrupted during execution" + e1);
          }
      }
docs: *
  @param resetCheckpointCallback - the callback that allows for the user to
  clean up the checkpoint at the end of the streaming session so a
  new session could, if necessary, start from 0 position.