From c16297f862ce1ef365dacd9ef1a3759712efbf75 Mon Sep 17 00:00:00 2001 From: Alex Petrov Date: Fri, 19 Jul 2024 18:01:16 +0200 Subject: [PATCH] Add an ability to reconstruct arbitrary epoch state from the log to TCM Patch by Alex Petrov; reviewed by Marcus Eriksson for CASSANDRA-19790. --- .../apache/cassandra/metrics/TCMMetrics.java | 2 + src/java/org/apache/cassandra/net/Verb.java | 3 + .../DistributedMetadataLogKeyspace.java | 35 +++++ .../cassandra/tcm/AbstractLocalProcessor.java | 2 +- .../tcm/AtomicLongBackedProcessor.java | 10 ++ .../cassandra/tcm/ClusterMetadataService.java | 5 + .../cassandra/tcm/PaxosBackedProcessor.java | 5 + .../org/apache/cassandra/tcm/Processor.java | 4 + .../cassandra/tcm/ReconstructLogState.java | 86 +++++++++++++ .../apache/cassandra/tcm/RemoteProcessor.java | 24 ++++ .../tcm/StubClusterMetadataService.java | 6 + .../apache/cassandra/tcm/log/LocalLog.java | 7 +- .../apache/cassandra/tcm/log/LogReader.java | 49 +++++++ .../apache/cassandra/tcm/log/LogStorage.java | 12 ++ .../tcm/migration/GossipProcessor.java | 6 + .../test/log/CoordinatorPathTestBase.java | 5 + .../test/log/ReconstructEpochTest.java | 121 ++++++++++++++++++ .../distributed/test/log/TestProcessor.java | 6 + .../cassandra/tcm/log/LocalLogTest.java | 5 + 19 files changed, 391 insertions(+), 2 deletions(-) create mode 100644 src/java/org/apache/cassandra/tcm/ReconstructLogState.java create mode 100644 test/distributed/org/apache/cassandra/distributed/test/log/ReconstructEpochTest.java diff --git a/src/java/org/apache/cassandra/metrics/TCMMetrics.java b/src/java/org/apache/cassandra/metrics/TCMMetrics.java index 01061b725a..909280c577 100644 --- a/src/java/org/apache/cassandra/metrics/TCMMetrics.java +++ b/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 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) diff --git a/src/java/org/apache/cassandra/net/Verb.java b/src/java/org/apache/cassandra/net/Verb.java index ff2cb4ef7a..e6c3264c48 100644 --- a/src/java/org/apache/cassandra/net/Verb.java +++ b/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 ), diff --git a/src/java/org/apache/cassandra/schema/DistributedMetadataLogKeyspace.java b/src/java/org/apache/cassandra/schema/DistributedMetadataLogKeyspace.java index 1e4a41c9ff..087fa51b8c 100644 --- a/src/java/org/apache/cassandra/schema/DistributedMetadataLogKeyspace.java +++ b/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() { diff --git a/src/java/org/apache/cassandra/tcm/AbstractLocalProcessor.java b/src/java/org/apache/cassandra/tcm/AbstractLocalProcessor.java index da4e2e53ba..6d126becc8 100644 --- a/src/java/org/apache/cassandra/tcm/AbstractLocalProcessor.java +++ b/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; diff --git a/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java b/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java index df00253433..41efdb245b 100644 --- a/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java +++ b/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 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 diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java b/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java index 3e3d8389ae..b9063adf5f 100644 --- a/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java +++ b/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{" + diff --git a/src/java/org/apache/cassandra/tcm/PaxosBackedProcessor.java b/src/java/org/apache/cassandra/tcm/PaxosBackedProcessor.java index 45b5945cbc..dbaac24041 100644 --- a/src/java/org/apache/cassandra/tcm/PaxosBackedProcessor.java +++ b/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 unwrap(Promise promise) { if (!promise.isDone() || !promise.isSuccess()) diff --git a/src/java/org/apache/cassandra/tcm/Processor.java b/src/java/org/apache/cassandra/tcm/Processor.java index fdb4cf23bb..3d29b43375 100644 --- a/src/java/org/apache/cassandra/tcm/Processor.java +++ b/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); } diff --git a/src/java/org/apache/cassandra/tcm/ReconstructLogState.java b/src/java/org/apache/cassandra/tcm/ReconstructLogState.java new file mode 100644 index 0000000000..f6a60f070a --- /dev/null +++ b/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 + { + + 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 + { + public static final Handler instance = new Handler(); + + public void doVerb(Message 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()); + } + } +} diff --git a/src/java/org/apache/cassandra/tcm/RemoteProcessor.java b/src/java/org/apache/cassandra/tcm/RemoteProcessor.java index ed10512a88..635be54cf9 100644 --- a/src/java/org/apache/cassandra/tcm/RemoteProcessor.java +++ b/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 request = new AsyncPromise<>(); + List 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 diff --git a/src/java/org/apache/cassandra/tcm/StubClusterMetadataService.java b/src/java/org/apache/cassandra/tcm/StubClusterMetadataService.java index d3318ecb74..8a69f9acc4 100644 --- a/src/java/org/apache/cassandra/tcm/StubClusterMetadataService.java +++ b/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(); + } } diff --git a/src/java/org/apache/cassandra/tcm/log/LocalLog.java b/src/java/org/apache/cassandra/tcm/log/LocalLog.java index 6e54e4657f..a84f0920e0 100644 --- a/src/java/org/apache/cassandra/tcm/log/LocalLog.java +++ b/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(); diff --git a/src/java/org/apache/cassandra/tcm/log/LogReader.java b/src/java/org/apache/cassandra/tcm/log/LogReader.java index 688f1c7653..b8d62e7587 100644 --- a/src/java/org/apache/cassandra/tcm/log/LogReader.java +++ b/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 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 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 entries; diff --git a/src/java/org/apache/cassandra/tcm/log/LogStorage.java b/src/java/org/apache/cassandra/tcm/log/LogStorage.java index 3d5e681b16..7772d7d07e 100644 --- a/src/java/org/apache/cassandra/tcm/log/LogStorage.java +++ b/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() { diff --git a/src/java/org/apache/cassandra/tcm/migration/GossipProcessor.java b/src/java/org/apache/cassandra/tcm/migration/GossipProcessor.java index 0cb654f19a..6c02318f48 100644 --- a/src/java/org/apache/cassandra/tcm/migration/GossipProcessor.java +++ b/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`."); + } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/CoordinatorPathTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/log/CoordinatorPathTestBase.java index cfa0ec4be7..2bf1047b55 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/log/CoordinatorPathTestBase.java +++ b/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); diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/ReconstructEpochTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/ReconstructEpochTest.java new file mode 100644 index 0000000000..5166ae8998 --- /dev/null +++ b/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 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 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()); + } + }; + } + +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/TestProcessor.java b/test/distributed/org/apache/cassandra/distributed/test/log/TestProcessor.java index f5fabfb4ac..6f359af057 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/log/TestProcessor.java +++ b/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()) diff --git a/test/unit/org/apache/cassandra/tcm/log/LocalLogTest.java b/test/unit/org/apache/cassandra/tcm/log/LocalLogTest.java index fbdafb131d..20bb5c79b0 100644 --- a/test/unit/org/apache/cassandra/tcm/log/LocalLogTest.java +++ b/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() {