diff --git a/CHANGES.txt b/CHANGES.txt
index a3e062f026..49b06ef337 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -6,6 +6,7 @@
  * Allow empty keystore_password in encryption_options (CASSANDRA-18778)
  * Skip ColumnFamilyStore#topPartitions initialization when client or tool mode (CASSANDRA-18697)
 Merged from 4.0:
+ * Remove completed coordinator sessions (CASSANDRA-18903)
  * Make StartupConnectivityChecker only run a connectivity check if there are no nodes which are running a version prior to Cassandra 4 (CASSANDRA-18968)
  * Retrieve keyspaces metadata and schema version concistently in DescribeStatement (CASSANDRA-18921)
  * Gossip NPE due to shutdown event corrupting empty statuses (CASSANDRA-18913)
diff --git a/src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java b/src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java
index 86ecfb76f0..56c50e16e7 100644
--- a/src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java
+++ b/src/java/org/apache/cassandra/repair/consistent/ConsistentSession.java
@@ -206,6 +206,12 @@ public abstract class ConsistentSession
         this.participants = ImmutableSet.copyOf(builder.participants);
     }
 
+    public boolean isCompleted()
+    {
+        State s = getState();
+        return s == State.FINALIZED || s == State.FAILED;
+    }
+
     public State getState()
     {
         return state;
diff --git a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java
index 88f30029c3..761f07c180 100644
--- a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java
+++ b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java
@@ -21,6 +21,7 @@ package org.apache.cassandra.repair.consistent;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Consumer;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
@@ -70,9 +71,12 @@ public class CoordinatorSession extends ConsistentSession
     private volatile long repairStart = Long.MIN_VALUE;
     private volatile long finalizeStart = Long.MIN_VALUE;
 
+    private final Consumer<CoordinatorSession> listener;
+
     public CoordinatorSession(Builder builder)
     {
         super(builder);
+        this.listener = builder.listener;
         for (InetAddressAndPort participant : participants)
         {
             participantStates.put(participant, State.PREPARING);
@@ -81,6 +85,13 @@ public class CoordinatorSession extends ConsistentSession
 
     public static class Builder extends AbstractBuilder
     {
+        Consumer<CoordinatorSession> listener;
+
+        public void withListener(Consumer<CoordinatorSession> listener)
+        {
+            this.listener = listener;
+        }
+
         public CoordinatorSession build()
         {
             validate();
@@ -97,6 +108,8 @@ public class CoordinatorSession extends ConsistentSession
     {
         logger.trace("Setting coordinator state to {} for repair {}", state, sessionID);
         super.setState(state);
+        if (listener != null)
+            listener.accept(this);
     }
 
     @VisibleForTesting
diff --git a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSessions.java b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSessions.java
index 328c16e645..edf8d2bd72 100644
--- a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSessions.java
+++ b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSessions.java
@@ -24,6 +24,9 @@ import java.util.Set;
 
 import com.google.common.base.Preconditions;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.repair.messages.FailSession;
 import org.apache.cassandra.repair.messages.FinalizePromise;
@@ -37,6 +40,7 @@ import org.apache.cassandra.utils.TimeUUID;
  */
 public class CoordinatorSessions
 {
+    private static final Logger logger = LoggerFactory.getLogger(CoordinatorSessions.class);
     private final Map<TimeUUID, CoordinatorSession> sessions = new HashMap<>();
 
     protected CoordinatorSession buildSession(CoordinatorSession.Builder builder)
@@ -62,6 +66,7 @@ public class CoordinatorSessions
         builder.withRepairedAt(prs.repairedAt);
         builder.withRanges(prs.getRanges());
         builder.withParticipants(participants);
+        builder.withListener(this::onSessionStateUpdate);
         CoordinatorSession session = buildSession(builder);
         sessions.put(session.sessionID, session);
         return session;
@@ -72,6 +77,15 @@ public class CoordinatorSessions
         return sessions.get(sessionId);
     }
 
+    public synchronized void onSessionStateUpdate(CoordinatorSession session)
+    {
+        if (session.isCompleted())
+        {
+            logger.info("Removing completed session {} with state {}", session.sessionID, session.getState());
+            sessions.remove(session.sessionID);
+        }
+    }
+
     public void handlePrepareResponse(PrepareConsistentResponse msg)
     {
         CoordinatorSession session = getSession(msg.parentSession);
diff --git a/src/java/org/apache/cassandra/repair/consistent/LocalSession.java b/src/java/org/apache/cassandra/repair/consistent/LocalSession.java
index a6f81d7fe5..204dd2a8b5 100644
--- a/src/java/org/apache/cassandra/repair/consistent/LocalSession.java
+++ b/src/java/org/apache/cassandra/repair/consistent/LocalSession.java
@@ -37,12 +37,6 @@ public class LocalSession extends ConsistentSession
         this.lastUpdate = builder.lastUpdate;
     }
 
-    public boolean isCompleted()
-    {
-        State s = getState();
-        return s == State.FINALIZED || s == State.FAILED;
-    }
-
     public int getStartedAt()
     {
         return startedAt;
diff --git a/test/distributed/org/apache/cassandra/distributed/test/IncRepairCoordinatorErrorTest.java b/test/distributed/org/apache/cassandra/distributed/test/IncRepairCoordinatorErrorTest.java
index 447cd822e2..14a47f1928 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/IncRepairCoordinatorErrorTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/IncRepairCoordinatorErrorTest.java
@@ -46,11 +46,12 @@ public class IncRepairCoordinatorErrorTest extends TestBaseImpl
                    .to(3)
                    .messagesMatching((from, to, msg) -> msg.verb() == FINALIZE_COMMIT_MSG.id).drop();
             cluster.get(1).nodetoolResult("repair", KEYSPACE).asserts().success();
+            assertThat(cluster.get(1).logs().watchFor("Removing completed session .* with state FINALIZED").getResult()).isNotEmpty();
+
             TimeUUID result = (TimeUUID) cluster.get(1).executeInternal("select parent_id from system_distributed.repair_history")[0][0];
             cluster.get(3).runOnInstance(() -> {
                 ActiveRepairService.instance.failSession(result.toString(), true);
             });
-            assertThat(cluster.get(1).logs().watchFor("Can't transition endpoints .* to FAILED").getResult()).isNotEmpty();
         }
     }
 }