Add an ability to reconstruct arbitrary epoch state from the log to TCM

Patch by Alex Petrov; reviewed by Marcus Eriksson for CASSANDRA-19790.

src/java/org/apache/cassandra/metrics/TCMMetrics.java

@@ -66,6 +66,7 @@ public class TCMMetrics
     public final Meter coordinatorBehindSchema;
     public final Meter coordinatorBehindPlacements;
     public final Gauge<Long> epochAwareDebounceTrackerSize;
+    public final Meter reconstructLogStateCall;
 
     private TCMMetrics()
     {
@@ -127,6 +128,7 @@ public class TCMMetrics
 
         coordinatorBehindSchema = Metrics.meter(factory.createMetricName("CoordinatorBehindSchema"));
         coordinatorBehindPlacements = Metrics.meter(factory.createMetricName("CoordinatorBehindPlacements"));
+        reconstructLogStateCall = Metrics.meter(factory.createMetricName("ReconstructLogStateCall"));
     }
 
     public void recordCommitFailureLatency(long latency, TimeUnit timeUnit, boolean isRejection)

src/java/org/apache/cassandra/net/Verb.java

@@ -130,6 +130,7 @@ import org.apache.cassandra.tcm.Discovery;
 import org.apache.cassandra.tcm.Epoch;
 import org.apache.cassandra.tcm.FetchCMSLog;
 import org.apache.cassandra.tcm.FetchPeerLog;
+import org.apache.cassandra.tcm.ReconstructLogState;
 import org.apache.cassandra.tcm.migration.CMSInitializationResponse;
 import org.apache.cassandra.tcm.migration.Election;
 import org.apache.cassandra.tcm.migration.CMSInitializationRequest;
@@ -300,6 +301,8 @@ public enum Verb
     TCM_DISCOVER_REQ       (813, P0, rpcTimeout,      INTERNAL_METADATA,    () -> NoPayload.serializer,                         () -> Discovery.instance.requestHandler,    TCM_DISCOVER_RSP       ),
     TCM_FETCH_PEER_LOG_RSP (818, P0, rpcTimeout,      FETCH_LOG,            MessageSerializers::logStateSerializer,             RESPONSE_HANDLER                                 ),
     TCM_FETCH_PEER_LOG_REQ (819, P0, rpcTimeout,      FETCH_LOG,            () -> FetchPeerLog.serializer,                      () -> FetchPeerLog.Handler.instance,        TCM_FETCH_PEER_LOG_RSP ),
+    TCM_RECONSTRUCT_EPOCH_RSP (820, P0, rpcTimeout,   FETCH_LOG,            MessageSerializers::logStateSerializer,             () -> ResponseVerbHandler.instance                                 ),
+    TCM_RECONSTRUCT_EPOCH_REQ (821, P0, rpcTimeout,   FETCH_LOG,            () -> ReconstructLogState.serializer,               () -> ReconstructLogState.Handler.instance, TCM_FETCH_PEER_LOG_RSP ),
 
     INITIATE_DATA_MOVEMENTS_RSP (814, P1, rpcTimeout, MISC, () -> NoPayload.serializer,             RESPONSE_HANDLER                                  ),
     INITIATE_DATA_MOVEMENTS_REQ (815, P1, rpcTimeout, MISC, () -> DataMovement.serializer,          () -> DataMovementVerbHandler.instance, INITIATE_DATA_MOVEMENTS_RSP ),

src/java/org/apache/cassandra/schema/DistributedMetadataLogKeyspace.java

@@ -163,6 +163,20 @@ public final class DistributedMetadataLogKeyspace
         return (consistentFetch ? serialLogReader : localLogReader).getLogState(since);
     }
 
