Repository: aeron-io/aeron (language: java)

Issue:
[AeronArchive] Allow disabling UDP aeron.archive.control.response.channel 
Hi,

The defaul value for `aeron.archive.control.response.channel` creates a localhost subscription. This is in addition to an `aeron:ipc` one that is also created by default.

Equally it doesn't seem to be possible to disable the UDP channel altogether. Set `aeron.archive.control.response.channel` to something non-udpish makes it error out.

This is an issue because a localhost socket can't be accelerated by most kernel bypass LD_PRELOAD solutions which in turn means that the default forces the Aeron receive thread to poll the system even if kernel bypassed.

My understanding of aeron archive might be lacking but I don't understand why you'd want a localhost subscription in addition to an `ipc` one anyway. 


Discussion / hints:
I think you are looking at an older version.  The latest release removes the default, requiring it to be set.  Because, as you mention, a default of localhost does not make a lot of sense.  However, it does need to be set, but I can see the value of an IPC-only archive configuration so that the UDP publication and subscription resources are not created.  I think this should be an explicit setup/configuration option.  When archive is used with cluster there are cases where remote access between archives is required, so defaulting it to IPC-only could break for some people.
 > The latest release removes the default, requiring it to be set.

I see, https://github.com/real-logic/aeron/wiki/Configuration-Options#aeron-archive-client-options probably needs some updating to reflect that.

 > but I can see the value of an IPC-only archive configuration so that the UDP publication and subscription resources are not created

Yep, that's what I am asking for :)

Fix patch:
diff --git a/aeron-archive/src/main/java/io/aeron/archive/Archive.java b/aeron-archive/src/main/java/io/aeron/archive/Archive.java
index fa6372b9f2..09c0e7d56a 100644
--- a/aeron-archive/src/main/java/io/aeron/archive/Archive.java
+++ b/aeron-archive/src/main/java/io/aeron/archive/Archive.java
@@ -503,6 +503,14 @@ public static final class Configuration
          */
         public static final String ARCHIVE_ID_PROP_NAME = "aeron.archive.id";
 
+        /**
+         * Is control channel enabled. Defaults to {@code true}.
+         * <p>
+         * If set to anything other than {@code true} then control channel is disabled, i.e. the Archive will run in
+         * IPC-only mode.
+         */
+        public static final String CONTROL_CHANNEL_ENABLED_PROP_NAME = "aeron.archive.control.channel.enabled";
+
         /**
          * Update interval in ms for archive mark file.
          */
