Put the stream in try-catch to stop the back-end reading thread if error happens
<br />
<code>
try (Stream<Integer> stream = Stream.parallelConcat(a,b, ...)) {
stream.forEach(N::println);
}
</code>

@param a
@return
541823
@SafeVarargs
    public static <T> Stream<T> parallelConcat(final Iterator<? extends T>... a) {
        return parallelConcat(a, DEFAULT_READING_THREAD_NUM, calculateQueueSize(a.length));
    }
1050798
private boolean before() throws IOException {
        if (closed) {
            throw new IOException("Attempted read from closed stream.");
        }
        switch (length) {
            case EOF:
                return false;
            case UNKNOWN:
                length = readLength();
                pos = 0;
                if (length == EOF) {
                    HeaderList.parse(src); // result is ignored
                    return false;
                }
                return true;
            default:
                return true;
        }
    }
396441
@Override
    public int read() throws IOException {
        
        //Return -1 here because if we are closed or finished, we're at the end of the stream
        if(isClosed || isFinished()){
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                Tr.debug(tc, "Nothing to read, stream is closed : " + isClosed + ", or finished : " + isFinished());
            return -1;
        }
        
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "read" , "SRTInputStream31.read()");
        }
        
        int returnByte;
        
        if(this.request.isAsyncStarted()&& (this.getReadListener()!= null)){
            synchronized(this){
//                Check if isReady had returned false. If it did then an IllegalArgumentException will be thrown if a listener is set
                isReadyFalseCheck();
                returnByte = super.read();
            }
        } else{
            returnByte = super.read();
        }
        
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "read" , "SRTInputStream31.read() : " + returnByte);
        }
        
        return returnByte;
    }
1122387
@SafeVarargs
    @SuppressWarnings("unchecked")
    public static <T> Stream<T> stream(Iterable<T>... iterables) {
        List<Iterator<T>> iterators = Arrays.stream(iterables)
            .map(Iterable::iterator)
            .collect(Collectors.toList());

        return stream(iterators.toArray(new Iterator[iterables.length]));
    }
271209
private void readAsync() throws IOException {
    stateChangeLock.lock();
    final byte[] arr = readAheadBuffer.array();
    try {
      if (endOfStream || readInProgress) {
        return;
      }
      checkReadException();
      readAheadBuffer.position(0);
      readAheadBuffer.flip();
      readInProgress = true;
    } finally {
      stateChangeLock.unlock();
    }
    executorService.execute(() -> {
      stateChangeLock.lock();
      try {
        if (isClosed) {
          readInProgress = false;
          return;
        }
        // Flip this so that the close method will not close the underlying input stream when we
        // are reading.
        isReading = true;
      } finally {
        stateChangeLock.unlock();
      }

      // Please note that it is safe to release the lock and read into the read ahead buffer
      // because either of following two conditions will hold - 1. The active buffer has
      // data available to read so the reader will not read from the read ahead buffer.
      // 2. This is the first time read is called or the active buffer is exhausted,
      // in that case the reader waits for this async read to complete.
      // So there is no race condition in both the situations.
      int read = 0;
      int off = 0, len = arr.length;
      Throwable exception = null;
      try {
        // try to fill the read ahead buffer.
        // if a reader is waiting, possibly return early.
        do {
          read = underlyingInputStream.read(arr, off, len);
          if (read <= 0) break;
          off += read;
          len -= read;
        } while (len > 0 && !isWaiting.get());
      } catch (Throwable ex) {
        exception = ex;
        if (ex instanceof Error) {
          // `readException` may not be reported to the user. Rethrow Error to make sure at least
          // The user can see Error in UncaughtExceptionHandler.
          throw (Error) ex;
        }
      } finally {
        stateChangeLock.lock();
        readAheadBuffer.limit(off);
        if (read < 0 || (exception instanceof EOFException)) {
          endOfStream = true;
        } else if (exception != null) {
          readAborted = true;
          readException = exception;
        }
        readInProgress = false;
        signalAsyncReadComplete();
        stateChangeLock.unlock();
        closeUnderlyingInputStreamIfNecessary();
      }
    });
  }
725381
public static <T> Stream<T> parallelConcat(final Collection<? extends Stream<? extends T>> c) {
        return parallelConcat(c, DEFAULT_READING_THREAD_NUM);
    }
1362729
public static <T> Stream<T> iteratorToStream(final Iterator<T> iterator, final Boolean parallel) {
        return stream(spliteratorUnknownSize(iterator, IMMUTABLE), parallel);
    }
245683
public static boolean close(final InputStream stream) {
      // do not attempt to close null stream, but return sucess
      if (stream == null) {
         return true;
      }

      boolean success = true;

      try {
         stream.close();
      } catch (IOException e) {
         success = false;
      }

      return success;
   }
1223191
private Node tryAppend(Node s, boolean haveData) {
        for (Node t = tail, p = t;;) {        // move p to last node and append
            Node n, u;                        // temps for reads of next & tail
            if (p == null && (p = head) == null) {
                if (casHead(null, s))
                    return s;                 // initialize
            }
            else if (p.cannotPrecede(haveData))
                return null;                  // lost race vs opposite mode
            else if ((n = p.next) != null)    // not last; keep traversing
                p = p != t && t != (u = tail) ? (t = u) : // stale tail
                    (UNLINKED != n) ? n : null; // restart if off list
            else if (!p.casNext(null, s))
                p = p.next;                   // re-read on CAS failure
            else {
                if (p != t) {                 // update if slack now >= 2
                    while ((tail != t || !casTail(t, s)) &&
                           (t = tail)   != null &&
                           (s = t.next) != null && // advance and retry
                           (s = s.next) != null && s != UNLINKED);
                }
                return p;
            }
        }
    }
