8365066: RecordingStream and RemoteRecordingStream do not terminate when the associated Recording is stopped or closed externally

src/jdk.jfr/share/classes/jdk/jfr/consumer/EventStream.java

@@ -35,6 +35,7 @@
 
 import jdk.jfr.internal.consumer.EventDirectoryStream;
 import jdk.jfr.internal.consumer.EventFileStream;
+import jdk.jfr.internal.consumer.FileEventSource;
 
 /**
  * Represents a stream of events.
@@ -112,7 +113,7 @@ public interface EventStream extends AutoCloseable {
     public static EventStream openRepository() throws IOException {
         return new EventDirectoryStream(
             null,
-            null,
+            new FileEventSource(),
             Collections.emptyList(),
             false
         );
@@ -137,7 +138,7 @@ public static EventStream openRepository(Path directory) throws IOException {
         Objects.requireNonNull(directory, "directory");
         return new EventDirectoryStream(
             directory,
-            null,
+            new FileEventSource(),
             Collections.emptyList(),
             true
         );

src/jdk.jfr/share/classes/jdk/jfr/consumer/RecordingStream.java

@@ -43,6 +43,8 @@
 import jdk.jfr.RecordingState;
 import jdk.jfr.internal.PlatformRecording;
 import jdk.jfr.internal.PrivateAccess;
+import jdk.jfr.internal.management.EventSource;
+import jdk.jfr.internal.consumer.LocalRecordingEventSource;
 import jdk.jfr.internal.util.Utils;
 import jdk.jfr.internal.consumer.EventDirectoryStream;
 import jdk.jfr.internal.management.StreamBarrier;
@@ -82,6 +84,7 @@ public void accept(Long endNanos) {
     private final EventDirectoryStream directoryStream;
     private long maxSize;
     private Duration maxAge;
+    private final EventSource eventSource;
 
     /**
      * Creates an event stream for the current JVM (Java Virtual Machine).
@@ -100,9 +103,10 @@ private RecordingStream(Map<String, String> settings) {
         this.recording.setName("Recording Stream: " + creationTime);
         try {
             PlatformRecording pr = PrivateAccess.getInstance().getPlatformRecording(recording);
+            this.eventSource = new LocalRecordingEventSource(pr);
             this.directoryStream = new EventDirectoryStream(
                 null,
-                pr,
+                eventSource,
                 configurations(),
                 false
             );
@@ -392,9 +396,9 @@ public boolean stop() {
         boolean stopped = false;
         try {
             try (StreamBarrier sb = directoryStream.activateStreamBarrier()) {
-                stopped = recording.stop();
+                stopped = eventSource.stop();
                 directoryStream.setCloseOnComplete(false);
-                sb.setStreamEnd(recording.getStopTime().toEpochMilli());
+                sb.setStreamEnd(eventSource.getStopTime());
             }
             directoryStream.awaitTermination();
         } catch (InterruptedException | IOException e) {

src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/AbstractEventStream.java

@@ -43,6 +43,7 @@
 import jdk.jfr.internal.LogLevel;
 import jdk.jfr.internal.LogTag;
 import jdk.jfr.internal.Logger;
+import jdk.jfr.internal.management.EventSource;
 
 /*
  * Purpose of this class is to simplify the implementation of
@@ -60,10 +61,12 @@ public abstract class AbstractEventStream implements EventStream {
     private volatile boolean waitForChunks = true;
     private Dispatcher dispatcher;
     private boolean daemon = false;
+    protected final EventSource eventSource;
 
 
-    AbstractEventStream(List<Configuration> configurations) throws IOException {
+    AbstractEventStream(List<Configuration> configurations, EventSource eventSource) throws IOException {
         this.configurations = configurations;
+        this.eventSource = eventSource;
     }
 
     @Override
@@ -207,8 +210,6 @@ public final void awaitTermination(Duration timeout) throws InterruptedException
 
     protected abstract void process() throws IOException;
 
-    protected abstract boolean isRecordingStream();
-
     protected final void closeParser() {
         parserState.close();
     }
@@ -250,7 +251,7 @@ private void startInternal(long startNanos) {
             if (streamConfiguration.started) {
                 throw new IllegalStateException("Event stream can only be started once");
             }
-            if (isRecordingStream() && streamConfiguration.startTime == null) {
+            if (eventSource.requiresStartTime() && streamConfiguration.startTime == null) {
                 streamConfiguration.setStartNanos(startNanos);
             }
             streamConfiguration.setStarted(true);

src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/EventDirectoryStream.java

@@ -41,7 +41,7 @@
 import jdk.jfr.internal.LogLevel;
 import jdk.jfr.internal.LogTag;
 import jdk.jfr.internal.Logger;
-import jdk.jfr.internal.PlatformRecording;
+import jdk.jfr.internal.management.EventSource;
 import jdk.jfr.internal.util.Utils;
 import jdk.jfr.internal.management.StreamBarrier;
 
@@ -55,7 +55,6 @@ public final class EventDirectoryStream extends AbstractEventStream {
     private static final Comparator<? super RecordedEvent> EVENT_COMPARATOR = JdkJfrConsumer.instance().eventComparator();
 
     private final RepositoryFiles repositoryFiles;
-    private final PlatformRecording recording;
     private final StreamBarrier barrier = new StreamBarrier();
     private final AtomicLong streamId = new AtomicLong();
     private ChunkParser currentParser;
@@ -66,11 +65,10 @@ public final class EventDirectoryStream extends AbstractEventStream {
 
     public EventDirectoryStream(
             Path p,
-            PlatformRecording recording,
+            EventSource eventSource,
             List<Configuration> configurations,
             boolean allowSubDirectories) throws IOException {
-        super(configurations);
-        this.recording = recording;
+        super(configurations, eventSource);
         this.repositoryFiles = new RepositoryFiles(p, allowSubDirectories);
         this.streamId.incrementAndGet();
         Logger.log(LogTag.JFR_SYSTEM_PARSER, LogLevel.INFO, "Stream " + streamId + " started.");
@@ -131,7 +129,7 @@ protected void processRecursionSafe() throws IOException {
         Dispatcher lastDisp = null;
         Dispatcher disp = dispatcher();
         Path path;
-        boolean validStartTime = isRecordingStream() || disp.startTime != null;
+        boolean validStartTime = eventSource.requiresStartTime() || disp.startTime != null;
         if (validStartTime) {
             path = repositoryFiles.firstPath(disp.startNanos, true);
         } else {
@@ -169,6 +167,19 @@ protected void processRecursionSafe() throws IOException {
                                      "ns (epoch), parser at " + lastFlush + "ns (epoch).");
                         return;
                     }
+
+                    if(!barrier.used()) {
+                        RecordingState state = eventSource.getState();
+                        if (state == RecordingState.CLOSED) {
+                            return;
+                        } else if (state == RecordingState.STOPPED) {
+                            long stopTime = eventSource.getStopTime();
+                            if (lastFlush > stopTime) {
+                                logStreamEnd("stopped at " + stopTime + "ns (epoch), parser at " + lastFlush + "ns (epoch).");
+                                return;
+                            }
+                        }
+                    }
                 }
                 long endNanos = currentParser.getStartNanos() + currentParser.getChunkDuration();
                 long endMillis = Instant.ofEpochSecond(0, endNanos).toEpochMilli();
@@ -181,9 +192,13 @@ protected void processRecursionSafe() throws IOException {
                     return;
                 }
 
-                if (isRecordingStream()) {
-                    if (recording.getState() == RecordingState.STOPPED && !barrier.used()) {
-                        logStreamEnd("recording stopped externally.");
+                if(!barrier.used()) {
+                    RecordingState state = eventSource.getState();
+                    if (state == RecordingState.CLOSED){
+                        logStreamEnd("Event source is closed externally");
+                        return;
+                    }else if(state == RecordingState.STOPPED) {
+                        logStreamEnd("Event source is stopped externally");
                         return;
                     }
                 }
@@ -226,10 +241,6 @@ private void logStreamEnd(String text) {
         Logger.log(LogTag.JFR_SYSTEM_PARSER, LogLevel.INFO, msg);
     }
 
-    protected boolean isRecordingStream() {
-        return recording != null;
-    }
-
     private void processOrdered(Dispatcher c) throws IOException {
         if (sortedCache == null) {
             sortedCache = new RecordedEvent[100_000];

src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/EventFileStream.java

@@ -45,7 +45,7 @@ public final class EventFileStream extends AbstractEventStream {
     private RecordedEvent[] cacheSorted;
 
     public EventFileStream(Path file) throws IOException {
-        super(Collections.emptyList());
+        super(Collections.emptyList(), new FileEventSource());
         this.input = new RecordingInput(file.toFile());
         this.input.setStreamed();
     }
@@ -71,11 +71,6 @@ public void close() {
         }
     }
 
-    @Override
-    protected boolean isRecordingStream() {
-        return false;
-    }
-
     @Override
     protected void process() throws IOException {
         Dispatcher disp = dispatcher();

src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/FileEventSource.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright (c) 2025, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+package jdk.jfr.internal.consumer;
+
+import jdk.jfr.internal.management.EventSource;
+
+public class FileEventSource implements EventSource {
+}

src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/LocalRecordingEventSource.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright (c) 2025, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+package jdk.jfr.internal.consumer;
+
+import jdk.jfr.RecordingState;
+
+import jdk.jfr.internal.PlatformRecording;
+import jdk.jfr.internal.management.EventSource;
+
+public class LocalRecordingEventSource implements EventSource {
+
+    private final PlatformRecording recording;
+
+    public LocalRecordingEventSource(PlatformRecording recording) {
+        this.recording = recording;
+    }
+
+    @Override
+    public boolean requiresStartTime() {
+        return true;
+    }
+
+    @Override
+    public long getStopTime() {
+        return recording.getStopTime().toEpochMilli();
+    }
+
+    @Override
+    public boolean stop() {
+        return recording.stop("stopped by RecordingStream");
+    }
+
+    @Override
+    public RecordingState getState()  {
+        return recording.getState();
+    }
+}

src/jdk.jfr/share/classes/jdk/jfr/internal/management/EventSource.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright (c) 2025, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+package jdk.jfr.internal.management;
+
+import jdk.jfr.RecordingState;
+
+public interface EventSource {
+
+    default long getStopTime() {
+        return Long.MAX_VALUE;
+    }
+
+    default boolean stop() {
+        return true;
+    }
+
+    default boolean requiresStartTime(){
+        return false;
+    }
+
+    default RecordingState getState() {
+        return RecordingState.RUNNING;
+    }
+}

src/jdk.jfr/share/classes/jdk/jfr/internal/management/ManagementSupport.java

@@ -163,10 +163,10 @@ public static Configuration newConfiguration(String name, String label, String d
     // with configuration objects
     public static EventStream newEventDirectoryStream(
             Path directory,
-            List<Configuration> confs) throws IOException {
+            List<Configuration> confs, EventSource eventSource) throws IOException {
         return new EventDirectoryStream(
             directory,
-            null,
+            eventSource,
             confs,
             false
         );