@@ -763,7 +771,7 @@ public static long replayerCycleThresholdNs()
          */
         public static boolean deleteArchiveOnStart()
         {
-            return "true".equalsIgnoreCase(getProperty(ARCHIVE_DIR_DELETE_ON_START_PROP_NAME, "false"));
+            return "true".equals(getProperty(ARCHIVE_DIR_DELETE_ON_START_PROP_NAME, "false"));
         }
 
         /**
@@ -862,6 +870,17 @@ public static String replayChecksum()
         {
             return getProperty(REPLAY_CHECKSUM_PROP_NAME);
         }
+
+        /**
+         * Should the control channel be enabled.
+         *
+         * @return {@code true} if the control channel to be enabled.
+         * @see #CONTROL_CHANNEL_ENABLED_PROP_NAME
+         */
+        public static boolean controlChannelEnabled()
+        {
+            return "true".equals(System.getProperty(CONTROL_CHANNEL_ENABLED_PROP_NAME, "true"));
+        }
     }
 
     /**
@@ -893,6 +912,7 @@ public static final class Context implements Cloneable
         private AeronArchive.Context archiveClientContext;
         private AgentInvoker mediaDriverAgentInvoker;
 
+        private boolean controlChannelEnabled = Configuration.controlChannelEnabled();
         private String controlChannel = AeronArchive.Configuration.controlChannel();
         private int controlStreamId = AeronArchive.Configuration.controlStreamId();
         private String localControlChannel = AeronArchive.Configuration.localControlChannel();
@@ -999,15 +1019,18 @@ public void conclude()
             io.aeron.driver.Configuration.validateMtuLength(controlMtuLength);
             checkTermLength(controlTermBufferLength);
 
-            if (null == controlChannel)
+            if (controlChannelEnabled)
             {
-                throw new ConfigurationException("Archive.Context.controlChannel must be set");
-            }
+                if (null == controlChannel)
+                {
+                    throw new ConfigurationException("Archive.Context.controlChannel must be set");
+                }
 
-            if (!controlChannel.startsWith(CommonContext.UDP_CHANNEL))
-            {
-                throw new ConfigurationException(
-                    "Archive.Context.controlChannel must be UDP media: uri=" + controlChannel);
+                if (!controlChannel.startsWith(CommonContext.UDP_CHANNEL))
+                {
+                    throw new ConfigurationException(
+                        "Archive.Context.controlChannel must be UDP media: uri=" + controlChannel);
+                }
             }
 
             if (!localControlChannel.startsWith(CommonContext.IPC_CHANNEL))
@@ -1288,26 +1311,33 @@ else if (segmentFileLength < TERM_MIN_LENGTH || segmentFileLength > TERM_MAX_LEN
 
             if (null == archiveClientContext.controlResponseChannel())
             {
-                final ChannelUri controlChannelUri = ChannelUri.parse(controlChannel);
-                final String endpoint = controlChannelUri.get(ENDPOINT_PARAM_NAME);
-                int separatorIndex = -1;
-
-                if (null == endpoint || -1 == (separatorIndex = endpoint.lastIndexOf(':')))
+                if (controlChannelEnabled)
                 {
-                    throw new ConfigurationException(
-                        "Unable to derive Archive.Context.archiveClientContext.controlResponseChannel as " +
-                        "Archive.Context.controlChannel.endpoint=" + endpoint +
-                        " and is not in the <host>:<port> format");
+                    final ChannelUri controlChannelUri = ChannelUri.parse(controlChannel);
+                    final String endpoint = controlChannelUri.get(ENDPOINT_PARAM_NAME);
+                    int separatorIndex = -1;
 
-                }
+                    if (null == endpoint || -1 == (separatorIndex = endpoint.lastIndexOf(':')))
+                    {
+                        throw new ConfigurationException(
+                            "Unable to derive Archive.Context.archiveClientContext.controlResponseChannel as " +
+                                "Archive.Context.controlChannel.endpoint=" + endpoint +
+                                " and is not in the <host>:<port> format");
+                    }
 
-                final String responseEndpoint = endpoint.substring(0, separatorIndex) + ":0";
-                final String responseChannel = new ChannelUriStringBuilder()
-                    .media("udp")
-                    .endpoint(responseEndpoint)
-                    .build();
+                    final String responseEndpoint = endpoint.substring(0, separatorIndex) + ":0";
+                    final String responseChannel = new ChannelUriStringBuilder()
+                        .media("udp")
+                        .endpoint(responseEndpoint)
+                        .build();
 
-                archiveClientContext.controlResponseChannel(responseChannel);
+                    archiveClientContext.controlResponseChannel(responseChannel);
+                }
+                else
+                {
+                    throw new ConfigurationException("Archive.Context.archiveClientContext.controlResponseChannel " +
+                        "must be set if Archive.Context.controlChannelEnabled is false");
+                }
             }
 
             archiveClientContext.aeron(aeron).lock(NoOpLock.INSTANCE).errorHandler(errorHandler);
@@ -1415,7 +1445,7 @@ public Context deleteArchiveOnStart(final boolean deleteArchiveOnStart)
         /**
          * Should an existing archive be deleted on start. Useful only for testing.
          *
-         * @return true if an existing archive should be deleted on start up.
+         * @return {@code true} if an existing archive should be deleted on start up.
          */
         public boolean deleteArchiveOnStart()
         {
@@ -1552,6 +1582,30 @@ public AeronArchive.Context archiveClientContext()
             return archiveClientContext;
         }
 