1058657
static byte[] readInputStreamIntoByteArrayAndClose(InputStream inputStream) throws IOException {

        byte[] bytes = null;
        try{
            bytes = ByteStreams.toByteArray(inputStream);
        }catch(IOException e){
            throw e;
        } finally {
            IOUtils.closeQuietly(inputStream);
            if(bytes==null) {
                bytes = new byte[ 1 ];
            }
        }

        return bytes;
    }
307321
private void closeStreams(final Process process) throws IOException {
    IOException caught = null;

    try {
      process.getOutputStream().close();
    }
    catch (IOException e) {
      if (e.getMessage().equals("Stream closed")) {
        /**
         * OutputStream's contract for the close() method: If the stream is already closed then invoking this method has no effect.
         *
         * When a UNIXProcess exits ProcessPipeOutputStream automatically closes its target FileOutputStream and replaces it with NullOutputStream.
         * However the ProcessPipeOutputStream doesn't close itself at that moment.
         * As ProcessPipeOutputStream extends BufferedOutputStream extends FilterOutputStream closing it flushes the buffer first.
         * In Java 7 closing FilterOutputStream ignores any exception thrown by the target OutputStream. Since Java 8 these exceptions are now thrown.
         *
         * So since Java 8 after UNIXProcess detects the exit and there's something in the output buffer closing this stream throws IOException
         * with message "Stream closed" from NullOutputStream.
         */
        log.trace("Failed to close process output stream:", e);
      }
      else {
        log.error("Failed to close process output stream:", e);
        caught = add(caught, e);
      }
    }

    try {
      process.getInputStream().close();
    }
    catch (IOException e) {
      log.error("Failed to close process input stream:", e);
      caught = add(caught, e);
    }

    try {
      process.getErrorStream().close();
    }
    catch (IOException e) {
      log.error("Failed to close process error stream:", e);
      caught = add(caught, e);
    }

    if (caught != null) {
      throw caught;
    }
  }
910528
public long skip(long n) throws IllegalArgumentException, IOException {

        if (lookaheadChar == UNDEFINED) {
            lookaheadChar = super.read();
        }

        // illegal argument
        if (n < 0) {
            throw new IllegalArgumentException("negative argument not supported");
        }

        // no skipping
        if (n == 0 || lookaheadChar == END_OF_STREAM) {
            return 0;
        }

        // skip and reread the lookahead-char
        long skiped = 0;
        if (n > 1) {
            skiped = super.skip(n - 1);
        }
        lookaheadChar = super.read();
        // fixme uh: we should check the skiped sequence for line-terminations...
        lineCounter = Integer.MIN_VALUE;
        return skiped + 1;
    }
33069
public static void executeHelp(String[] args, PrintStream stream) throws Exception {
        String subCmd = (args.length > 0) ? args[0] : "";
        if(subCmd.equals("get")) {
            SubCommandQuotaGet.printHelp(stream);
        } else if(subCmd.equals("set")) {
            SubCommandQuotaSet.printHelp(stream);
        } else if(subCmd.equals("reserve-memory")) {
            SubCommandQuotaReserveMemory.printHelp(stream);
        } else if(subCmd.equals("unset")) {
            SubCommandQuotaUnset.printHelp(stream);
        } else {
            printHelp(stream);
        }
    }
1358737
public static void skip(InputStream stream, long skip) throws IOException
   {
      long count = skip;
      while (count > 0)
      {
         count -= stream.skip(count);
      }
   }
972025
public void closeAll() {

        // Close any active streams
        for (InterceptedStream<T> stream : streams.values())
            close(stream.getStream());

        // Remove now-useless references
        streams.clear();

    }

    /**
     * Blocks until the given stream is closed, or until another stream with
     * the same index replaces it.
     *
     * @param stream
     *     The stream to wait for.
     */
    public void waitFor(InterceptedStream<T> stream) {

        T underlyingStream = stream.getStream();

        // Wait for stream to close
        synchronized (underlyingStream) {
            while (streams.get(stream.getIndex()) == stream) {
                try {
                    underlyingStream.wait(STREAM_WAIT_TIMEOUT);
                }
                catch (InterruptedException e) {
                    // Ignore
                }
            }
        }

    }

    /**
     * Returns the stream stored in this map under the given index.
     *
     * @param index
     *     The index of the stream to return.
     *
     * @return
     *     The stream having the given index, or null if no such stream is
     *     stored within this map.
     */
    public InterceptedStream<T> get(String index) {
        return streams.get(index);
    }

    /**
     * Adds the given stream to this map, storing it under its associated
     * index. If another stream already exists within this map having the same
     * index, that stream will be closed and replaced.
     *
     * @param stream
     *     The stream to store within this map.
     */
    public void put(InterceptedStream<T> stream) {

        // Add given stream to map
        InterceptedStream<T> oldStream =
                streams.put(stream.getIndex(), stream);

        // If a previous stream DID exist, close it
        if (oldStream != null)
            close(oldStream.getStream());

    }

}
1351984
private void closeStream(InputStream stream) {

        if (stream != null) {
            try {
                stream.close();
            } catch (Exception ex) {
                LOG.error(ex.getLocalizedMessage(), ex);
            }
        }
    }
669685
@Override
    public long skip(long n) throws IOException {
        //Return -1 here because if we are closed or finished, we're at the end of the stream
        if(isClosed || isFinished()){
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                Tr.debug(tc, "Nothing to read, stream is closed : " + isClosed + ", or finished : " + isFinished());
            return -1;
        }
        
        return super.skip(n);
    }