+    /**
+     * Reconstructs the log state by returning a _consistent_ base snapshot of a start epoch, and
+     * a list of transformations between start and end.
+     *
+     * TODO: this is a rather expensive operation, and should be use sparingly. If we decide we need to
+     *  rely on reconstructing arbitrary epochs during normal operation, we need to add a caching mechanism
+     *  here. One more alternative is to keep a lazily-initialized AccordTopology table on CMS nodes for a
+     *  number of recent epochs, and keep a node-local cache of this table on other nodes.
+     */
+    public static LogState getLogState(Epoch start, Epoch end)
+    {
+        return serialLogReader.getLogState(start, end);
+    }
+
     public static class DistributedTableLogReader implements LogReader
     {
         private final ConsistencyLevel consistencyLevel;
@@ -200,6 +214,27 @@ public final class DistributedMetadataLogKeyspace
             return entryHolder;
         }
 
+        public EntryHolder getEntries(Epoch since, Epoch until) throws IOException
+        {
+            // during gossip upgrade we have epoch = Long.MIN_VALUE + 1 (and the reverse partitioner doesn't support negative keys)
+            since = since.isBefore(Epoch.EMPTY) ? Epoch.EMPTY : since;
+            // note that we want all entries with epoch >= since - but since we use a reverse partitioner, we actually
+            // want all entries where the token is less than token(since)
+            UntypedResultSet resultSet = execute(String.format("SELECT epoch, kind, transformation, entry_id FROM %s.%s WHERE token(epoch) <= token(?) AND token(epoch) >= token(?)",
+                                                               SchemaConstants.METADATA_KEYSPACE_NAME, TABLE_NAME),
+                                                 consistencyLevel, since.getEpoch(), until.getEpoch());
+            EntryHolder entryHolder = new EntryHolder(since);
+            for (UntypedResultSet.Row row : resultSet)
+            {
+                long entryId = row.getLong("entry_id");
+                Epoch epoch = Epoch.create(row.getLong("epoch"));
+                Transformation.Kind kind = Transformation.Kind.fromId(row.getInt("kind"));
+                Transformation transform = kind.fromVersionedBytes(row.getBlob("transformation"));
+                entryHolder.add(new Entry(new Entry.Id(entryId), epoch, transform));
+            }
+            return entryHolder;
+        }
+
         @Override
         public MetadataSnapshots snapshots()
         {

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

@@ -191,7 +191,7 @@ public abstract class AbstractLocalProcessor implements Processor
             // We can use local log here since we always call this method only if local log is up-to-date:
             // in case of a successful commit, we apply against latest metadata locally before committing,
             // and in case of a rejection, we fetch latest entries to verify linearizability.
-            logState = log.getCommittedEntries(lastKnown);
+            logState = log.getLocalEntries(lastKnown);
         }
 
         return logState;

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

@@ -76,6 +76,11 @@ public class AtomicLongBackedProcessor extends AbstractLocalProcessor
         return log.waitForHighestConsecutive();
     }
 
+    public LogState reconstruct(Epoch lowEpoch, Epoch highEpoch, Retry.Deadline retryPolicy)
+    {
+        return log.getLocalEntries(lowEpoch);
+    }
+
     public static class InMemoryStorage implements LogStorage
     {
         private final List<Entry> entries;
@@ -130,6 +135,11 @@ public class AtomicLongBackedProcessor extends AbstractLocalProcessor
         {
             throw new IllegalStateException("We have overridden all callers of this method, it should never be called");
         }
+
+        public EntryHolder getEntries(Epoch since, Epoch until)
+        {
+            throw new IllegalStateException("We have overridden all callers of this method, it should never be called");
+        }
     }
 
     public static class InMemoryMetadataSnapshots implements MetadataSnapshots

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

@@ -902,6 +902,11 @@ public class ClusterMetadataService
             return delegate().fetchLogAndWait(waitFor, retryPolicy);
         }
 