+        /**
+         * Should the UDP control channel be enabled.
+         *
+         * @return {@code true} if the UDP control channel should be enabled.
+         * @see Configuration#CONTROL_CHANNEL_ENABLED_PROP_NAME
+         */
+        public boolean controlChannelEnabled()
+        {
+            return controlChannelEnabled;
+        }
+
+        /**
+         * Set if the UDP control channel should be enabled.
+         *
+         * @param controlChannelEnabled indication of if the recording events channel should be enabled.
+         * @return this for a fluent API.
+         * @see Configuration#CONTROL_CHANNEL_ENABLED_PROP_NAME
+         */
+        public Context controlChannelEnabled(final boolean controlChannelEnabled)
+        {
+            this.controlChannelEnabled = controlChannelEnabled;
+            return this;
+        }
+
         /**
          * Get the channel URI on which the control request subscription will listen.
          *
@@ -1651,7 +1705,7 @@ public Context controlTermBufferSparse(final boolean controlTermBufferSparse)
         /**
          * Should the control streams use sparse file term buffers.
          *
-         * @return true if the control stream should use sparse file term buffers.
+         * @return {@code true} if the control stream should use sparse file term buffers.
          * @see io.aeron.archive.client.AeronArchive.Configuration#CONTROL_TERM_BUFFER_SPARSE_PROP_NAME
          */
         public boolean controlTermBufferSparse()
@@ -1775,7 +1829,7 @@ public Context recordingEventsStreamId(final int recordingEventsStreamId)
         /**
          * Should the recording events channel be enabled.
          *
-         * @return true if the recording events channel should be enabled.
+         * @return {@code true} if the recording events channel should be enabled.
          * @see io.aeron.archive.client.AeronArchive.Configuration#RECORDING_EVENTS_ENABLED_PROP_NAME
          */
         public boolean recordingEventsEnabled()
diff --git a/aeron-archive/src/main/java/io/aeron/archive/ArchiveConductor.java b/aeron-archive/src/main/java/io/aeron/archive/ArchiveConductor.java
index e5a30fa0e4..00891aad69 100644
--- a/aeron-archive/src/main/java/io/aeron/archive/ArchiveConductor.java
+++ b/aeron-archive/src/main/java/io/aeron/archive/ArchiveConductor.java
@@ -151,9 +151,17 @@ abstract class ArchiveConductor
         recordingEventsProxy = ctx.recordingEventsEnabled() ? new RecordingEventsProxy(
             aeron.addExclusivePublication(ctx.recordingEventsChannel(), ctx.recordingEventsStreamId())) : null;
 
-        final ChannelUri controlChannelUri = ChannelUri.parse(ctx.controlChannel());
-        controlChannelUri.put(CommonContext.SPARSE_PARAM_NAME, Boolean.toString(ctx.controlTermBufferSparse()));
-        controlSubscription = aeron.addSubscription(controlChannelUri.toString(), ctx.controlStreamId(), this, null);
+        if (ctx.controlChannelEnabled())
+        {
+            final ChannelUri controlChannelUri = ChannelUri.parse(ctx.controlChannel());
+            controlChannelUri.put(CommonContext.SPARSE_PARAM_NAME, Boolean.toString(ctx.controlTermBufferSparse()));
+            controlSubscription = aeron.addSubscription(
+                controlChannelUri.toString(), ctx.controlStreamId(), this, null);
+        }
+        else
+        {
+            controlSubscription = null;
+        }
         localControlSubscription = aeron.addSubscription(
             ctx.localControlChannel(), ctx.localControlStreamId(), this, null);
     }
diff --git a/aeron-archive/src/main/java/io/aeron/archive/ArchiveMarkFile.java b/aeron-archive/src/main/java/io/aeron/archive/ArchiveMarkFile.java
index d0dcfb772d..8f5b6afe8c 100644
--- a/aeron-archive/src/main/java/io/aeron/archive/ArchiveMarkFile.java
+++ b/aeron-archive/src/main/java/io/aeron/archive/ArchiveMarkFile.java
@@ -319,7 +319,7 @@ private static int alignedTotalFileLength(final Archive.Context ctx)
         final int headerLength =
             MarkFileHeaderEncoder.BLOCK_LENGTH +
             (4 * VarAsciiEncodingEncoder.lengthEncodingLength()) +
