Merge branch 'cassandra-6.0' into trunk

CHANGES.txt

@@ -3,6 +3,8 @@
  * Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767)
  * Add a guardrail for misprepared statements (CASSANDRA-21139)
 Merged from 6.0:
+ * Add policy for selecting CMS host when submitting commit request (CASSANDRA-21456)
+ * Fix TCM log catchup from peer with snapshots and gaps in the log sequence (CASSANDRA-21455)
  * Speed up nodetool doc generation by producing all command help in a single jvm (CASSANDRA-21444)
  * Always send TCM commit failures as Messaging failures (CASSANDRA-21457)
  * Fix ReadCommand serializedSize() using incorrect epoch (CASSANDRA-21438)

src/java/org/apache/cassandra/config/Config.java

@@ -191,6 +191,7 @@ public class Config
     public volatile DurationSpec.IntMillisecondsBound cms_default_max_retry_backoff = null;
     public String cms_retry_delay = "50ms*attempts <= 500ms ... 100ms*attempts <= 1s,retries=10";
 
+    public volatile CMSCommitMemberPreferencePolicy cms_commit_member_preference_policy = CMSCommitMemberPreferencePolicy.random;
     public volatile int epoch_aware_debounce_inflight_tracker_max_size = 100;
 
     /**
@@ -1256,6 +1257,31 @@ public class Config
         group
     }
 
+    /**
+     * Strategy for selecting which CMS member to contact for TCM commits.
+     */
+    public enum CMSCommitMemberPreferencePolicy
+    {
+        /** Shuffle candidates randomly (original behavior) */
+        random,
+
+        /** Shuffle local DC candidates randomly, followed by shuffled non-local */
+        local_random,
+
+        /** Sort by address - all nodes converge on same member with the goal
+         *  of reducing Paxos contention, but will increase hot-spotting on the
+         *  determined member
+         */
+        deterministic,
+
+        /** Sort local members first by name, then non-local. Nodes in each DC
+         *  will converge on same member with the goal of reducing Paxos contention d
+         *  below the random but with lower latency, though higher contention than
+         *  globally deterministic.
+         */
+        local_deterministic
+    }
+
     public enum FlushCompression
     {
         none,
@@ -1562,4 +1588,4 @@ public class Config
      * 6.0 and later.
      */
     public volatile boolean gossip_quarantine_disabled = false;
-}
+}

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -175,6 +175,7 @@ import static org.apache.cassandra.io.util.FileUtils.ONE_GIB;
 import static org.apache.cassandra.io.util.FileUtils.ONE_MIB;
 import static org.apache.cassandra.journal.Params.FlushMode.PERIODIC;
 import static org.apache.cassandra.utils.Clock.Global.logInitializationOutcome;
+import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
 
 public class DatabaseDescriptor
 {
@@ -6092,6 +6093,21 @@ public class DatabaseDescriptor
         return conf.cms_await_timeout;
     }
 
+    public static Config.CMSCommitMemberPreferencePolicy getCmsCommitMemberPreferencePolicy()
+    {
+        return conf.cms_commit_member_preference_policy;
+    }
+
+    public static void setCmsCommitMemberPreferencePolicy(Config.CMSCommitMemberPreferencePolicy policy)
+    {
+        conf.cms_commit_member_preference_policy = policy;
+    }
+
+    public static void setCmsCommitMemberPreferencePolicy(String policy)
+    {
+        setCmsCommitMemberPreferencePolicy(Config.CMSCommitMemberPreferencePolicy.valueOf(toLowerCaseLocalized(policy)));
+    }
+
     public static int getEpochAwareDebounceInFlightTrackerMaxSize()
     {
         return conf.epoch_aware_debounce_inflight_tracker_max_size;

src/java/org/apache/cassandra/service/StorageService.java

@@ -1762,6 +1762,19 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         return pojoMapToString(snapshotAsMap, format);
     }
 