+        public LogState reconstruct(Epoch lowEpoch, Epoch highEpoch, Retry.Deadline retryPolicy)
+        {
+            return delegate().reconstruct(lowEpoch, highEpoch, retryPolicy);
+        }
+
         public String toString()
         {
             return "SwitchableProcessor{" +

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

@@ -167,6 +167,11 @@ public class PaxosBackedProcessor extends AbstractLocalProcessor
         throw new ReadTimeoutException(ConsistencyLevel.QUORUM, blockFor - collected.size(), blockFor, false);
     }
 
+    public LogState reconstruct(Epoch lowEpoch, Epoch highEpoch, Retry.Deadline retryPolicy)
+    {
+        return DistributedMetadataLogKeyspace.getLogState(lowEpoch, highEpoch);
+    }
+
     private static <T> T unwrap(Promise<T> promise)
     {
         if (!promise.isDone() || !promise.isSuccess())

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

@@ -24,6 +24,7 @@ import com.codahale.metrics.Meter;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.metrics.TCMMetrics;
 import org.apache.cassandra.tcm.log.Entry;
+import org.apache.cassandra.tcm.log.LogState;
 import org.apache.cassandra.utils.Clock;
 
 public interface Processor
@@ -100,5 +101,8 @@ public interface Processor
                                Retry.Deadline.after(DatabaseDescriptor.getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS),
                                                     new Retry.Jitter(TCMMetrics.instance.fetchLogRetries)));
     }
+
     ClusterMetadata fetchLogAndWait(Epoch waitFor, Retry.Deadline retryPolicy);
+
+    LogState reconstruct(Epoch lowEpoch, Epoch highEpoch, Retry.Deadline retryPolicy);
 }

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