-            ctx.controlChannel().length() +
+            (null != ctx.controlChannel() ? ctx.controlChannel().length() : 0) +
             ctx.localControlChannel().length() +
             (null != ctx.recordingEventsChannel() ? ctx.recordingEventsChannel().length() : 0) +
             ctx.aeronDirectoryName().length();
diff --git a/aeron-archive/src/main/java/io/aeron/archive/client/AeronArchive.java b/aeron-archive/src/main/java/io/aeron/archive/client/AeronArchive.java
index f2acd33053..1194c3e15d 100644
--- a/aeron-archive/src/main/java/io/aeron/archive/client/AeronArchive.java
+++ b/aeron-archive/src/main/java/io/aeron/archive/client/AeronArchive.java
@@ -849,7 +849,7 @@ public void stopRecording(final String channel, final int streamId)
      *
      * @param channel  to stop recording for.
      * @param streamId to stop recording for.
-     * @return true if the recording was stopped or false if the subscription is not currently active.
+     * @return {@code true} if the recording was stopped or false if the subscription is not currently active.
      */
     public boolean tryStopRecording(final String channel, final int streamId)
     {
@@ -910,7 +910,7 @@ public void stopRecording(final long subscriptionId)
      * {@link #extendRecording(long, String, int, SourceLocation)}.
      *
      * @param subscriptionId is the {@link Subscription#registrationId()} for the recording in the archive.
-     * @return true if the recording was stopped or false if the subscription is not currently active.
+     * @return {@code true} if the recording was stopped or false if the subscription is not currently active.
      */
     public boolean tryStopRecording(final long subscriptionId)
     {
@@ -939,7 +939,7 @@ public boolean tryStopRecording(final long subscriptionId)
      * Try stop an active recording by its recording id.
      *
      * @param recordingId for which active recording should be stopped.
-     * @return true if the recording was stopped or false if the recording is not currently active.
+     * @return {@code true} if the recording was stopped or false if the recording is not currently active.
      */
     public boolean tryStopRecordingByIdentity(final long recordingId)
     {
@@ -2026,7 +2026,7 @@ public void stopReplication(final long replicationId)
      * Attempt to stop a replication session by id returned from {@link #replicate(long, long, int, String, String)}.
      *
      * @param replicationId to stop replication for.
-     * @return true if the replication was stopped, false if the replication is not active.
+     * @return {@code true} if the replication was stopped, false if the replication is not active.
      * @see #replicate(long, long, int, String, String)
      */
     public boolean tryStopReplication(final long replicationId)
@@ -2677,13 +2677,14 @@ public static long messageTimeoutNs()
         /**
          * Should term buffer files be sparse for control request and response streams.
          *
-         * @return true if term buffer files should be sparse for control request and response streams.
+         * @return {@code true} if term buffer files should be sparse for control request and response streams.
          * @see #CONTROL_TERM_BUFFER_SPARSE_PROP_NAME
          */
         public static boolean controlTermBufferSparse()
         {
-            final String propValue = System.getProperty(CONTROL_TERM_BUFFER_SPARSE_PROP_NAME);
-            return null != propValue ? "true".equals(propValue) : CONTROL_TERM_BUFFER_SPARSE_DEFAULT;
+            final String propValue = System.getProperty(
+                CONTROL_TERM_BUFFER_SPARSE_PROP_NAME, Boolean.toString(CONTROL_TERM_BUFFER_SPARSE_DEFAULT));
+            return "true".equals(propValue);
         }
 
         /**
@@ -2801,13 +2802,14 @@ public static int recordingEventsStreamId()
         /**
          * Should the recording events stream be enabled.
          *
-         * @return true if the recording events stream be enabled.
+         * @return {@code true} if the recording events stream be enabled.
          * @see #RECORDING_EVENTS_ENABLED_PROP_NAME
          */
         public static boolean recordingEventsEnabled()
         {
-            final String propValue = System.getProperty(RECORDING_EVENTS_ENABLED_PROP_NAME);
-            return null != propValue ? Boolean.parseBoolean(propValue) : RECORDING_EVENTS_ENABLED_DEFAULT;
+            final String propValue = System.getProperty(
+                RECORDING_EVENTS_ENABLED_PROP_NAME, Boolean.toString(RECORDING_EVENTS_ENABLED_DEFAULT));
+            return "true".equals(propValue);
         }
     }
 
@@ -3096,7 +3098,7 @@ public Context controlTermBufferSparse(final boolean controlTermBufferSparse)
         /**
          * Should the control streams use sparse file term buffers.
          *
-         * @return true if the control stream should use sparse file term buffers.
+         * @return {@code true} if the control stream should use sparse file term buffers.
          * @see Configuration#CONTROL_TERM_BUFFER_SPARSE_PROP_NAME
          */
         public boolean controlTermBufferSparse()


Test patch:
diff --git a/aeron-archive/src/test/java/io/aeron/archive/ArchiveContextTest.java b/aeron-archive/src/test/java/io/aeron/archive/ArchiveContextTest.java
index 953236b60f..d3f3e32f1e 100644
--- a/aeron-archive/src/test/java/io/aeron/archive/ArchiveContextTest.java
+++ b/aeron-archive/src/test/java/io/aeron/archive/ArchiveContextTest.java
@@ -19,6 +19,7 @@
 import io.aeron.AeronCounters;
 import io.aeron.Counter;
 import io.aeron.RethrowingErrorHandler;
+import io.aeron.archive.client.AeronArchive;
 import io.aeron.exceptions.ConfigurationException;
 import io.aeron.security.AuthorisationService;
 import io.aeron.security.AuthorisationServiceSupplier;
@@ -32,6 +33,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.ArgumentCaptor;
 
@@ -624,6 +626,94 @@ void shouldValidateErrorBufferLengthSetViaSystemProperty(final int errorBufferLe
         }
     }
 
+    @Test
+    void controlChannelEnabledReturnsTrueWhenPropertyIsNotSet()
+    {
+        System.clearProperty(CONTROL_CHANNEL_ENABLED_PROP_NAME);
+        assertTrue(Archive.Configuration.controlChannelEnabled());
+    }
+
+    @ParameterizedTest
+    @CsvSource({ "'', false", "true, true", "True, false", "xyz, false" })
+    void controlChannelEnabledReturnsTrueWhenPropertyIsNotSet(final String propValue, final boolean expected)
+    {
+        System.setProperty(CONTROL_CHANNEL_ENABLED_PROP_NAME, propValue);
+        try
+        {
+            assertEquals(expected, Archive.Configuration.controlChannelEnabled());
+        }
+        finally
+        {
+            System.clearProperty(CONTROL_CHANNEL_ENABLED_PROP_NAME);
+        }
+    }
+
+    @Test
+    void controlChannelMustBeSpecified()
+    {
+        context.controlChannel(null);
+
+        final ConfigurationException exception = assertThrowsExactly(ConfigurationException.class, context::conclude);
+        assertEquals("ERROR - Archive.Context.controlChannel must be set", exception.getMessage());
+    }
+
+    @Test
+    void controlChannelMustBeUdpChannel()
+    {
+        context.controlChannel("aeron:ipc");
+
+        final ConfigurationException exception = assertThrowsExactly(ConfigurationException.class, context::conclude);
+        assertEquals(
+            "ERROR - Archive.Context.controlChannel must be UDP media: uri=" + context.controlChannel(),
+            exception.getMessage());
+    }
+
+    @Test
+    void controlChannelMustHaveValidEndpointSpecifiedIfControlResponseChannelOfTheReplicationClientIsNotSet()
+    {
+        context.controlChannel("aeron:udp?endpoint=localhost");
+
+        final ConfigurationException exception = assertThrowsExactly(ConfigurationException.class, context::conclude);
+        assertEquals(
+            "ERROR - Unable to derive Archive.Context.archiveClientContext.controlResponseChannel as " +
+            "Archive.Context.controlChannel.endpoint=localhost and is not in the <host>:<port> format",
+            exception.getMessage());
+    }
+
+    @Test
+    void whenControlChannelIsDisabledTheControlResponseChannelOnTheReplicationClientMustBeSet()
+    {
+        context.controlChannelEnabled(false);
+        context.controlChannel("rubbish");
+        final AeronArchive.Context archiveClientContext = new AeronArchive.Context();
+        archiveClientContext.controlResponseChannel(null);
+        context.archiveClientContext(archiveClientContext);
+
+        final ConfigurationException exception = assertThrowsExactly(ConfigurationException.class, context::conclude);
+        assertEquals(
+            "ERROR - Archive.Context.archiveClientContext.controlResponseChannel must be set if " +
+            "Archive.Context.controlChannelEnabled is false",
+            exception.getMessage());
+    }
+
+    @Test
+    void controlChannelCanBeDisabled()
+    {
+        context.controlChannelEnabled(false);
+        context.controlChannel(null);
+        final AeronArchive.Context archiveClientContext = new AeronArchive.Context();
+        final String responseChannel = "aeron:udp?localhost:0";
+        archiveClientContext.controlResponseChannel(responseChannel);
+        context.archiveClientContext(archiveClientContext);
+
+        context.conclude();
+
+        assertFalse(context.controlChannelEnabled());
+        assertNull(context.controlChannel());
+        assertSame(archiveClientContext, context.archiveClientContext());
+        assertEquals(responseChannel, context.archiveClientContext().controlResponseChannel());
+    }
+
     private Counter mockArchiveCounter(
         final long archiveId, final int typeId, final int id, final ArgumentCaptor<DirectBuffer> tempBuffer)
     {
diff --git a/aeron-archive/src/test/java/io/aeron/archive/ArchiveTest.java b/aeron-archive/src/test/java/io/aeron/archive/ArchiveTest.java
index 69cec81d12..7fc51a2275 100644
--- a/aeron-archive/src/test/java/io/aeron/archive/ArchiveTest.java
+++ b/aeron-archive/src/test/java/io/aeron/archive/ArchiveTest.java
@@ -28,6 +28,7 @@
 import io.aeron.archive.status.RecordingPos;
 import io.aeron.driver.MediaDriver;
 import io.aeron.driver.ThreadingMode;
+import io.aeron.logbuffer.BufferClaim;
 import io.aeron.logbuffer.FrameDescriptor;
 import io.aeron.logbuffer.LogBufferDescriptor;
 import io.aeron.protocol.DataHeaderFlyweight;
@@ -535,6 +536,59 @@ void shouldResolveControlResponseEndpointAddress(final String endpoint)
         }
     }
 
+    @Test
+    @InterruptAfter(10)
+    void shouldRunWithControlChannelDisabled() throws IOException
+    {
+        final int streamId = 7;
+        final String channel = "aeron:ipc";
+
+        final MediaDriver.Context driverCtx = new MediaDriver.Context()
+            .dirDeleteOnStart(true)
+            .threadingMode(ThreadingMode.SHARED);
+        final Archive.Context archiveCtx = TestContexts.localhostArchive()
+            .controlChannelEnabled(false)
+            .controlChannel(null)
+            .archiveClientContext(new AeronArchive.Context().controlResponseChannel("aeron:udp?endpoint=localhost:0"))
+            .deleteArchiveOnStart(true)
+            .threadingMode(SHARED);
+
+        try (ArchivingMediaDriver ignore = ArchivingMediaDriver.launch(driverCtx, archiveCtx);
+            AeronArchive archive = AeronArchive.connect(new AeronArchive.Context()
+                .controlRequestChannel(archiveCtx.localControlChannel())
+                .controlResponseChannel(archiveCtx.localControlChannel()));
+            Aeron aeron = Aeron.connect(new Aeron.Context().aeronDirectoryName(driverCtx.aeronDirectoryName())))
+        {
+            final long subscriptionId = archive.startRecording(channel, streamId, LOCAL);
+            try (Publication publication = aeron.addExclusivePublication(channel, streamId))
+            {
+                final int sessionId = publication.sessionId();
+
+                final CountersReader counters = aeron.countersReader();
+                final int counterId = Tests.awaitRecordingCounterId(counters, sessionId);
+
+                final BufferClaim bufferClaim = new BufferClaim();
+                for (int i = 0; i < 111; i++)
+                {
+                    while (publication.tryClaim(4, bufferClaim) < 0)
+                    {
+                        Tests.yield();
+                    }
+                    bufferClaim.buffer().putInt(bufferClaim.offset(), i);
+                    bufferClaim.commit();
+                }
+
+                Tests.awaitPosition(counters, counterId, publication.position());
+            }
+            archive.stopRecording(subscriptionId);
+        }
+        finally
+        {
+            archiveCtx.deleteDirectory();
+            driverCtx.deleteDirectory();
+        }
+    }
+
     private static Catalog openCatalog(final String archiveDirectoryName)
     {
         final IntConsumer intConsumer = (version) -> {};
diff --git a/aeron-system-tests/src/test/java/io/aeron/archive/ReplicateRecordingTest.java b/aeron-system-tests/src/test/java/io/aeron/archive/ReplicateRecordingTest.java
index 2fee6b5ad5..2e9e427af1 100644
--- a/aeron-system-tests/src/test/java/io/aeron/archive/ReplicateRecordingTest.java
+++ b/aeron-system-tests/src/test/java/io/aeron/archive/ReplicateRecordingTest.java
@@ -48,6 +48,7 @@
 import java.util.concurrent.TimeUnit;
 
 import static io.aeron.Aeron.NULL_VALUE;
+import static io.aeron.CommonContext.IPC_CHANNEL;
 import static io.aeron.CommonContext.generateRandomDirName;
 import static io.aeron.archive.ArchiveSystemTests.*;
 import static io.aeron.archive.client.AeronArchive.NULL_POSITION;
@@ -1056,6 +1057,106 @@ public void shouldReplicateStoppedRecordingOverAnExistingTruncatedRecordingRepla
         assertNotEquals(dstRecording.stopTimestamp(), replicatedRecording.stopTimestamp());
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = { true, false })
+    @InterruptAfter(10)
+    void shouldReplicateLiveRecordingAndMergeIntoADestinationIpcOnlyArchive(final boolean useParams)
+    {
+        final String messagePrefix = "Message-Prefix-";
+        final int messageCount = 10;
+        final long srcRecordingId;
+
+        CloseHelper.closeAll(dstAeronArchive, dstAeron, dstArchive, dstDriver);
+
+        final long subscriptionId = srcAeronArchive.startRecording(LIVE_CHANNEL, LIVE_STREAM_ID, LOCAL);
+
+        dstDriver = TestMediaDriver.launch(new MediaDriver.Context()
+                .aeronDirectoryName(generateRandomDirName())
+                .termBufferSparseFile(true)
+                .threadingMode(ThreadingMode.SHARED)
+                .spiesSimulateConnection(true)
+                .timerIntervalNs(TIMER_INTERVAL_NS)
+                .dirDeleteOnStart(true),
+            systemTestWatcher);
+        systemTestWatcher.dataCollector().add(dstDriver.context().aeronDirectory());
+
+        dstAeron = Aeron.connect(new Aeron.Context().aeronDirectoryName(dstDriver.aeronDirectoryName()));
+
+        dstArchive = Archive.launch(new Archive.Context()
+            .catalogCapacity(CATALOG_CAPACITY)
+            .aeronDirectoryName(dstDriver.aeronDirectoryName())
+            .controlChannel(null)
+            .controlChannelEnabled(false)
+            .localControlChannel(IPC_CHANNEL)
+            .localControlStreamId(4444)
+            .archiveClientContext(
+                new AeronArchive.Context().controlResponseChannel("aeron:udp?endpoint=localhost:0"))
+            .recordingEventsEnabled(false)
+            .replicationChannel("aeron:udp?endpoint=localhost:0")
+            .deleteArchiveOnStart(true)
+            .archiveDir(new File(SystemUtil.tmpDirName(), "ipc-archive"))
+            .fileSyncLevel(0)
+            .threadingMode(ArchiveThreadingMode.SHARED));
+        systemTestWatcher.dataCollector().add(dstArchive.context().archiveDir());
+
+        dstAeronArchive = AeronArchive.connect(new AeronArchive.Context()
+            .controlRequestChannel(dstArchive.context().localControlChannel())
+            .controlRequestStreamId(dstArchive.context().localControlStreamId())
+            .controlResponseChannel(IPC_CHANNEL)
+            .controlResponseStreamId(5555)
+            .idleStrategy(YieldingIdleStrategy.INSTANCE)
+            .aeron(dstAeron));
+
+        dstRecordingSignalConsumer = injectRecordingSignalConsumer(dstAeronArchive);
+
+        srcRecordingSignalConsumer.reset();
+        try (Publication publication = srcAeron.addPublication(LIVE_CHANNEL, LIVE_STREAM_ID))
+        {
+            awaitSignal(srcAeronArchive, srcRecordingSignalConsumer, RecordingSignal.START);
+            final long signaledRecordingId = srcRecordingSignalConsumer.recordingId;
+            final CountersReader srcCounters = srcAeron.countersReader();
+            final int counterId = Tests.awaitRecordingCounterId(srcCounters, publication.sessionId());
+            srcRecordingId = RecordingPos.getRecordingId(srcCounters, counterId);
+            assertEquals(srcRecordingId, signaledRecordingId);
+
+            offer(publication, messageCount, messagePrefix);
+            Tests.awaitPosition(srcCounters, counterId, publication.position());
+
+            dstRecordingSignalConsumer.reset();
+            if (useParams)
+            {
+                dstAeronArchive.replicate(
+                    srcRecordingId,
+                    SRC_CONTROL_STREAM_ID,
+                    SRC_CONTROL_REQUEST_CHANNEL,
+                    new ReplicationParams().liveDestination(LIVE_CHANNEL));
+            }
+            else
+            {
+                dstAeronArchive.replicate(
+                    srcRecordingId, NULL_VALUE, SRC_CONTROL_STREAM_ID, SRC_CONTROL_REQUEST_CHANNEL, LIVE_CHANNEL);
+            }
+
+            offer(publication, messageCount, messagePrefix);
+
+            awaitSignal(dstAeronArchive, dstRecordingSignalConsumer, RecordingSignal.REPLICATE);
+            final long dstRecordingId = dstRecordingSignalConsumer.recordingId;
+            resetAndAwaitSignal(dstAeronArchive, dstRecordingSignalConsumer, dstRecordingId, RecordingSignal.EXTEND);
+            resetAndAwaitSignal(dstAeronArchive, dstRecordingSignalConsumer, dstRecordingId, RecordingSignal.MERGE);
+
+            final CountersReader dstCounters = dstAeron.countersReader();
+            final int dstCounterId =
+                RecordingPos.findCounterIdByRecording(dstCounters, dstRecordingId);
+
+            offer(publication, messageCount, messagePrefix);
+            Tests.awaitPosition(dstCounters, dstCounterId, publication.position());
+        }
+
+        srcRecordingSignalConsumer.reset();
+        srcAeronArchive.stopRecording(subscriptionId);
+        awaitSignal(srcAeronArchive, srcRecordingSignalConsumer, srcRecordingId, RecordingSignal.STOP);
+    }
+
     private void readRecordingIntoBuffer(final long srcRecordingId, final ExpandableArrayBuffer srcRecordingData)
     {
         final RecordingDescriptorCollector collector = new RecordingDescriptorCollector(10);