code: public final synchronized void shutdown() {  // synchronize on 'this' to avoid races with registerStreams
          stopped = true;
          // If the cleanup task is running tell it to stop looping, and then remove it from the scheduled executor
          if (cleanupTaskFuture != null) {
              cleanupTaskFuture.cancel(false);
          }
  
          // Close remaining streams
          for (Map.Entry<InputStreamKey, TimedStreamEntry> entry : streamMap.entrySet()) {
              InputStreamKey key = entry.getKey();
              TimedStreamEntry timedStreamEntry = entry.getValue();
              //noinspection SynchronizationOnLocalVariableOrMethodParameter
              synchronized (timedStreamEntry) { // ensure there's no race with a request that got a ref before we removed it
                  closeStreamEntry(timedStreamEntry, key.requestId, key.index);
              }
          }
      }
docs: Closes any registered stream entries that have not yet been consumed