@@ -0,0 +1,86 @@
+/*
+ * 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.tcm;
+
+import java.io.IOException;
+
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.metrics.TCMMetrics;
+import org.apache.cassandra.net.IVerbHandler;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.schema.DistributedMetadataLogKeyspace;
+import org.apache.cassandra.tcm.log.LogState;
+import org.apache.cassandra.utils.FBUtilities;
+
+public class ReconstructLogState
+{
+    public static final Serializer serializer = new Serializer();
+
+    public final Epoch lowerBound;
+    public final Epoch higherBound;
+
+    public ReconstructLogState(Epoch lowerBound, Epoch higherBound)
+    {
+        this.lowerBound = lowerBound;
+        this.higherBound = higherBound;
+    }
+
+    static class Serializer implements IVersionedSerializer<ReconstructLogState>
+    {
+
+        public void serialize(ReconstructLogState t, DataOutputPlus out, int version) throws IOException
+        {
+            Epoch.serializer.serialize(t.lowerBound, out);
+            Epoch.serializer.serialize(t.higherBound, out);
+        }
+
+        public ReconstructLogState deserialize(DataInputPlus in, int version) throws IOException
+        {
+            Epoch lowerBound = Epoch.serializer.deserialize(in);
+            Epoch higherBound = Epoch.serializer.deserialize(in);
+            return new ReconstructLogState(lowerBound, higherBound);
+        }
+
+        public long serializedSize(ReconstructLogState t, int version)
+        {
+            return Epoch.serializer.serializedSize(t.lowerBound) +
+                   Epoch.serializer.serializedSize(t.higherBound);
+        }
+    }
+
+    public static class Handler implements IVerbHandler<ReconstructLogState>
+    {
+        public static final Handler instance = new Handler();
+
+        public void doVerb(Message<ReconstructLogState> message) throws IOException
+        {
+            TCMMetrics.instance.reconstructLogStateCall.mark();
+            ReconstructLogState request = message.payload;
+
+            if (!ClusterMetadataService.instance().isCurrentMember(FBUtilities.getBroadcastAddressAndPort()))
+                throw new NotCMSException("This node is not in the CMS, can't generate a consistent log fetch response to " + message.from());
+
+            LogState result = DistributedMetadataLogKeyspace.getLogState(request.lowerBound, request.higherBound);
+            MessagingService.instance().send(message.responseWith(result), message.from());
+        }
+    }
+}

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

@@ -151,6 +151,30 @@ public final class RemoteProcessor implements Processor
         }
     }
 
+    @Override
+    public LogState reconstruct(Epoch lowEpoch, Epoch highEpoch, Retry.Deadline retryPolicy)
+    {
+        try
+        {
+            Promise<LogState> request = new AsyncPromise<>();
+            List<InetAddressAndPort> candidates = new ArrayList<>(log.metadata().fullCMSMembers());
+            sendWithCallbackAsync(request,
+                                  Verb.TCM_RECONSTRUCT_EPOCH_REQ,
+                                  new ReconstructLogState(lowEpoch, highEpoch),
+                                  new CandidateIterator(candidates),
+                                  new Retry.Backoff(TCMMetrics.instance.fetchLogRetries));
+            return request.get(retryPolicy.remainingNanos(), TimeUnit.NANOSECONDS);
+        }
+        catch (InterruptedException e)
+        {
+            throw new RuntimeException("Can not reconstruct during shutdown", e);
+        }
+        catch (ExecutionException | TimeoutException e)
+        {
+            throw new RuntimeException("Could not reconstruct", e);
+        }
+    }
+
     public static ClusterMetadata fetchLogAndWait(CandidateIterator candidateIterator, LocalLog log)
     {
         try

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

@@ -33,6 +33,7 @@ import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState;
 import org.apache.cassandra.tcm.Commit.Replicator;
 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.Directory;
 import org.apache.cassandra.tcm.ownership.DataPlacements;
 import org.apache.cassandra.tcm.ownership.PlacementProvider;
@@ -145,6 +146,11 @@ public class StubClusterMetadataService extends ClusterMetadataService
         {
             throw new UnsupportedOperationException();
         }
+
+        public LogState reconstruct(Epoch lowEpoch, Epoch highEpoch, Retry.Deadline retryPolicy)
+        {
+            throw new UnsupportedOperationException();
+        }
     }
 
 

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

@@ -351,11 +351,16 @@ public abstract class LocalLog implements Closeable
         }
     }
 
-    public LogState getCommittedEntries(Epoch since)
+    public LogState getLocalEntries(Epoch since)
     {
         return storage.getLogState(since, false);
     }
 
+    public LogState getLocalEntries(Epoch since, Epoch until)
+    {
+        return storage.getLogState(since, until);
+    }
+
     public ClusterMetadata waitForHighestConsecutive()
     {
         runOnce();

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

@@ -27,6 +27,7 @@ import java.util.TreeSet;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Ordering;
 
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.tcm.Epoch;
 import org.apache.cassandra.tcm.MetadataSnapshots;
@@ -37,6 +38,7 @@ public interface LogReader
      * Gets all entries where epoch >= since - could be empty if since is a later epoch than the current highest seen
      */
     EntryHolder getEntries(Epoch since) throws IOException;