+    @Override
+    public String getCmsCommitMemberPreferencePolicy()
+    {
+        return DatabaseDescriptor.getCmsCommitMemberPreferencePolicy().name();
+    }
+
+    @Override
+    public void setCmsCommitMemberPreferencePolicy(String policy)
+    {
+        DatabaseDescriptor.setCmsCommitMemberPreferencePolicy(policy);
+        logger.info("Set cms_commit_member_preference_policy to {}", policy);
+    }
+
     public Map<String,List<Integer>> getConcurrency(List<String> stageNames)
     {
         Stream<Stage> stageStream = stageNames.isEmpty() ? stream(Stage.values()) : stageNames.stream().map(Stage::fromPoolName);

src/java/org/apache/cassandra/service/StorageServiceMBean.java

@@ -1176,6 +1176,18 @@ public interface StorageServiceMBean extends NotificationEmitter
     List<String> getAccordManagedKeyspaces();
     List<String> getAccordManagedTables();
 
+    /** Get the CMS commit member preference policy
+     *
+     * @return how to choose the cms member preference order for commits
+     */
+    public String getCmsCommitMemberPreferencePolicy();
+
+    /** Update the CMS commit member preference policy
+     *
+     * @param policy see Config.CMSCommitMemberPreferencePolicy
+     */
+    public void setCmsCommitMemberPreferencePolicy(String policy);
+
     /** Gets the concurrency settings for processing stages*/
     static class StageConcurrency implements Serializable
     {

src/java/org/apache/cassandra/tcm/RemoteProcessor.java

@@ -31,16 +31,19 @@ import java.util.function.Consumer;
 import java.util.function.Supplier;
 
 import com.codahale.metrics.Timer;
+import com.google.common.annotations.VisibleForTesting;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.concurrent.ScheduledExecutors;
+import org.apache.cassandra.config.Config;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.exceptions.RequestFailure;
 import org.apache.cassandra.exceptions.RequestFailureReason;
 import org.apache.cassandra.gms.FailureDetector;
 import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.Locator;
 import org.apache.cassandra.metrics.TCMMetrics;
 import org.apache.cassandra.net.Message;
 import org.apache.cassandra.net.MessageDelivery;
@@ -52,6 +55,7 @@ import org.apache.cassandra.tcm.Discovery.DiscoveredNodes;
 import org.apache.cassandra.tcm.log.Entry;
 import org.apache.cassandra.tcm.log.LocalLog;
 import org.apache.cassandra.tcm.log.LogState;
+import org.apache.cassandra.tcm.membership.Location;
 import org.apache.cassandra.utils.AbstractIterator;
 import org.apache.cassandra.utils.Clock;
 import org.apache.cassandra.utils.FBUtilities;
@@ -128,11 +132,79 @@ public final class RemoteProcessor implements Processor
             }
         }
 
-        Collections.shuffle(candidates);
+        sortCandidates(candidates);
 
         return candidates;
     }
 
