natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Revert changes to serving FetchCMSLog/FetchPeerLog requests & remove ReconstructLogState 

Patch by Sam Tunnicliffe; reviewed by Alex Petrov for CASSANDRA-20719

Co-authored-by: Sam Tunnicliffe <samt@apache.org>
This commit is contained in:
Sam Tunnicliffe 2025-05-12 13:51:09 +01:00 committed by Alex Petrov
parent 6a37d88f5d
commit 693eab8776
23 changed files with 23 additions and 661 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/java/org/apache/cassandra/net/Verb.java
View File

@ -134,7 +134,6 @@ 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;
@ -306,8 +305,6 @@ 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, shortTimeout, FETCH_METADATA, MessageSerializers::logStateSerializer, RESPONSE_HANDLER ),
TCM_FETCH_PEER_LOG_REQ (819, P0, rpcTimeout, FETCH_METADATA, () -> FetchPeerLog.serializer, () -> FetchPeerLog.Handler.instance, TCM_FETCH_PEER_LOG_RSP ),
TCM_RECONSTRUCT_EPOCH_RSP (820, P0, rpcTimeout, FETCH_METADATA, MessageSerializers::logStateSerializer, () -> ResponseVerbHandler.instance ),
TCM_RECONSTRUCT_EPOCH_REQ (821, P0, rpcTimeout, FETCH_METADATA, () -> 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 ),

14
src/java/org/apache/cassandra/schema/DistributedMetadataLogKeyspace.java
View File

@ -163,20 +163,6 @@ 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, boolean includeSnapshot)
{
return serialLogReader.getLogState(start, end, includeSnapshot);
}
public static class DistributedTableLogReader implements LogReader
{
private final ConsistencyLevel consistencyLevel;

67
src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java
View File

@ -18,10 +18,8 @@
package org.apache.cassandra.tcm;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
@ -32,11 +30,8 @@ import com.google.common.collect.ImmutableList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.utils.Invariants;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.tcm.log.Entry;
import org.apache.cassandra.tcm.log.LocalLog;
import org.apache.cassandra.tcm.log.LogReader;
import org.apache.cassandra.tcm.log.LogState;
import org.apache.cassandra.tcm.log.LogStorage;
@ -81,39 +76,6 @@ public class AtomicLongBackedProcessor extends AbstractLocalProcessor
return log.waitForHighestConsecutive();
}
@Override
public LogState getLocalState(Epoch lowEpoch, Epoch highEpoch, boolean includeSnapshot)
{
try
{
LogReader.EntryHolder state = log.storage().getEntries(Epoch.EMPTY, highEpoch);
ClusterMetadata metadata = new ClusterMetadata(DatabaseDescriptor.getPartitioner());
Iterator<Entry> iter = state.iterator();
ImmutableList.Builder<Entry> rest = new ImmutableList.Builder<>();
while (iter.hasNext())
{
Entry current = iter.next();
if (current.epoch.isEqualOrBefore(lowEpoch))
metadata = current.transform.execute(metadata).success().metadata;
else
rest.add(current);
}
return new LogState(metadata, rest.build());
}
catch (IOException t)
{
throw new RuntimeException(t);
}
}
@Override
public LogState getLogState(Epoch lowEpoch, Epoch highEpoch, boolean includeSnapshot, Retry retryPolicy)
{
return getLocalState(lowEpoch, highEpoch, includeSnapshot);
}
public static class InMemoryStorage implements LogStorage
{
private final List<Entry> entries;
@ -137,7 +99,11 @@ public class AtomicLongBackedProcessor extends AbstractLocalProcessor
@Override
public synchronized LogState getLogState(Epoch startEpoch)
{
return getLogState(startEpoch, Epoch.MAX);
ImmutableList.Builder<Entry> builder = ImmutableList.builder();
ClusterMetadata latest = metadataSnapshots.getLatestSnapshot();
Epoch actualSince = latest != null && latest.epoch.isAfter(startEpoch) ? latest.epoch : startEpoch;
entries.stream().filter(e -> e.epoch.isAfter(actualSince)).forEach(builder::add);
return new LogState(latest, builder.build());
}
@Override
@ -174,29 +140,6 @@ public class AtomicLongBackedProcessor extends AbstractLocalProcessor
entries.stream().filter(e -> e.epoch.isAfter(since) && e.epoch.isEqualOrBefore(until)).forEach(entryHolder::add);
return entryHolder;
}
public LogState getLogState(Epoch start, Epoch end)
{
EntryHolder state = getEntries(Epoch.EMPTY);
ClusterMetadata metadata = new ClusterMetadata(DatabaseDescriptor.getPartitioner());
Iterator<Entry> iter = state.iterator();
ImmutableList.Builder<Entry> rest = new ImmutableList.Builder<>();
while (iter.hasNext())
{
Entry current = iter.next();
if (current.epoch.isAfter(end))
break;
if (current.epoch.isEqualOrBefore(start))
{
Invariants.require(current.epoch.isDirectlyAfter(metadata.epoch));
metadata = current.transform.execute(metadata).success().metadata;
}
else if (current.epoch.isAfter(start))
rest.add(current);
}
return new LogState(metadata, rest.build());
}
}
public static class InMemoryMetadataSnapshots implements MetadataSnapshots

