[Archive] Return an error code when playing and stopping an empty recording.

aeron-archive/src/main/java/io/aeron/archive/ArchiveConductor.java

@@ -737,6 +737,12 @@ else if (ctx.replayBuffer().capacity() < recordingSummary.mtuLength)
         else if (AeronArchive.REPLAY_ALL_AND_STOP == length)
         {
             replayLength = (stopPosition - replayPosition);
+            if (0 == replayLength)
+            {
+                final String msg =
+                    "When replaying and stopping the replay length must be non-zero, recordingId=" + recordingId;
+                controlSession.sendErrorResponse(correlationId, EMPTY_RECORDING, msg);
+            }
         }
         else
         {

aeron-archive/src/main/java/io/aeron/archive/client/ArchiveException.java

@@ -98,6 +98,11 @@ public class ArchiveException extends AeronException
      */
     public static final int REPLICATION_CONNECTION_FAILURE = 14;
 
+    /**
+     * The recording specified for replay is empty.
+     */
+    public static final short EMPTY_RECORDING = 15;
+
     private static final long serialVersionUID = 386758252787901080L;
 
     /**

aeron-system-tests/src/test/java/io/aeron/archive/ArchiveReplayTest.java

@@ -17,10 +17,12 @@
 
 import io.aeron.Aeron;
 import io.aeron.ChannelUri;
+import io.aeron.CommonContext;
 import io.aeron.Image;
 import io.aeron.Publication;
 import io.aeron.Subscription;
 import io.aeron.archive.client.AeronArchive;
+import io.aeron.archive.client.ArchiveException;
 import io.aeron.archive.client.ReplayParams;
 import io.aeron.archive.status.RecordingPos;
 import io.aeron.driver.MediaDriver;
@@ -56,6 +58,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.fail;
 
 @ExtendWith({ InterruptingTestCallback.class, EventLogExtension.class })
 public class ArchiveReplayTest
@@ -161,38 +164,20 @@ void shouldExitOnEmptyRecording()
 
             final int replayStreamId = 10001;
 
-            final long replaySessionId = aeronArchive.startReplay(
-                recordingId,
-                IPC_CHANNEL,
-                replayStreamId,
-                new ReplayParams().position(NULL_POSITION).length(REPLAY_ALL_AND_STOP));
-
-            final String replayChannel = ChannelUri.addSessionId(IPC_CHANNEL, (int)replaySessionId);
-            final AtomicReference<Image> availableImage = new AtomicReference<>();
-            final AtomicReference<Image> unavailableImage = new AtomicReference<>();
-            final Subscription replay = aeron.addSubscription(
-                replayChannel, replayStreamId, availableImage::set, unavailableImage::set);
-
-            Image image;
-            while (null == (image = availableImage.get()))
+            try
             {
-                aeronArchive.checkForErrorResponse();
-                Tests.yield();
-            }
+                aeronArchive.startReplay(
+                    recordingId,
+                    CommonContext.IPC_CHANNEL,
+                    replayStreamId,
+                    new ReplayParams().position(AeronArchive.NULL_POSITION).length(AeronArchive.REPLAY_ALL_AND_STOP));
 
-            while (!image.isEndOfStream())
-            {
-                aeronArchive.checkForErrorResponse();
-                Tests.yield();
+                fail("Should have thrown exception");
             }
-
-            while (image != unavailableImage.get())
+            catch (final ArchiveException ex)
             {
-                Tests.yield();
+                assertEquals(ArchiveException.EMPTY_RECORDING, ex.errorCode());
             }
-
-            CloseHelper.quietClose(replay);
-            aeronArchive.stopReplay(replaySessionId);
         }
     }
 
@@ -270,38 +255,20 @@ void shouldExitOnEmptyLiveRecording()
 
                 final int replayStreamId = 10001;
 
-                final long replaySessionId = aeronArchive.startReplay(
-                    recordingId,
-                    IPC_CHANNEL,
-                    replayStreamId,
-                    new ReplayParams().position(NULL_POSITION).length(REPLAY_ALL_AND_STOP));
-
-                final String replayChannel = ChannelUri.addSessionId(IPC_CHANNEL, (int)replaySessionId);
-                final AtomicReference<Image> availableImage = new AtomicReference<>();
-                final AtomicReference<Image> unavailableImage = new AtomicReference<>();
-                final Subscription replay = aeron.addSubscription(
-                    replayChannel, replayStreamId, availableImage::set, unavailableImage::set);
-
-                Image image;
-                while (null == (image = availableImage.get()))
+                try
                 {
-                    aeronArchive.checkForErrorResponse();
-                    Tests.yield();
-                }
+                    aeronArchive.startReplay(
+                        recordingId,
+                        IPC_CHANNEL,
+                        replayStreamId,
+                        new ReplayParams().position(NULL_POSITION).length(REPLAY_ALL_AND_STOP));
 
-                while (!image.isEndOfStream())
-                {
-                    aeronArchive.checkForErrorResponse();
-                    Tests.yield();
+                    fail("Should have thrown exception");
                 }
-
-                while (image != unavailableImage.get())
+                catch (final ArchiveException ex)
                 {
-                    Tests.yield();
+                    assertEquals(ArchiveException.EMPTY_RECORDING, ex.errorCode());
                 }
-
-                CloseHelper.quietClose(replay);
-                aeronArchive.stopReplay(replaySessionId);
             }
         }
     }