+    private static void sortCandidates(List<InetAddressAndPort> candidates)
+    {
+        Config.CMSCommitMemberPreferencePolicy policy = DatabaseDescriptor.getCmsCommitMemberPreferencePolicy();
+        sortCandidates(candidates, policy, DatabaseDescriptor.getLocator());
+    }
+
+    @VisibleForTesting
+    static void sortCandidates(List<InetAddressAndPort> candidates,
+                               Config.CMSCommitMemberPreferencePolicy policy,
+                               Locator locator)
+    {
+        switch (policy)
+        {
+            case random:
+                Collections.shuffle(candidates);
+                break;
+            case local_random:
+                shuffleLocalDcFirstThenShuffleRest(candidates, locator);
+                break;
+            case deterministic:
+                Collections.sort(candidates);
+                break;
+            case local_deterministic:
+                sortLocalDcFirstThenByAddress(candidates, locator);
+                break;
+            default:
+                throw new IllegalStateException(policy.toString());
+        }
+    }
+
+    @VisibleForTesting
+    static void shuffleLocalDcFirstThenShuffleRest(List<InetAddressAndPort> candidates, Locator locator)
+    {
+        Location local = locator.local();
+
+        List<InetAddressAndPort> localDc = new ArrayList<>();
+        List<InetAddressAndPort> remoteDc = new ArrayList<>();
+
+        for (InetAddressAndPort ep : candidates)
+        {
+            if (local.sameDatacenter(locator.location(ep)))
+                localDc.add(ep);
+            else
+                remoteDc.add(ep);
+        }
+
+        Collections.shuffle(localDc);
+        Collections.shuffle(remoteDc);
+
+        candidates.clear();
+        candidates.addAll(localDc);
+        candidates.addAll(remoteDc);
+    }
+
+    @VisibleForTesting
+    static void sortLocalDcFirstThenByAddress(List<InetAddressAndPort> candidates, Locator locator)
+    {
+        Location local = locator.local();
+
+        candidates.sort((a, b) -> {
+            boolean aLocal = local.sameDatacenter(locator.location(a));
+            boolean bLocal = local.sameDatacenter(locator.location(b));
+            if (aLocal != bLocal)
+                return aLocal ? -1 : 1;
+            return a.compareTo(b);
+        });
+    }
+
     @Override
     public ClusterMetadata fetchLogAndWait(Epoch waitFor, Retry retryPolicy)
     {

src/java/org/apache/cassandra/tcm/log/LogReader.java

@@ -79,7 +79,11 @@ public interface LogReader
             if (snapshotEpochs.size() <= 1 || !allowSnapshots)
             {
                 entries = getEntries(startEpoch);
-                if (entries.isContinuous())
+                // Only return entries directly if they are continuous AND they reach at least as far as any
+                // known snapshot. If a node caught up via a ForceSnapshot (which is not written to local_metadata_log),
+                // its log will have a gap between the old entries and the snapshot epoch. In that case entries may
+                // appear continuous up to their last epoch, but there is a snapshot beyond them that is needed.
+                if (entries.isContinuous() && (snapshotEpochs.isEmpty() || !snapshotEpochs.get(0).isAfter(entries.latestEpoch())))
                     return new LogState(null, entries.immutable());
                 else if (!allowSnapshots)
                     throw new IllegalStateException("Can't construct a continuous log since " + startEpoch + " and inclusion of snapshots is disallowed");
@@ -136,6 +140,11 @@ public interface LogReader
                 entries.add(entry);
         }
 
+        private Epoch latestEpoch()
+        {
+            return entries.isEmpty() ? since : entries.last().epoch;
+        }
+
         private boolean isContinuous()
         {
             Epoch prev = since;

test/distributed/org/apache/cassandra/distributed/test/log/CatchupViaSnapshotTest.java

@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.log;
+
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.Epoch;
+
+import static org.junit.Assert.assertEquals;
+
+public class CatchupViaSnapshotTest extends TestBaseImpl
+{
+    @Test
+    public void catchupViaSnapshotTest() throws Exception
+    {
+        try (Cluster cluster = init(builder().withNodes(3)
+                                             .withConfig(c -> c.set("metadata_snapshot_frequency", "10"))
+                                             .start()))
+        {
+            cluster.schemaChange(withKeyspace("alter keyspace %s with replication = {'class':'SimpleStrategy', 'replication_factor':3}"));
+            cluster.schemaChange(withKeyspace("create table %s.tbl (id int primary key)"));
+
+            // isolate node2 and node3
+            cluster.filters().inbound().from(1).to(2,3).drop();
+
+            for (int i = 0; i < 30; i++)
+                cluster.coordinator(1).execute(withKeyspace("alter table %s.tbl with comment='abc" + i + "'"), ConsistencyLevel.ONE);
+
+            // Snapshot needs to be the last transformation in the log
+            cluster.get(1).nodetoolResult("cms", "snapshot").asserts().success();
+
+            // allow node2 to catch up:
+            cluster.filters().reset();
+            cluster.filters().inbound().from(1).to(3).drop();
+            String node1Address = cluster.get(1).config().broadcastAddress().getHostString();
+
+            fetchLogFromPeerAsync(cluster.get(2), node1Address);
+            // allow node3 to catch up
+            cluster.filters().reset();
+            String node2Address = cluster.get(2).config().broadcastAddress().getHostString();
+            // by now node2 has an incomplete log, it only caught up from the snapshot above
+            // this means the log is continuous, but its current epoch is beyond the last entry in the log
+            fetchLogFromPeerAsync(cluster.get(3), node2Address);
+
+            long expectedEpoch = cluster.get(1).callOnInstance(() -> ClusterMetadata.current().epoch.getEpoch());
+            for (IInvokableInstance i : cluster)
+                assertEquals(expectedEpoch, (long)i.callOnInstance(() -> ClusterMetadata.current().epoch.getEpoch()));
+        }
+    }
+
+    private static void fetchLogFromPeerAsync(IInvokableInstance i, String address)
+    {
+        i.runOnInstance(() -> {
+            try
+            {
+                ClusterMetadataService.instance().fetchLogFromPeerAsync(InetAddressAndPort.getByNameUnchecked(address), Epoch.create(30)).get();
+            }
+            catch (Exception e)
+            {
+                throw new RuntimeException(e);
+            }
+        });
+    }
+}

test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java

@@ -96,6 +96,7 @@ public class DatabaseDescriptorRefTest
     "org.apache.cassandra.config.CassandraRelevantProperties$PropertyConverter",
     "org.apache.cassandra.config.Config",
     "org.apache.cassandra.config.Config$1",
+    "org.apache.cassandra.config.Config$CMSCommitMemberPreferencePolicy",
     "org.apache.cassandra.config.Config$CommitFailurePolicy",
     "org.apache.cassandra.config.Config$CQLStartTime",
     "org.apache.cassandra.config.Config$CommitLogSync",

test/unit/org/apache/cassandra/tcm/RemoteProcessorTest.java

@@ -19,13 +19,19 @@
 package org.apache.cassandra.tcm;
 
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 import org.junit.Test;
 
+import org.apache.cassandra.config.Config;
 import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.Locator;
+import org.apache.cassandra.tcm.membership.Location;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -107,4 +113,164 @@ public class RemoteProcessorTest
         }
         return allEndpoints;
     }