18
src/java/org/apache/cassandra/tcm/ClusterMetadataService.java
View File

@ -172,16 +172,16 @@ public class ClusterMetadataService
{
log = logSpec.sync().withStorage(new AtomicLongBackedProcessor.InMemoryStorage()).createLog();
localProcessor = wrapProcessor.apply(new AtomicLongBackedProcessor(log, logSpec.isReset()));
fetchLogHandler = new FetchCMSLog.Handler((e, ignored) -> logSpec.storage().getLogState(e));
}
else
{
log = logSpec.async().createLog();
localProcessor = wrapProcessor.apply(new PaxosBackedProcessor(log));
fetchLogHandler = new FetchCMSLog.Handler();
}
fetchLogHandler = new FetchCMSLog.Handler();
Commit.Replicator replicator = CassandraRelevantProperties.TCM_USE_NO_OP_REPLICATOR.getBoolean()
Commit.Replicator replicator = CassandraRelevantProperties.TCM_USE_TEST_NO_OP_REPLICATOR.getBoolean()
? Commit.Replicator.NO_OP
: new Commit.DefaultReplicator(() -> log.metadata().directory);
@ -938,18 +938,6 @@ public class ClusterMetadataService
return delegate().fetchLogAndWait(waitFor, retryPolicy);
}
@Override
public LogState getLocalState(Epoch start, Epoch end, boolean includeSnapshot)
{
return delegate().getLocalState(start, end, includeSnapshot);
}
@Override
public LogState getLogState(Epoch start, Epoch end, boolean includeSnapshot, Retry retryPolicy)
{
return delegate().getLogState(start, end, includeSnapshot, retryPolicy);
}
public String toString()
{
return "SwitchableProcessor{" +

22
src/java/org/apache/cassandra/tcm/FetchCMSLog.java
View File

@ -19,8 +19,7 @@
package org.apache.cassandra.tcm;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import java.util.function.BiFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -33,11 +32,10 @@ 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;
import static org.apache.cassandra.config.DatabaseDescriptor.getCmsAwaitTimeout;
public class FetchCMSLog
{
public static final Serializer serializer = new Serializer();
@ -91,16 +89,16 @@ public class FetchCMSLog
* to node-local (which only relevant in cases of CMS expansions/shrinks, and can only be requested by the
* CMS node that collects the highest epoch from the quorum of peers).
*/
private final Supplier<Processor> processor;
private final BiFunction<Epoch, Boolean, LogState> logStateSupplier;
public Handler()
{
this(() -> ClusterMetadataService.instance().processor());
this(DistributedMetadataLogKeyspace::getLogState);
}
public Handler(Supplier<Processor> processor)
public Handler(BiFunction<Epoch, Boolean, LogState> logStateSupplier)
{
this.processor = processor;
this.logStateSupplier = logStateSupplier;
}
public void doVerb(Message<FetchCMSLog> message) throws IOException
@ -116,13 +114,7 @@ public class FetchCMSLog
// If both we and the other node believe it should be caught up with a linearizable read
boolean consistentFetch = request.consistentFetch && !ClusterMetadataService.instance().isCurrentMember(message.from());
Retry retry = Retry.untilElapsed(getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS), TCMMetrics.instance.fetchLogRetries);
LogState delta;
if (consistentFetch)
delta = processor.get().getLogState(message.payload.lowerBound, Epoch.MAX, false, retry);
else
delta = processor.get().getLocalState(message.payload.lowerBound, Epoch.MAX, false);
LogState delta = logStateSupplier.apply(message.payload.lowerBound, consistentFetch);
TCMMetrics.instance.cmsLogEntriesServed(message.payload.lowerBound, delta.latestEpoch());
logger.info("Responding to {}({}) with log delta: {}", message.from(), request, delta);
MessagingService.instance().send(message.responseWith(delta), message.from());

4
src/java/org/apache/cassandra/tcm/FetchPeerLog.java
View File

@ -81,9 +81,7 @@ public class FetchPeerLog
ClusterMetadata metadata = ClusterMetadata.current();
logger.info("Received peer log fetch request {} from {}: start = {}, current = {}", request, message.from(), message.payload.start, metadata.epoch);
LogState delta = ClusterMetadataService.instance()
.processor()
.getLocalState(message.payload.start, Epoch.MAX, false);
LogState delta = ClusterMetadataService.instance().log().storage().getLogState(message.payload.start);
TCMMetrics.instance.peerLogEntriesServed(message.payload.start, delta.latestEpoch());
logger.info("Responding with log delta: {}", delta);
MessagingService.instance().send(message.responseWith(delta), message.from());

25
src/java/org/apache/cassandra/tcm/PaxosBackedProcessor.java
View File

@ -168,31 +168,6 @@ public class PaxosBackedProcessor extends AbstractLocalProcessor
throw new ReadTimeoutException(ConsistencyLevel.QUORUM, blockFor - collected.size(), blockFor, false);
}
@Override
public LogState getLocalState(Epoch start, Epoch end, boolean includeSnapshot)
{
return log.storage().getLogState(start, end, includeSnapshot);
}
@Override
public LogState getLogState(Epoch start, Epoch end, boolean includeSnapshot, Retry retryPolicy)
{
while (true)
{
if (Thread.currentThread().isInterrupted())
throw new RuntimeException("Can not reconstruct during shutdown", new InterruptedException());
try
{
return DistributedMetadataLogKeyspace.getLogState(start, end, includeSnapshot);
}
catch (RuntimeException e) // honestly best to only retry timeouts, but everything gets wrapped in a RuntimeException...
{
if (!retryPolicy.maybeSleep())
throw new RuntimeException(String.format("Could not reconstruct range %d, %d", start.getEpoch(), end.getEpoch()), new TimeoutException());
}
}
}
private static <T> T unwrap(Promise<T> promise)
{
if (!promise.isDone() || !promise.isSuccess())

38
src/java/org/apache/cassandra/tcm/Processor.java
View File

@ -18,18 +18,13 @@
package org.apache.cassandra.tcm;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import com.codahale.metrics.Meter;
import accord.utils.Invariants;
import org.apache.cassandra.metrics.TCMMetrics;
import org.apache.cassandra.service.WaitStrategy;
import org.apache.cassandra.tcm.log.Entry;
import org.apache.cassandra.tcm.log.LogState;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.config.DatabaseDescriptor.getCmsAwaitTimeout;
@ -103,37 +98,4 @@ public interface Processor
}
ClusterMetadata fetchLogAndWait(Epoch waitFor, Retry retryPolicy);
/**
* Queries node's _local_ state. It is not guaranteed to be contiguous, but can be used for restoring CMS state/
*/
LogState getLocalState(Epoch start, Epoch end, boolean includeSnapshot);
/**
* Queries global log state.
*/
LogState getLogState(Epoch start, Epoch end, boolean includeSnapshot, Retry retryPolicy);
/**
* Reconstructs
*/
default List<ClusterMetadata> reconstruct(Epoch lowEpoch, Epoch highEpoch, Retry retryPolicy)
{
LogState logState = getLogState(lowEpoch, highEpoch, true, retryPolicy);
if (logState.isEmpty()) return Collections.emptyList();
List<ClusterMetadata> cms = new ArrayList<>(logState.entries.size());
ClusterMetadata acc = logState.baseState;
cms.add(acc);
for (Entry entry : logState.entries)
{
Invariants.require(entry.epoch.isDirectlyAfter(acc.epoch), "%s should have been directly after %s", entry.epoch, acc.epoch);
Transformation.Result res = entry.transform.execute(acc);
assert res.isSuccess() : res.toString();
acc = res.success().metadata;
cms.add(acc);
}
return cms;
}
}

