mirror of https://github.com/apache/cassandra
Accord: startup race condition where accord journal tries to access the 2i index before its ready
patch by David Capwell; reviewed by Alex Petrov for CASSANDRA-20686
This commit is contained in:
David Capwell
parent
83f58b4e1b
commit
4d141e4b0a
|
|
@ -1,4 +1,5 @@
|
|||
5.1
|
||||
* Accord: startup race condition where accord journal tries to access the 2i index before its ready (CASSANDRA-20686)
|
||||
* Adopt Unsafe::invokeCleaner for Direct ByteBuffer cleaning (CASSANDRA-20677)
|
||||
* Add additional metrics around hints (CASSANDRA-20499)
|
||||
* Support for add and replace in IntervalTree (CASSANDRA-20513)
|
||||
|
|
|
|||
|
|
@ -154,6 +154,7 @@ public class AccordSpec
|
|||
public StringRetryStrategy retry_bootstrap = new StringRetryStrategy("10s*attempts <= 600s");
|
||||
public StringRetryStrategy retry_fetch_min_epoch = new StringRetryStrategy("200ms...1s*attempts <= 1s,retries=3");
|
||||
public StringRetryStrategy retry_fetch_topology = new StringRetryStrategy("200ms...1s*attempts <= 1s,retries=100");
|
||||
public StringRetryStrategy retry_journal_index_ready = new StringRetryStrategy("100ms");
|
||||
|
||||
public volatile DurationSpec.IntSecondsBound fast_path_update_delay = null;
|
||||
|
||||
|
|
|
|||
|
|
@ -165,6 +165,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
|
|||
this.node = node;
|
||||
status = Status.STARTING;
|
||||
journal.start();
|
||||
journalTable.start();
|
||||
}
|
||||
|
||||
public boolean started()
|
||||
|
|
|
|||
|
|
@ -22,10 +22,12 @@ import java.nio.ByteBuffer;
|
|||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Consumer;
|
||||
import javax.annotation.CheckForNull;
|
||||
import javax.annotation.Nullable;
|
||||
|
||||
import com.google.common.annotations.VisibleForTesting;
|
||||
import com.google.common.collect.AbstractIterator;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
|
@ -33,7 +35,9 @@ import org.slf4j.LoggerFactory;
|
|||
import accord.primitives.Timestamp;
|
||||
import accord.primitives.TxnId;
|
||||
import accord.utils.Invariants;
|
||||
import accord.utils.UncheckedInterruptedException;
|
||||
import org.agrona.collections.LongHashSet;
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.cql3.ColumnIdentifier;
|
||||
import org.apache.cassandra.cql3.Operator;
|
||||
import org.apache.cassandra.db.ColumnFamilyStore;
|
||||
|
|
@ -72,6 +76,7 @@ import org.apache.cassandra.journal.Journal;
|
|||
import org.apache.cassandra.journal.KeySupport;
|
||||
import org.apache.cassandra.journal.RecordConsumer;
|
||||
import org.apache.cassandra.schema.ColumnMetadata;
|
||||
import org.apache.cassandra.service.RetryStrategy;
|
||||
import org.apache.cassandra.service.accord.api.TokenKey;
|
||||
import org.apache.cassandra.service.accord.serializers.Version;
|
||||
import org.apache.cassandra.utils.CloseableIterator;
|
||||
|
|
@ -149,6 +154,38 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
|
|||
return new TableRangeSearcher();
|
||||
}
|
||||
|
||||
public void start()
|
||||
{
|
||||
if (index == null) return;
|
||||
Index tableIndex = cfs.indexManager.getIndexByName(AccordKeyspace.JOURNAL_INDEX_NAME);
|
||||
RetryStrategy retry = DatabaseDescriptor.getAccord().retry_journal_index_ready.retry();
|
||||
for (int i = 0; !cfs.indexManager.isIndexQueryable(tableIndex); i++)
|
||||
{
|
||||
logger.debug("Journal index {} is not ready wait... waiting", AccordKeyspace.JOURNAL_INDEX_NAME);
|
||||
maybeWait(retry, i);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* This method is here to make it easier for org.apache.cassandra.distributed.test.accord.journal.JournalAccessRouteIndexOnStartupRaceTest
|
||||
* to check when we need to do waiting
|
||||
*/
|
||||
@VisibleForTesting
|
||||
private static void maybeWait(RetryStrategy retry, int i)
|
||||
{
|
||||
long waitTime = retry.computeWait(i, TimeUnit.MICROSECONDS);
|
||||
if (waitTime == -1)
|
||||
throw new IllegalStateException("Gave up waiting on journal index to be ready");
|
||||
try
|
||||
{
|
||||
TimeUnit.MICROSECONDS.sleep(waitTime);
|
||||
}
|
||||
catch (InterruptedException e)
|
||||
{
|
||||
throw new UncheckedInterruptedException(e);
|
||||
}
|
||||
}
|
||||
|
||||
public interface Reader
|
||||
{
|
||||
void read(DataInputPlus input, Version userVersion) throws IOException;
|
||||
|
|
|
|||
|
|
@ -187,6 +187,20 @@ public class ClusterUtils
|
|||
cluster.stream().forEach(ClusterUtils::stopUnchecked);
|
||||
}
|
||||
|
||||
/**
|
||||
* Restart an instance in a blocking manner.
|
||||
*
|
||||
* @param instance the instance to restart
|
||||
*/
|
||||
public static void restartUnchecked(IInstance instance)
|
||||
{
|
||||
logger.info("Stopping instance {} to restart it", instance);
|
||||
stopUnchecked(instance);
|
||||
logger.info("Instance {} stopped, trying to start it now", instance);
|
||||
instance.startup();
|
||||
logger.info("Instance {} startup was success", instance);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new instance and add it to the cluster, without starting it.
|
||||
*
|
||||
|
|
@ -1302,11 +1316,7 @@ public class ClusterUtils
|
|||
*/
|
||||
public static File getCommitLogDirectory(IInstance instance)
|
||||
{
|
||||
IInstanceConfig conf = instance.config();
|
||||
// this isn't safe as it assumes the implementation of InstanceConfig
|
||||
// might need to get smarter... some day...
|
||||
String d = (String) conf.get("commitlog_directory");
|
||||
return new File(d);
|
||||
return getDirectory(instance, "commitlog_directory");
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -1317,11 +1327,7 @@ public class ClusterUtils
|
|||
*/
|
||||
public static File getHintsDirectory(IInstance instance)
|
||||
{
|
||||
IInstanceConfig conf = instance.config();
|
||||
// this isn't safe as it assumes the implementation of InstanceConfig
|
||||
// might need to get smarter... some day...
|
||||
String d = (String) conf.get("hints_directory");
|
||||
return new File(d);
|
||||
return getDirectory(instance, "hints_directory");
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -1331,11 +1337,33 @@ public class ClusterUtils
|
|||
* @return saved caches directory
|
||||
*/
|
||||
public static File getSavedCachesDirectory(IInstance instance)
|
||||
{
|
||||
return getDirectory(instance, "saved_caches_directory");
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the journal directory for the given instance.
|
||||
* This directory is used by the Accord consensus protocol to store its journal files.
|
||||
*
|
||||
* @param instance to get the journal directory for
|
||||
* @return journal directory
|
||||
*/
|
||||
public static File getJournalDirectory(IInstance instance)
|
||||
{
|
||||
return getDirectory(instance, "accord.journal_directory");
|
||||
}
|
||||
|
||||
public static File getCdcRawDirectory(IInstance instance)
|
||||
{
|
||||
return getDirectory(instance, "cdc_raw_directory");
|
||||
}
|
||||
|
||||
private static File getDirectory(IInstance instance, String key)
|
||||
{
|
||||
IInstanceConfig conf = instance.config();
|
||||
// this isn't safe as it assumes the implementation of InstanceConfig
|
||||
// might need to get smarter... some day...
|
||||
String d = (String) conf.get("saved_caches_directory");
|
||||
String d = (String) conf.get(key);
|
||||
return new File(d);
|
||||
}
|
||||
|
||||
|
|
@ -1352,9 +1380,27 @@ public class ClusterUtils
|
|||
out.add(getCommitLogDirectory(instance));
|
||||
out.add(getHintsDirectory(instance));
|
||||
out.add(getSavedCachesDirectory(instance));
|
||||
out.add(getJournalDirectory(instance));
|
||||
out.add(getCdcRawDirectory(instance));
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Stops the instance then deletes all directories, effectively resets the instance to a clean state.
|
||||
*
|
||||
* @param inst the instance to clean up
|
||||
*/
|
||||
public static void cleanup(IInvokableInstance inst)
|
||||
{
|
||||
stopUnchecked(inst);
|
||||
for (var f : getDirectories(inst))
|
||||
{
|
||||
if (f.exists())
|
||||
f.deleteRecursive();
|
||||
}
|
||||
inst.startup();
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the name of the Partitioner for the given instance.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -0,0 +1,158 @@
|
|||
/*
|
||||
* 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.accord.journal;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.concurrent.Callable;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import accord.local.durability.DurabilityService;
|
||||
import accord.primitives.Ranges;
|
||||
import accord.primitives.Timestamp;
|
||||
import accord.utils.async.AsyncChains;
|
||||
import net.bytebuddy.ByteBuddy;
|
||||
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
|
||||
import net.bytebuddy.implementation.MethodDelegation;
|
||||
import net.bytebuddy.implementation.bind.annotation.SuperCall;
|
||||
import org.apache.cassandra.db.Keyspace;
|
||||
import org.apache.cassandra.distributed.Cluster;
|
||||
import org.apache.cassandra.distributed.api.IInvokableInstance;
|
||||
import org.apache.cassandra.distributed.shared.ClusterUtils;
|
||||
import org.apache.cassandra.distributed.test.TestBaseImpl;
|
||||
import org.apache.cassandra.index.accord.RouteJournalIndex;
|
||||
import org.apache.cassandra.schema.TableMetadata;
|
||||
import org.apache.cassandra.service.RetryStrategy;
|
||||
import org.apache.cassandra.service.accord.AccordJournalTable;
|
||||
import org.apache.cassandra.service.accord.AccordService;
|
||||
import org.apache.cassandra.service.accord.TokenRange;
|
||||
import org.apache.cassandra.service.consensus.TransactionalMode;
|
||||
import org.apache.cassandra.utils.Shared;
|
||||
import org.apache.cassandra.utils.concurrent.CountDownLatch;
|
||||
|
||||
import static net.bytebuddy.matcher.ElementMatchers.named;
|
||||
import static org.apache.cassandra.schema.SchemaConstants.ACCORD_KEYSPACE_NAME;
|
||||
import static org.apache.cassandra.service.accord.AccordKeyspace.JOURNAL;
|
||||
|
||||
/**
|
||||
* This is a specific history of {@link StatefulJournalRestartTest}, which offers a more general way of testing this
|
||||
*/
|
||||
public class JournalAccessRouteIndexOnStartupRaceTest extends TestBaseImpl
|
||||
{
|
||||
private static final Logger logger = LoggerFactory.getLogger(JournalAccessRouteIndexOnStartupRaceTest.class);
|
||||
|
||||
@Test
|
||||
public void test() throws IOException
|
||||
{
|
||||
try (Cluster cluster = Cluster.build(1).withInstanceInitializer(BBHelper::install).start())
|
||||
{
|
||||
IInvokableInstance node = cluster.get(1);
|
||||
node.nodetoolResult("disableautocompaction", ACCORD_KEYSPACE_NAME, JOURNAL).asserts().success();
|
||||
|
||||
init(cluster);
|
||||
cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl(pk int primary key) WITH " + TransactionalMode.full.asCqlParam()));
|
||||
ClusterUtils.awaitAccordEpochReady(cluster, ClusterUtils.getCurrentEpoch(node).getEpoch());
|
||||
|
||||
insert(node, KEYSPACE, "tbl");
|
||||
|
||||
logger.info("Restarting instance with blocked 2i, triggering race condition");
|
||||
ClusterUtils.stopUnchecked(node);
|
||||
State.block();
|
||||
node.startup();
|
||||
}
|
||||
}
|
||||
|
||||
private static void insert(IInvokableInstance node, String ks, String table)
|
||||
{
|
||||
node.runOnInstance(() -> {
|
||||
AccordService accord = (AccordService) AccordService.instance();
|
||||
TableMetadata metadata = Keyspace.open(ks).getColumnFamilyStore(table).metadata();
|
||||
Ranges ranges = Ranges.single(TokenRange.fullRange(metadata.id, metadata.partitioner));
|
||||
for (int i = 0; i < 10; i++)
|
||||
{
|
||||
AsyncChains.getBlockingAndRethrow(accord.sync(null, Timestamp.NONE, ranges, null, DurabilityService.SyncLocal.Self, DurabilityService.SyncRemote.Quorum));
|
||||
|
||||
accord.journal().closeCurrentSegmentForTestingIfNonEmpty();
|
||||
accord.journal().runCompactorForTesting();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Shared
|
||||
public static class State
|
||||
{
|
||||
private static volatile CountDownLatch LATCH = null;
|
||||
|
||||
public static void block()
|
||||
{
|
||||
LATCH = CountDownLatch.newCountDownLatch(1);
|
||||
}
|
||||
|
||||
public static void unblock()
|
||||
{
|
||||
if (LATCH == null) return;
|
||||
LATCH.decrement();
|
||||
LATCH = null;
|
||||
}
|
||||
|
||||
public static void fromServer()
|
||||
{
|
||||
CountDownLatch latch = LATCH;
|
||||
if (latch == null) return;
|
||||
latch.awaitThrowUncheckedOnInterrupt();
|
||||
}
|
||||
}
|
||||
|
||||
public static class BBHelper
|
||||
{
|
||||
public static void install(ClassLoader cl, int id)
|
||||
{
|
||||
new ByteBuddy().rebase(RouteJournalIndex.class)
|
||||
.method(named("getInitializationTask"))
|
||||
.intercept(MethodDelegation.to(BBHelper.class))
|
||||
.make()
|
||||
.load(cl, ClassLoadingStrategy.Default.INJECTION);
|
||||
|
||||
new ByteBuddy().rebase(AccordJournalTable.class)
|
||||
.method(named("maybeWait"))
|
||||
.intercept(MethodDelegation.to(BBHelper.class))
|
||||
.make()
|
||||
.load(cl, ClassLoadingStrategy.Default.INJECTION);
|
||||
}
|
||||
|
||||
public static Callable<?> getInitializationTask(@SuperCall Callable<Callable<?>> zuper) throws Exception
|
||||
{
|
||||
Callable<?> delegate = zuper.call();
|
||||
return () -> {
|
||||
State.fromServer();
|
||||
return delegate.call();
|
||||
};
|
||||
}
|
||||
|
||||
public static void maybeWait(RetryStrategy retry, int i, @SuperCall Runnable zuper)
|
||||
{
|
||||
if (i > 10)
|
||||
State.unblock();
|
||||
zuper.run();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,136 @@
|
|||
/*
|
||||
* 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.accord.journal;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.time.Duration;
|
||||
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import accord.local.durability.DurabilityService;
|
||||
import accord.primitives.Ranges;
|
||||
import accord.primitives.Timestamp;
|
||||
import accord.utils.Property;
|
||||
import accord.utils.Property.SimpleCommand;
|
||||
import accord.utils.async.AsyncChains;
|
||||
import org.apache.cassandra.db.Keyspace;
|
||||
import org.apache.cassandra.distributed.Cluster;
|
||||
import org.apache.cassandra.distributed.api.IInvokableInstance;
|
||||
import org.apache.cassandra.distributed.shared.ClusterUtils;
|
||||
import org.apache.cassandra.distributed.test.TestBaseImpl;
|
||||
import org.apache.cassandra.schema.TableMetadata;
|
||||
import org.apache.cassandra.service.accord.AccordService;
|
||||
import org.apache.cassandra.service.accord.TokenRange;
|
||||
import org.apache.cassandra.service.consensus.TransactionalMode;
|
||||
|
||||
import static accord.utils.Property.commands;
|
||||
import static accord.utils.Property.stateful;
|
||||
import static org.apache.cassandra.schema.SchemaConstants.ACCORD_KEYSPACE_NAME;
|
||||
import static org.apache.cassandra.service.accord.AccordKeyspace.JOURNAL;
|
||||
|
||||
/**
|
||||
* There are 2 errors blocking this test from being run
|
||||
* <p>
|
||||
* <pre>
|
||||
* INFO [node1_AccordExecutor[0,13]] 2025-05-27 16:02:51,792 SubstituteLogger.java:169 - ERROR 23:02:51 Uncaught accord exception
|
||||
* java.lang.NullPointerException: Cannot invoke "accord.topology.Topology.nodes()" because "topology" is null
|
||||
* at org.apache.cassandra.service.accord.AccordConfigurationService.reportEpochClosed(AccordConfigurationService.java:490)
|
||||
* at accord.coordinate.CoordinationAdapter$Adapters$ExclusiveSyncPointAdapter.invokeSuccess(CoordinationAdapter.java:351)
|
||||
* at accord.coordinate.CoordinationAdapter$Adapters$SyncPointAdapter.persist(CoordinationAdapter.java:282)
|
||||
* at accord.coordinate.CoordinationAdapter.persist(CoordinationAdapter.java:80)
|
||||
* at accord.coordinate.CoordinationAdapter$Adapters$SyncPointAdapter.execute(CoordinationAdapter.java:274)
|
||||
* at accord.coordinate.CoordinationAdapter$Adapters$AbstractExclusiveSyncPointAdapter.execute(CoordinationAdapter.java:313)
|
||||
* at accord.coordinate.Propose.onAccepted(Propose.java:199)
|
||||
* at accord.coordinate.Propose.onSuccess(Propose.java:150)
|
||||
* </pre>
|
||||
* <p>
|
||||
* and
|
||||
* <p>
|
||||
* <pre>
|
||||
* INFO [node1_isolatedExecutor:1] 2025-05-27 15:38:06,880 SubstituteLogger.java:169 - ERROR 22:38:06 Exiting due to error while processing journal AccordJournal during initialization.
|
||||
* java.lang.IllegalStateException: duplicate key detected [6,1748385477237760,151(RX),1] == [6,1748385477237760,151(RX),1]
|
||||
* at accord.utils.Invariants.createIllegalState(Invariants.java:77)
|
||||
* at accord.utils.Invariants.illegalState(Invariants.java:82)
|
||||
* at accord.utils.Invariants.require(Invariants.java:272)
|
||||
* at org.apache.cassandra.service.accord.AccordJournal.replay(AccordJournal.java:452)
|
||||
* at org.apache.cassandra.service.accord.AccordService.replayJournal(AccordService.java:246)
|
||||
* at org.apache.cassandra.service.accord.AccordService.startup(AccordService.java:235)
|
||||
* at org.apache.cassandra.distributed.impl.Instance.partialStartup(Instance.java:878)
|
||||
* </pre>
|
||||
*/
|
||||
@Ignore("Unstable, need to fix")
|
||||
public class StatefulJournalRestartTest extends TestBaseImpl
|
||||
{
|
||||
private static final Logger logger = LoggerFactory.getLogger(StatefulJournalRestartTest.class);
|
||||
|
||||
@Test
|
||||
public void test() throws IOException
|
||||
{
|
||||
try (Cluster cluster = Cluster.build(1).withInstanceInitializer(JournalAccessRouteIndexOnStartupRaceTest.BBHelper::install).start())
|
||||
{
|
||||
stateful().withSeed(42).withExamples(2).withSteps(10).withStepTimeout(Duration.ofMinutes(1))
|
||||
.check(commands(() -> ignore -> setup(cluster))
|
||||
.add(new SimpleCommand<>("Insert Txn", StatefulJournalRestartTest::insert))
|
||||
.add(new SimpleCommand<>("Restart", ClusterUtils::restartUnchecked))
|
||||
.add(new SimpleCommand<>("Restart with race", StatefulJournalRestartTest::restartWithRace))
|
||||
.onSuccess((state, sut, history) -> logger.info("Successful for the following:\nState {}\nHistory:\n{}", state, Property.formatList("\t\t", history)))
|
||||
.destroyState(ClusterUtils::cleanup)
|
||||
.build());
|
||||
}
|
||||
}
|
||||
|
||||
private static IInvokableInstance setup(Cluster cluster)
|
||||
{
|
||||
IInvokableInstance node = cluster.get(1);
|
||||
node.nodetoolResult("disableautocompaction", ACCORD_KEYSPACE_NAME, JOURNAL).asserts().success();
|
||||
init(cluster);
|
||||
cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl(pk int primary key) WITH " + TransactionalMode.full.asCqlParam()));
|
||||
ClusterUtils.awaitAccordEpochReady(cluster, ClusterUtils.getCurrentEpoch(node).getEpoch());
|
||||
return node;
|
||||
}
|
||||
|
||||
private static void insert(IInvokableInstance node)
|
||||
{
|
||||
String ks = KEYSPACE;
|
||||
String table = "tbl";
|
||||
node.runOnInstance(() -> {
|
||||
AccordService accord = (AccordService) AccordService.instance();
|
||||
TableMetadata metadata = Keyspace.open(ks).getColumnFamilyStore(table).metadata();
|
||||
Ranges ranges = Ranges.single(TokenRange.fullRange(metadata.id, metadata.partitioner));
|
||||
for (int i = 0; i < 10; i++)
|
||||
{
|
||||
AsyncChains.getBlockingAndRethrow(accord.sync(null, Timestamp.NONE, ranges, null, DurabilityService.SyncLocal.Self, DurabilityService.SyncRemote.Quorum));
|
||||
|
||||
accord.journal().closeCurrentSegmentForTestingIfNonEmpty();
|
||||
accord.journal().runCompactorForTesting();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private static void restartWithRace(IInvokableInstance node)
|
||||
{
|
||||
logger.info("Restarting instance with blocked 2i, triggering race condition");
|
||||
ClusterUtils.stopUnchecked(node);
|
||||
JournalAccessRouteIndexOnStartupRaceTest.State.block();
|
||||
node.startup();
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue