available() 等于零，为什么后面还能读到数据？

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

public class AvailableBoundary {
    static final byte[] FIRST = "alpha|".getBytes(StandardCharsets.US_ASCII);
    static final byte[] SECOND = "omega".getBytes(StandardCharsets.US_ASCII);
    static final int WAIT_SECONDS = 5;

    record Chunk(int count, byte[] bytes) {}
    record Row(String stage, int available, boolean writerClosed,
               int totalRead, String lastRead, boolean sawEof) {
        String json() {
            return String.format(Locale.ROOT,
                "{\"stage\":\"%s\",\"available\":%d,\"writer_closed\":%s,"
                + "\"total_read\":%d,\"last_read\":\"%s\",\"saw_eof\":%s}",
                stage, available, writerClosed, totalRead, lastRead, sawEof);
        }
        String csv() {
            return String.join(",", stage, Integer.toString(available),
                Boolean.toString(writerClosed), Integer.toString(totalRead), lastRead,
                Boolean.toString(sawEof));
        }
    }

    static void require(boolean condition, String message) {
        if (!condition) throw new AssertionError(message);
    }
    static void await(CountDownLatch signal, String name) throws InterruptedException {
        require(signal.await(WAIT_SECONDS, TimeUnit.SECONDS), "timed out: " + name);
    }
    static void producerOk(AtomicReference<Throwable> failure) {
        if (failure.get() != null) throw new AssertionError("producer failed", failure.get());
    }
    static Chunk boundedRead(ExecutorService reader, PipedInputStream input) throws Exception {
        Future<Chunk> future = reader.submit(() -> {
            byte[] buffer = new byte[8];
            int count = input.read(buffer); // Positive buffer size; actual read result matters.
            return new Chunk(count, count > 0 ? Arrays.copyOf(buffer, count) : new byte[0]);
        });
        try { return future.get(WAIT_SECONDS, TimeUnit.SECONDS); }
        finally { if (!future.isDone()) future.cancel(true); }
    }
    static void add(List<Row> rows, String stage, PipedInputStream input,
                    AtomicBoolean closed, ByteArrayOutputStream received,
                    String lastRead, boolean eof, int expectedAvailable) throws IOException {
        int available = input.available();
        require(available == expectedAvailable, stage + ": available=" + available);
        rows.add(new Row(stage, available, closed.get(), received.size(), lastRead, eof));
    }

    public static void main(String[] args) throws Exception {
        Path output = args.length == 0 ? Path.of("available-output") : Path.of(args[0]);
        PipedInputStream input = new PipedInputStream(64);
        PipedOutputStream pipe = new PipedOutputStream(input);
        CountDownLatch allowFirst = new CountDownLatch(1);
        CountDownLatch firstWritten = new CountDownLatch(1);
        CountDownLatch allowSecond = new CountDownLatch(1);
        CountDownLatch writerDone = new CountDownLatch(1);
        AtomicBoolean writerClosed = new AtomicBoolean(false);
        AtomicReference<Throwable> producerFailure = new AtomicReference<>();
        ExecutorService reader = Executors.newSingleThreadExecutor();
        Thread producer = new Thread(() -> {
            try {
                await(allowFirst, "first gate");
                pipe.write(FIRST);
                pipe.flush();
                firstWritten.countDown();
                await(allowSecond, "second gate");
                pipe.write(SECOND);
                pipe.flush();
                pipe.close();
                writerClosed.set(true); // Set only after this thread's close returned.
            } catch (Throwable failure) {
                producerFailure.set(failure);
            } finally {
                firstWritten.countDown();
                writerDone.countDown();
            }
        }, "fixture-producer");
        List<Row> rows = new ArrayList<>();
        ByteArrayOutputStream received = new ByteArrayOutputStream();
        producer.start();
        try {
            add(rows, "before_first_write", input, writerClosed, received, "not_called", false, 0);
            require(!writerClosed.get(), "writer must be open at the first gate");
            allowFirst.countDown();
            await(firstWritten, "first data");
            producerOk(producerFailure);
            add(rows, "first_chunk_buffered", input, writerClosed, received, "not_called", false, 6);
            Chunk first = boundedRead(reader, input);
            require(first.count == 6 && Arrays.equals(first.bytes, FIRST), "first chunk");
            received.write(first.bytes);
            add(rows, "between_chunks", input, writerClosed, received, "6", false, 0);
            require(!writerClosed.get(), "writer still waits at second gate");
            allowSecond.countDown();
            await(writerDone, "writer close");
            producerOk(producerFailure);
            require(writerClosed.get(), "producer close was not observed");
            add(rows, "second_buffered_writer_closed", input, writerClosed, received, "6", false, 5);
            Chunk second = boundedRead(reader, input);
            require(second.count == 5 && Arrays.equals(second.bytes, SECOND), "second chunk");
            received.write(second.bytes);
            add(rows, "all_payload_read", input, writerClosed, received, "5", false, 0);
            Chunk eof = boundedRead(reader, input);
            require(eof.count == -1, "actual EOF signal required");
            add(rows, "eof_observed", input, writerClosed, received, "-1", true, 0);
            require(received.toString(StandardCharsets.US_ASCII).equals("alpha|omega"), "payload");
        } finally {
            // Unblock every controlled wait even if an assertion or Future times out.
            allowFirst.countDown();
            allowSecond.countDown();
            input.close();
            pipe.close();
            producer.interrupt();
            producer.join(TimeUnit.SECONDS.toMillis(WAIT_SECONDS));
            reader.shutdownNow();
            require(reader.awaitTermination(WAIT_SECONDS, TimeUnit.SECONDS), "reader not terminated");
            require(!producer.isAlive(), "producer not terminated");
        }
        Files.createDirectories(output);
        String version = System.getProperty("java.version");
        String json = "{\n  \"java_version\":\"" + version
            + "\",\n  \"wait_limit_seconds\":5,\n  \"coordination\":\"latches_no_sleep\",\n  \"rows\":[\n    "
            + String.join(",\n    ", rows.stream().map(Row::json).toList()) + "\n  ]\n}\n";
        String csv = "stage,available,writer_closed,total_read,last_read,saw_eof\n"
            + String.join("\n", rows.stream().map(Row::csv).toList()) + "\n";
        Files.writeString(output.resolve("report.json"), json, StandardCharsets.UTF_8);
        Files.writeString(output.resolve("timeline.csv"), csv, StandardCharsets.UTF_8);
        Files.write(output.resolve("received.txt"), received.toByteArray());
        System.out.println("Java " + version + "; 6 stages; 11 bytes; workers terminated");
        System.out.println("available=0 before writing, between chunks, and after EOF");
        System.out.println("writer closed with 5 buffered bytes; actual final read=-1");
        System.out.println("PASS; " + output);
    }
}