+    EntryHolder getEntries(Epoch since, Epoch until) throws IOException;
     MetadataSnapshots snapshots();
 
     /**
@@ -117,6 +119,53 @@ public interface LogReader
         }
     }
 
+    default LogState getLogState(Epoch start, Epoch end)
+    {
+        try
+        {
+            ClusterMetadata closestSnapshot = snapshots().getSnapshotBefore(start);
+
+            // Snapshot could not be found, fetch enough epochs to reconstruct the start metadata
+            if (closestSnapshot == null)
+            {
+                closestSnapshot = new ClusterMetadata(DatabaseDescriptor.getPartitioner());
+                ImmutableList.Builder<Entry> entries = new ImmutableList.Builder<>();
+                EntryHolder entryHolder = getEntries(Epoch.EMPTY, end);
+                for (Entry entry : entryHolder.entries)
+                {
+                    if (entry.epoch.isAfter(start))
+                        entries.add(entry);
+                    else
+                        closestSnapshot = entry.transform.execute(closestSnapshot).success().metadata;
+                }
+                return new LogState(closestSnapshot, entries.build());
+            }
+            else if (closestSnapshot.epoch.isBefore(start))
+            {
+                ImmutableList.Builder<Entry> entries = new ImmutableList.Builder<>();
+                EntryHolder entryHolder = getEntries(closestSnapshot.epoch, end);
+                for (Entry entry : entryHolder.entries)
+                {
+                    if (entry.epoch.isAfter(start))
+                        entries.add(entry);
+                    else
+                        closestSnapshot = entry.transform.execute(closestSnapshot).success().metadata;
+                }
+                return new LogState(closestSnapshot, entries.build());
+            }
+            else
+            {
+                assert closestSnapshot.epoch.isEqualOrAfter(start) : String.format("Got %s, but requested snapshot of %s", closestSnapshot.epoch, start);
+                EntryHolder entryHolder = getEntries(closestSnapshot.epoch.nextEpoch(), end);
+                return new LogState(closestSnapshot, ImmutableList.copyOf(entryHolder.entries));
+            }
+        }
+        catch (IOException e)
+        {
+            throw new RuntimeException(e);
+        }
+    }
+
     class EntryHolder
     {
         SortedSet<Entry> entries;

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

@@ -56,6 +56,12 @@ public interface LogStorage extends LogReader
             return LogState.EMPTY;
         }
 
+        @Override
+        public LogState getLogState(Epoch start, Epoch end)
+        {
+            return LogState.EMPTY;
+        }
+
         @Override
         public LogState getPersistedLogState()
         {
@@ -68,6 +74,12 @@ public interface LogStorage extends LogReader
             return null;
         }
 
+        @Override
+        public EntryHolder getEntries(Epoch since, Epoch until)
+        {
+            return null;
+        }
+
         @Override
         public MetadataSnapshots snapshots()
         {

src/java/org/apache/cassandra/tcm/migration/GossipProcessor.java

@@ -25,6 +25,7 @@ import org.apache.cassandra.tcm.log.Entry;
 import org.apache.cassandra.tcm.Epoch;
 import org.apache.cassandra.tcm.Transformation;
 import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.log.LogState;
 
 public class GossipProcessor implements Processor
 {
@@ -39,4 +40,9 @@ public class GossipProcessor implements Processor
     {
         return ClusterMetadata.current();
     }
+
+    public LogState reconstruct(Epoch lowEpoch, Epoch highEpoch, Retry.Deadline retryPolicy)
+    {
+        throw new IllegalStateException("Can't reconstruct log state when running in gossip mode. Enable the ClusterMetadataService with `nodetool addtocms`.");
+    }
 }

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

@@ -752,6 +752,11 @@ public abstract class CoordinatorPathTestBase extends FuzzTestBase
                                                log.append(logState);
                                                return log.waitForHighestConsecutive();
                                            }
+
+                                           public LogState reconstruct(Epoch lowEpoch, Epoch highEpoch, Retry.Deadline retryPolicy)
+                                           {
+                                               return log.getLocalEntries(lowEpoch, highEpoch);
+                                           }
                                        },
                                        (a,b) -> {},
                                        false);

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

@@ -0,0 +1,121 @@
+/*
+ * 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 java.util.Iterator;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.codahale.metrics.Meter;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.metrics.TCMMetrics;
+import org.apache.cassandra.schema.DistributedMetadataLogKeyspace;
+import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.Epoch;
+import org.apache.cassandra.tcm.Retry;
+import org.apache.cassandra.tcm.log.Entry;
+import org.apache.cassandra.tcm.log.LogState;
+import org.apache.cassandra.utils.Clock;
+
+public class ReconstructEpochTest extends TestBaseImpl
+{
+    @Test
+    public void logReaderTest() throws Exception
+    {
+        try (Cluster cluster = init(builder().withNodes(2).start()))
+        {
+            cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (id int primary key)"));
+            for (int i = 0; i < 30; i++)
+            {
+                if (i > 0 && i % 5 == 0)
+                    cluster.get(1).runOnInstance(() -> ClusterMetadataService.instance().triggerSnapshot());
+                cluster.schemaChange(withKeyspace("ALTER TABLE %s.tbl WITH comment = '" + i + "'"));
+            }
+
+            cluster.get(1).runOnInstance(() -> {
+                for (int[] cfg : new int[][]{ new int[]{ 6, 9 },
+                                              new int[]{ 2, 20 },
+                                              new int[]{ 5, 5 },
+                                              new int[]{ 15, 20 }})
+                {
+                    int start = cfg[0];
+                    int end = cfg[1];
+                    LogState logState = DistributedMetadataLogKeyspace.getLogState(Epoch.create(start), Epoch.create(end));
+                    Assert.assertEquals(start, logState.baseState.epoch.getEpoch());
+                    Iterator<Entry> iter = logState.entries.iterator();
+                    for (int i = start + 1; i <= end; i++)
+                        Assert.assertEquals(i, iter.next().epoch.getEpoch());
+                }
+            });
+
+
+            cluster.get(2).runOnInstance(() -> {
+                for (int[] cfg : new int[][]{ new int[]{ 6, 9 },
+                                              new int[]{ 2, 20 },
+                                              new int[]{ 5, 5 },
+                                              new int[]{ 15, 20 }})
+                {
+                    int start = cfg[0];
+                    int end = cfg[1];
+                    LogState logState = ClusterMetadataService.instance()
+                                                              .processor()
+                                                              .reconstruct(Epoch.create(start),
+                                                                           Epoch.create(end),
+                                                                           unsafeRetryIndefinitely());
+
+                    Assert.assertEquals(start, logState.baseState.epoch.getEpoch());
+                    Iterator<Entry> iter = logState.entries.iterator();
+                    for (int i = start + 1; i <= end; i++)
+                        Assert.assertEquals(i, iter.next().epoch.getEpoch());
+                }
+            });
+        }
+    }
+
+    private static Retry.Deadline unsafeRetryIndefinitely()
+    {
+        long timeoutNanos = DatabaseDescriptor.getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS);
+        Meter retryMeter = TCMMetrics.instance.commitRetries;
+        return new Retry.Deadline(Clock.Global.nanoTime() + timeoutNanos,
+                                  new Retry.Jitter(retryMeter))
+        {
+            @Override
+            public boolean reachedMax()
+            {
+                return false;
+            }
+
+            @Override
+            public long remainingNanos()
+            {
+                return timeoutNanos;
+            }
+
+            public String toString()
+            {
+                return String.format("RetryIndefinitely{tries=%d}", currentTries());
+            }
+        };
+    }
+
+}

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

@@ -32,6 +32,7 @@ import org.apache.cassandra.tcm.Processor;
 import org.apache.cassandra.tcm.Retry;
 import org.apache.cassandra.tcm.Transformation;
 import org.apache.cassandra.tcm.log.Entry;
+import org.apache.cassandra.tcm.log.LogState;
 import org.apache.cassandra.utils.concurrent.WaitQueue;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -69,6 +70,11 @@ public class TestProcessor implements Processor
         return delegate.fetchLogAndWait(waitFor, retryPolicy);
     }
 
+    public LogState reconstruct(Epoch lowEpoch, Epoch highEpoch, Retry.Deadline retryPolicy)
+    {
+        return delegate.reconstruct(lowEpoch, highEpoch, retryPolicy);
+    }
+
     protected void waitIfPaused()
     {
         if (isPaused())

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

@@ -145,6 +145,11 @@ public class LocalLogTest
                 return new EntryHolder(since);
             }
 
+            public EntryHolder getEntries(Epoch since, Epoch until) throws IOException
+            {
+                return new EntryHolder(since);
+            }
+
             @Override
             public MetadataSnapshots snapshots()
             {