+
+    // ========== CMS Member Selection Tests ==========
+
+    @Test
+    public void testCandidatesDeterministic()
+    {
+        // Create endpoints in non-sorted order
+        List<InetAddressAndPort> candidates = new ArrayList<>(Arrays.asList(
+            InetAddressAndPort.getByNameUnchecked("127.0.0.5"),
+            InetAddressAndPort.getByNameUnchecked("127.0.0.1"),
+            InetAddressAndPort.getByNameUnchecked("127.0.0.9"),
+            InetAddressAndPort.getByNameUnchecked("127.0.0.3")
+        ));
+
+        // Deterministic policy should sort by address
+        RemoteProcessor.sortCandidates(candidates, Config.CMSCommitMemberPreferencePolicy.deterministic, null);
+
+        assertEquals(InetAddressAndPort.getByNameUnchecked("127.0.0.1"), candidates.get(0));
+        assertEquals(InetAddressAndPort.getByNameUnchecked("127.0.0.3"), candidates.get(1));
+        assertEquals(InetAddressAndPort.getByNameUnchecked("127.0.0.5"), candidates.get(2));
+        assertEquals(InetAddressAndPort.getByNameUnchecked("127.0.0.9"), candidates.get(3));
+    }
+
+    @Test
+    public void testCandidatesRandom()
+    {
+        // Create endpoints
+        List<InetAddressAndPort> original = Arrays.asList(
+            InetAddressAndPort.getByNameUnchecked("127.0.0.1"),
+            InetAddressAndPort.getByNameUnchecked("127.0.0.2"),
+            InetAddressAndPort.getByNameUnchecked("127.0.0.3"),
+            InetAddressAndPort.getByNameUnchecked("127.0.0.4")
+        );
+        List<InetAddressAndPort> candidates = new ArrayList<>(original);
+
+        // Random policy should shuffle (but contain same elements)
+        RemoteProcessor.sortCandidates(candidates, Config.CMSCommitMemberPreferencePolicy.random, null);
+
+        // Same elements, possibly different order
+        assertEquals(new HashSet<>(original), new HashSet<>(candidates));
+        assertEquals(original.size(), candidates.size());
+    }
+
+    @Test
+    public void testCandidatesLocalDeterministic()
+    {
+        // DC1 endpoints (local)
+        InetAddressAndPort dc1_1 = InetAddressAndPort.getByNameUnchecked("127.0.0.5");
+        InetAddressAndPort dc1_2 = InetAddressAndPort.getByNameUnchecked("127.0.0.1");
+        // DC2 endpoints (remote)
+        InetAddressAndPort dc2_1 = InetAddressAndPort.getByNameUnchecked("127.0.0.3");
+        InetAddressAndPort dc2_2 = InetAddressAndPort.getByNameUnchecked("127.0.0.2");
+
+        List<InetAddressAndPort> candidates = new ArrayList<>(Arrays.asList(dc2_1, dc1_1, dc2_2, dc1_2));
+
+        // Create a test locator
+        Location dc1 = new Location("DC1", "rack1");
+        Location dc2 = new Location("DC2", "rack1");
+        Map<InetAddressAndPort, Location> locationMap = new HashMap<>();
+        locationMap.put(dc1_1, dc1);
+        locationMap.put(dc1_2, dc1);
+        locationMap.put(dc2_1, dc2);
+        locationMap.put(dc2_2, dc2);
+
+        TestLocator locator = new TestLocator(dc1, locationMap);
+
+        // local_deterministic: local DC sorted first, then remote DC sorted
+        RemoteProcessor.sortCandidates(candidates, Config.CMSCommitMemberPreferencePolicy.local_deterministic, locator);
+
+        // Local DC first (sorted), then remote DC (sorted)
+        assertEquals(dc1_2, candidates.get(0)); // 127.0.0.1 (DC1)
+        assertEquals(dc1_1, candidates.get(1)); // 127.0.0.5 (DC1)
+        assertEquals(dc2_2, candidates.get(2)); // 127.0.0.2 (DC2)
+        assertEquals(dc2_1, candidates.get(3)); // 127.0.0.3 (DC2)
+    }
+
+    @Test
+    public void testCandidatesLocalRandom()
+    {
+        // DC1 endpoints (local)
+        InetAddressAndPort dc1_1 = InetAddressAndPort.getByNameUnchecked("127.0.0.1");
+        InetAddressAndPort dc1_2 = InetAddressAndPort.getByNameUnchecked("127.0.0.2");
+        // DC2 endpoints (remote)
+        InetAddressAndPort dc2_1 = InetAddressAndPort.getByNameUnchecked("127.0.0.3");
+        InetAddressAndPort dc2_2 = InetAddressAndPort.getByNameUnchecked("127.0.0.4");
+
+        List<InetAddressAndPort> candidates = new ArrayList<>(Arrays.asList(dc2_1, dc1_1, dc2_2, dc1_2));
+
+        // Create a test locator
+        Location dc1 = new Location("DC1", "rack1");
+        Location dc2 = new Location("DC2", "rack1");
+        Map<InetAddressAndPort, Location> locationMap = new HashMap<>();
+        locationMap.put(dc1_1, dc1);
+        locationMap.put(dc1_2, dc1);
+        locationMap.put(dc2_1, dc2);
+        locationMap.put(dc2_2, dc2);
+
+        TestLocator locator = new TestLocator(dc1, locationMap);
+
+        // local_random: local DC shuffled first, then remote DC shuffled
+        RemoteProcessor.sortCandidates(candidates, Config.CMSCommitMemberPreferencePolicy.local_random, locator);
+
+        // Local DC should be in first 2 positions, remote DC in last 2
+        Set<InetAddressAndPort> localDcEndpoints = new HashSet<>(Arrays.asList(dc1_1, dc1_2));
+        Set<InetAddressAndPort> remoteDcEndpoints = new HashSet<>(Arrays.asList(dc2_1, dc2_2));
+
+        assertTrue(localDcEndpoints.contains(candidates.get(0)));
+        assertTrue(localDcEndpoints.contains(candidates.get(1)));
+        assertTrue(remoteDcEndpoints.contains(candidates.get(2)));
+        assertTrue(remoteDcEndpoints.contains(candidates.get(3)));
+    }
+
+    @Test
+    public void testDeterministicPolicyProducesSameOrderAcrossCalls()
+    {
+        List<InetAddressAndPort> candidates1 = new ArrayList<>(Arrays.asList(
+            InetAddressAndPort.getByNameUnchecked("127.0.0.5"),
+            InetAddressAndPort.getByNameUnchecked("127.0.0.1"),
+            InetAddressAndPort.getByNameUnchecked("127.0.0.3")
+        ));
+        List<InetAddressAndPort> candidates2 = new ArrayList<>(Arrays.asList(
+            InetAddressAndPort.getByNameUnchecked("127.0.0.3"),
+            InetAddressAndPort.getByNameUnchecked("127.0.0.5"),
+            InetAddressAndPort.getByNameUnchecked("127.0.0.1")
+        ));
+
+        RemoteProcessor.sortCandidates(candidates1, Config.CMSCommitMemberPreferencePolicy.deterministic, null);
+        RemoteProcessor.sortCandidates(candidates2, Config.CMSCommitMemberPreferencePolicy.deterministic, null);
+
+        // Both should produce identical ordering regardless of initial order
+        assertEquals(candidates1, candidates2);
+    }
+
+    /**
+     * Test Locator implementation for unit testing
+     */
+    private static class TestLocator extends Locator
+    {
+        private final Location localLocation;
+        private final Map<InetAddressAndPort, Location> locationMap;
+
+        public TestLocator(Location localLocation, Map<InetAddressAndPort, Location> locationMap)
+        {
+            super(null, null, () -> localLocation, null);
+            this.localLocation = localLocation;
+            this.locationMap = locationMap;
+        }
+
+        @Override
+        public Location local()
+        {
+            return localLocation;
+        }
+
+        @Override
+        public Location location(InetAddressAndPort endpoint)
+        {
+            return locationMap.getOrDefault(endpoint, Location.UNKNOWN);
+        }
+    }
 }