106
src/java/org/apache/cassandra/tcm/ReconstructLogState.java
View File

@ -1,106 +0,0 @@
/*
* 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 java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import org.apache.cassandra.db.TypeSizes;
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.tcm.log.LogState;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.config.DatabaseDescriptor.getCmsAwaitTimeout;
public class ReconstructLogState
{
public static final Serializer serializer = new Serializer();
public final Epoch lowerBound;
public final Epoch higherBound;
public final boolean includeSnapshot;
public ReconstructLogState(Epoch lowerBound, Epoch higherBound, boolean includeSnapshot)
{
this.lowerBound = lowerBound;
this.higherBound = higherBound;
this.includeSnapshot = includeSnapshot;
}
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);
out.writeBoolean(t.includeSnapshot);
}
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, in.readBoolean());
}
public long serializedSize(ReconstructLogState t, int version)
{
return Epoch.serializer.serializedSize(t.lowerBound) +
Epoch.serializer.serializedSize(t.higherBound) +
TypeSizes.BOOL_SIZE;
}
}
public static class Handler implements IVerbHandler<ReconstructLogState>
{
public static final Handler instance = new Handler();
private final Supplier<Processor> processor;
public Handler()
{
this(() -> ClusterMetadataService.instance().processor());
}
public Handler(Supplier<Processor> processor)
{
this.processor = processor;
}
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 = processor.get().getLogState(request.lowerBound, request.higherBound, request.includeSnapshot,
Retry.untilElapsed(getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS), TCMMetrics.instance.fetchLogRetries));
MessagingService.instance().send(message.responseWith(result), message.from());
}
}
}

30
src/java/org/apache/cassandra/tcm/RemoteProcessor.java
View File

@ -151,36 +151,6 @@ public final class RemoteProcessor implements Processor
}
}
@Override
public LogState getLocalState(Epoch start, Epoch end, boolean includeSnapshot)
{
return log.getLocalEntries(start, end, includeSnapshot);
}
@Override
public LogState getLogState(Epoch lowEpoch, Epoch highEpoch, boolean includeSnapshot, Retry 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, includeSnapshot),
new CandidateIterator(candidates),
retryPolicy);
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(String.format("Could not reconstruct range %d, %d", lowEpoch.getEpoch(), highEpoch.getEpoch()), e);
}
}
public static ClusterMetadata fetchLogAndWait(CandidateIterator candidateIterator, LocalLog log)
{
try

14
src/java/org/apache/cassandra/tcm/StubClusterMetadataService.java
View File

@ -34,7 +34,6 @@ 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;
@ -152,21 +151,8 @@ public class StubClusterMetadataService extends ClusterMetadataService
{
throw new UnsupportedOperationException();
}
@Override
public LogState getLocalState(Epoch start, Epoch end, boolean includeSnapshot)
{
throw new UnsupportedOperationException();
}
@Override
public LogState getLogState(Epoch start, Epoch end, boolean includeSnapshot, Retry retryPolicy)
{
throw new UnsupportedOperationException();
}
}
public static Builder builder()
{
return new Builder();

5
src/java/org/apache/cassandra/tcm/log/LocalLog.java
View File

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

54
src/java/org/apache/cassandra/tcm/log/LogReader.java
View File

@ -28,8 +28,6 @@ import java.util.TreeSet;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Ordering;
import accord.utils.Invariants;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.MetadataSnapshots;
@ -121,58 +119,6 @@ public interface LogReader
}
}
default LogState getLogState(Epoch start, Epoch end, boolean includeSnapshot)
{
try
{
ClusterMetadata closestSnapshot = null;
if (includeSnapshot)
closestSnapshot = snapshots().getSnapshotBefore(start);
// Snapshot could not be found, fetch enough epochs to reconstruct the start metadata
if (closestSnapshot == null)
{
if (includeSnapshot)
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 if (includeSnapshot)
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<>();
// start is exclusive, so use the closest snapshot
EntryHolder entryHolder = getEntries(closestSnapshot.epoch, end);
for (Entry entry : entryHolder.entries)
{
if (entry.epoch.isAfter(start))
entries.add(entry);
else if (includeSnapshot)
closestSnapshot = entry.transform.execute(closestSnapshot).success().metadata;
}
return new LogState(closestSnapshot, entries.build());
}
else
{
Invariants.require(closestSnapshot.epoch.isEqualOrAfter(start),
"Got %s, but requested snapshot of %s", closestSnapshot.epoch, start);
EntryHolder entryHolder = getEntries(closestSnapshot.epoch, end);
return new LogState(closestSnapshot, ImmutableList.copyOf(entryHolder.entries));
}
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
class EntryHolder
{
SortedSet<Entry> entries;

6
src/java/org/apache/cassandra/tcm/log/LogStorage.java
View File

@ -56,12 +56,6 @@ public interface LogStorage extends LogReader
return LogState.EMPTY;
}
@Override
public LogState getLogState(Epoch start, Epoch end, boolean includeSnapshot)
{
return LogState.EMPTY;
}
@Override
public LogState getPersistedLogState()
{

13
src/java/org/apache/cassandra/tcm/migration/GossipProcessor.java
View File

@ -25,7 +25,6 @@ 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
{
@ -40,16 +39,4 @@ public class GossipProcessor implements Processor
{
return ClusterMetadata.current();
}
@Override
public LogState getLocalState(Epoch start, Epoch end, boolean includeSnapshot)
{
throw new IllegalStateException("Can't reconstruct log state when running in gossip mode. Enable the ClusterMetadataService with `nodetool addtocms`.");
}
@Override
public LogState getLogState(Epoch start, Epoch end, boolean includeSnapshot, Retry retryPolicy)
{
throw new IllegalStateException("Can't reconstruct log state when running in gossip mode. Enable the ClusterMetadataService with `nodetool addtocms`.");
}
}

12
test/distributed/org/apache/cassandra/distributed/test/log/CoordinatorPathTestBase.java
View File

@ -756,18 +756,6 @@ public abstract class CoordinatorPathTestBase extends FuzzTestBase
log.append(logState);
return log.waitForHighestConsecutive();
}
@Override
public LogState getLocalState(Epoch start, Epoch end, boolean includeSnapshot)
{
return log.getLocalEntries(start, end, includeSnapshot);
}
@Override
public LogState getLogState(Epoch start, Epoch end, boolean includeSnapshot, Retry retryPolicy)
{
return getLocalState(start, end, includeSnapshot);
}
},
(a,b) -> {},
false);

94
test/distributed/org/apache/cassandra/distributed/test/log/ReconstructEpochTest.java
View File

@ -1,94 +0,0 @@
/*
* 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 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 static org.apache.cassandra.config.DatabaseDescriptor.getCmsAwaitTimeout;
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), true);
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()
.getLogState(Epoch.create(start),
Epoch.create(end),
true,
Retry.untilElapsed(getCmsAwaitTimeout().to(TimeUnit.NANOSECONDS), TCMMetrics.instance.commitRetries));
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());
}
});
}
}
}

5
test/distributed/org/apache/cassandra/distributed/test/log/SystemKeyspaceStorageTest.java
View File

@ -98,6 +98,7 @@ public class SystemKeyspaceStorageTest extends CoordinatorPathTestBase
cluster.get(1).runOnInstance(() -> deleteSnapshot(toRemoveSnapshot.getEpoch()));
}
}
Epoch latestSnapshot = remainingSnapshots.get(remainingSnapshots.size() - 1);
Epoch lastEpoch = allEpochs.stream().max(Comparator.naturalOrder()).get();
repeat(10, () -> {
repeat(100, () -> {
@ -118,14 +119,12 @@ public class SystemKeyspaceStorageTest extends CoordinatorPathTestBase
}
else
{
assertEquals(since, logState.baseState.epoch);
assertEquals(latestSnapshot, logState.baseState.epoch);
start = logState.baseState.epoch;
if (logState.entries.isEmpty()) // no entries, snapshot should have the same epoch as since
assertEquals(since, start);
else // first epoch in entries should be snapshot epoch + 1
{
if (!start.nextEpoch().equals(logState.entries.get(0).epoch))
System.out.println(1);
assertEquals(start.nextEpoch(), logState.entries.get(0).epoch);
}
}

13
test/distributed/org/apache/cassandra/distributed/test/log/TestProcessor.java
View File

@ -32,7 +32,6 @@ 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;
@ -70,18 +69,6 @@ public class TestProcessor implements Processor
return delegate.fetchLogAndWait(waitFor, retryPolicy);
}
@Override
public LogState getLocalState(Epoch start, Epoch end, boolean includeSnapshot)
{
return delegate.getLocalState(start, end, includeSnapshot);
}
@Override
public LogState getLogState(Epoch start, Epoch end, boolean includeSnapshot, Retry retryPolicy)
{
return delegate.getLogState(start, end, includeSnapshot, retryPolicy);
}
protected void waitIfPaused()
{
if (isPaused())

21
test/unit/org/apache/cassandra/tcm/ValidatingClusterMetadataService.java
View File

@ -22,14 +22,12 @@ import java.io.IOException;
import java.util.List;
import java.util.TreeMap;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.tcm.log.Entry;
import org.apache.cassandra.tcm.log.LogState;
import org.apache.cassandra.tcm.sequences.LockedRanges;
import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializers;
@ -130,25 +128,6 @@ public class ValidatingClusterMetadataService extends StubClusterMetadataService
{
return delegate.fetchLogAndWait(waitFor, retryPolicy);
}
@Override
public LogState getLocalState(Epoch lowEpoch, Epoch highEpoch, boolean includeSnapshot)
{
if (!epochs.containsKey(lowEpoch))
throw new AssertionError("Unknown epoch: " + lowEpoch);
ClusterMetadata base = epochs.get(lowEpoch);
ImmutableList.Builder<Entry> entries = ImmutableList.builder();
int id = 0;
for (ClusterMetadata cm : epochs.subMap(lowEpoch, false, highEpoch, true).values())
entries.add(new Entry(new Entry.Id(id++), cm.epoch, new MockTransformer(cm)));
return new LogState(includeSnapshot ? base : null, entries.build());
}
@Override
public LogState getLogState(Epoch lowEpoch, Epoch highEpoch, boolean includeSnapshot, Retry retryPolicy)
{
return getLocalState(lowEpoch, highEpoch, includeSnapshot);
}
};
}

4
test/unit/org/apache/cassandra/tcm/log/DistributedLogStateTest.java
View File

@ -102,9 +102,9 @@ public class DistributedLogStateTest extends LogStateTestBase
}
@Override
public LogReader reader()
public LogState getLogState(Epoch since)
{
return reader;
return reader.getLogState(since);
}
@Override

4
test/unit/org/apache/cassandra/tcm/log/LocalStorageLogStateTest.java
View File

@ -90,9 +90,9 @@ public class LocalStorageLogStateTest extends LogStateTestBase
}
@Override
public LogReader reader()
public LogState getLogState(Epoch since)
{
return storage;
return storage.getLogState(since);
}
@Override

112
test/unit/org/apache/cassandra/tcm/log/LogStateTestBase.java
View File

@ -22,23 +22,18 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.stream.Stream;
import org.junit.Before;
import org.junit.Test;
import accord.utils.Gen;
import accord.utils.Gens;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.MetadataSnapshots;
import org.apache.cassandra.tcm.sequences.SequencesUtils;
import org.assertj.core.api.Assertions;
import static accord.utils.Property.qt;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@ -51,42 +46,13 @@ public abstract class LogStateTestBase
static int EXTRA_ENTRIES = 2;
static Epoch CURRENT_EPOCH = Epoch.create((NUM_SNAPSHOTS * SNAPSHOT_FREQUENCY) + EXTRA_ENTRIES);
static Epoch LATEST_SNAPSHOT_EPOCH = Epoch.create(NUM_SNAPSHOTS * SNAPSHOT_FREQUENCY);
private static final Gen.LongGen EPOCH_GEN = rs -> rs.nextLong(0, CURRENT_EPOCH.getEpoch()) + 1;
private static final Gen<Between> BETWEEN_GEN = rs -> {
long a = EPOCH_GEN.nextLong(rs);
long b = EPOCH_GEN.nextLong(rs);
while (b == a)
b = EPOCH_GEN.nextLong(rs);
if (b < a)
{
long tmp = a;
a = b;
b = tmp;
}
return new Between(Epoch.create(a), Epoch.create(b));
};
private static final Gen<MetadataSnapshots> SNAPSHOTS_GEN = Gens.<MetadataSnapshots>oneOf()
.add(i -> MetadataSnapshots.NO_OP)
.add(i -> throwing())
.add(rs -> rs.nextBoolean() ? withCorruptSnapshots(LATEST_SNAPSHOT_EPOCH) : withAvailableSnapshots(LATEST_SNAPSHOT_EPOCH))
.add(rs -> {
Epoch[] queriedEpochs = new Epoch[NUM_SNAPSHOTS];
for (int i = 0; i < NUM_SNAPSHOTS; i++)
queriedEpochs[i] = SequencesUtils.epoch((NUM_SNAPSHOTS - i) * SNAPSHOT_FREQUENCY);
return rs.nextBoolean() ? withCorruptSnapshots(queriedEpochs) : withAvailableSnapshots(queriedEpochs);
})
.build();
interface LogStateSUT
{
void cleanup() throws IOException;
void insertRegularEntry() throws IOException;
void snapshotMetadata() throws IOException;
LogReader reader();
default LogState getLogState(Epoch since)
{
return reader().getLogState(since);
}
LogState getLogState(Epoch since);
// just for manually checking the test data
void dumpTables() throws IOException;
@ -291,47 +257,6 @@ public abstract class LogStateTestBase
assertEntries(state.entries, since.nextEpoch(), CURRENT_EPOCH);
}
@Test
public void getLogStateBetween()
{
qt().forAll(SNAPSHOTS_GEN, BETWEEN_GEN).check((snapshots, between) -> {
LogStateSUT sut = getSystemUnderTest(snapshots);
LogState state = sut.reader().getLogState(between.start, between.end, true);
Assertions.assertThat(state.entries).describedAs("with and without snapshot should have the same entries").isEqualTo(sut.reader().getLogState(between.start, between.end, false).entries);
Assertions.assertThat(state.baseState.epoch).isEqualTo(between.start);
List<Entry> entries = state.entries;
Assertions.assertThat(entries.size()).isEqualTo(between.end.getEpoch() - between.start.getEpoch());
long expected = between.start.nextEpoch().getEpoch();
for (Entry e : entries)
{
long actual = e.epoch.getEpoch();
Assertions.assertThat(actual).describedAs("Unexpected epoch").isEqualTo(expected);
expected++;
}
});
}
@Test
public void getEntriesBetween()
{
qt().forAll(SNAPSHOTS_GEN, BETWEEN_GEN).check((snapshots, between) -> {
LogStateSUT sut = getSystemUnderTest(snapshots);
LogReader.EntryHolder entries = sut.reader().getEntries(between.start, between.end);
Assertions.assertThat(entries.since).isEqualTo(between.start);
Assertions.assertThat(entries.entries.size()).isEqualTo(between.end.getEpoch() - between.start.getEpoch());
long expected = between.start.nextEpoch().getEpoch();
for (Entry e : entries.entries)
{
long actual = e.epoch.getEpoch();
Assertions.assertThat(actual).describedAs("Unexpected epoch").isEqualTo(expected);
expected++;
}
});
}
private void assertEntries(List<Entry> entries, Epoch min, Epoch max)
{
int idx = 0;
@ -343,39 +268,4 @@ public abstract class LogStateTestBase
}
assertEquals(idx, entries.size());
}
private static class Between
{
private final Epoch start, end;
private Between(Epoch start, Epoch end)
{
this.start = start;
this.end = end;
}
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Between between = (Between) o;
return start.equals(between.start) && end.equals(between.end);
}
@Override
public int hashCode()
{
return Objects.hash(start, end);
}
@Override
public String toString()
{
return "Between{" +
"start=" + start.getEpoch() +
", end=" + end.getEpoch() +
'}';
}
}
}