test/unit/org/apache/cassandra/tcm/log/LocalStorageLogStateTest.java

@@ -21,13 +21,16 @@ package org.apache.cassandra.tcm.log;
 import java.io.IOException;
 
 import org.junit.BeforeClass;
+import org.junit.Test;
 
 import org.apache.cassandra.ServerTestUtils;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.UntypedResultSet;
 import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.SystemKeyspace;
 import org.apache.cassandra.db.commitlog.CommitLog;
 import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.tcm.Epoch;
 import org.apache.cassandra.tcm.MetadataSnapshots;
@@ -37,6 +40,8 @@ import org.apache.cassandra.tcm.transformations.TriggerSnapshot;
 import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
 import static org.apache.cassandra.db.SystemKeyspace.METADATA_LOG;
 import static org.apache.cassandra.schema.SchemaConstants.SYSTEM_KEYSPACE_NAME;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
 public class LocalStorageLogStateTest extends LogStateTestBase
 {
@@ -109,4 +114,47 @@ public class LocalStorageLogStateTest extends LogStateTestBase
         };
     }
 
+    @Test
+    public void catchUpViaForceSnapshotLeavesGappedLog() throws Exception
+    {
+        // Simulates a non-CMS node that caught up via a ForceSnapshot (real or synthetic).
+        // LocalLog.append(LogState) converts the received baseState into a synthetic ForceSnapshot,
+        // which is processed but not written to local_metadata_log.
+        // The snapshot is stored in metadata_snapshots by MetadataSnapshotListener.
+        // The resulting on-disk state is: entries 1..X in local_metadata_log, snapshot at S sometime after X in
+        // metadata_snapshots, with no entries between X and S.
+        // Any peer requesting log since epoch <= X must receive the snapshot — not just the continuous
+        // run of entries up to X, which would leave it unable to advance past the gap.
+        MetadataSnapshots realSnapshots = new MetadataSnapshots.SystemKeyspaceMetadataSnapshots();
+        LogStateSUT sut = getSystemUnderTest(realSnapshots);
+        sut.cleanup();
+        ColumnFamilyStore.getIfExists(SYSTEM_KEYSPACE_NAME, SystemKeyspace.SNAPSHOT_TABLE_NAME)
+                         .truncateBlockingWithoutSnapshot();
+
+        // insertRegularEntry inserts 2 entries on the first call (epoch 1 and 2) due to the
+        // Epoch.FIRST pre-init entry, then 1 per subsequent call. After 3 calls: epochs 1..4.
+        sut.insertRegularEntry();
+        sut.insertRegularEntry();
+        sut.insertRegularEntry();
+
+        // Simulate ForceSnapshot at epoch 50: snapshot stored, no intermediate log entries written
+        Epoch gapSnapshotEpoch = Epoch.create(50);
+        realSnapshots.storeSnapshot(ClusterMetadataTestHelper.minimalForTesting(Murmur3Partitioner.instance)
+                                                             .forceEpoch(gapSnapshotEpoch));
+
+        // A peer at epoch 3 sees entry [4] which is continuous, but does not bridge to epoch 50.
+        // Must return the snapshot rather than just entry 4, which would leave the peer stuck.
+        LogState state = sut.getLogState(Epoch.create(3));
+        assertEquals(gapSnapshotEpoch, state.baseState.epoch);
+        assertTrue(state.entries.isEmpty());
+
+        // A peer already at epoch 4 (the last log entry) previously got an empty response and stalled.
+        state = sut.getLogState(Epoch.create(4));
+        assertEquals(gapSnapshotEpoch, state.baseState.epoch);
+        assertTrue(state.entries.isEmpty());
+
+        ColumnFamilyStore.getIfExists(SYSTEM_KEYSPACE_NAME, SystemKeyspace.SNAPSHOT_TABLE_NAME)
+                         .truncateBlockingWithoutSnapshot();
+    }
+
 }