mirror of https://github.com/apache/cassandra
Improve:
- Journal debugging vtable support - Background task tracing support Fix: - HLC_BOUND only valid for strictly lower HLC - HAS_UNIQUE_HLC can only be safely computed if READY_TO_EXECUTE - Break recursion in CommandStore.ensureReadyToCoordinate - Fix find intersecting shard scheduler - Separate adhoc ShardScheduler from normal to avoid overwriting - Should still use execution listeners to detect invalid reads for certain transactions even with dataStoreDetectsFutureReads patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20746
This commit is contained in:
Benedict Elliott Smith
parent
30f094080f
commit
a866f92fcf
|
|
@ -1 +1 @@
|
|||
Subproject commit 68778350bb45a1545cbe38af290d7778ffb79454
|
||||
Subproject commit 71d235d56cb315fa5ae01ec24d3d9f08dd08ac6a
|
||||
|
|
@ -18,34 +18,47 @@
|
|||
package org.apache.cassandra.db.virtual;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Date;
|
||||
import java.util.HashSet;
|
||||
import java.util.Iterator;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
import java.util.function.Function;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import com.google.common.collect.BoundType;
|
||||
import com.google.common.collect.Range;
|
||||
import com.google.common.collect.Sets;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import accord.api.RoutingKey;
|
||||
import accord.api.TraceEventType;
|
||||
import accord.impl.CommandChange;
|
||||
import accord.impl.progresslog.DefaultProgressLog;
|
||||
import accord.impl.progresslog.TxnStateKind;
|
||||
import accord.local.Command;
|
||||
import accord.local.CommandStore;
|
||||
import accord.local.CommandStores;
|
||||
import accord.local.DurableBefore;
|
||||
import accord.local.MaxConflicts;
|
||||
import accord.local.RejectBefore;
|
||||
import accord.local.StoreParticipants;
|
||||
import accord.local.durability.ShardDurability;
|
||||
import accord.primitives.Participants;
|
||||
import accord.primitives.Status;
|
||||
import accord.primitives.TxnId;
|
||||
import accord.utils.Invariants;
|
||||
import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
|
||||
import org.apache.cassandra.db.ColumnFamilyStore;
|
||||
import org.apache.cassandra.db.DataRange;
|
||||
import org.apache.cassandra.db.DecoratedKey;
|
||||
import org.apache.cassandra.db.Keyspace;
|
||||
import org.apache.cassandra.db.marshal.AbstractType;
|
||||
|
|
@ -63,13 +76,17 @@ import org.apache.cassandra.service.accord.AccordCache;
|
|||
import org.apache.cassandra.service.accord.AccordCommandStore;
|
||||
import org.apache.cassandra.service.accord.AccordCommandStores;
|
||||
import org.apache.cassandra.service.accord.AccordExecutor;
|
||||
import org.apache.cassandra.service.accord.AccordJournal;
|
||||
import org.apache.cassandra.service.accord.AccordKeyspace;
|
||||
import org.apache.cassandra.service.accord.AccordService;
|
||||
import org.apache.cassandra.service.accord.AccordTracing;
|
||||
import org.apache.cassandra.service.accord.CommandStoreTxnBlockedGraph;
|
||||
import org.apache.cassandra.service.accord.api.AccordAgent;
|
||||
import org.apache.cassandra.service.accord.api.TokenKey;
|
||||
import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState;
|
||||
import org.apache.cassandra.service.consensus.migration.TableMigrationState;
|
||||
import org.apache.cassandra.tcm.ClusterMetadata;
|
||||
import org.apache.cassandra.utils.LocalizeString;
|
||||
|
||||
import static accord.local.RedundantStatus.Property.GC_BEFORE;
|
||||
import static accord.local.RedundantStatus.Property.LOCALLY_APPLIED;
|
||||
|
|
@ -81,6 +98,8 @@ import static accord.local.RedundantStatus.Property.PRE_BOOTSTRAP;
|
|||
import static accord.local.RedundantStatus.Property.SHARD_APPLIED;
|
||||
import static java.lang.String.format;
|
||||
import static com.google.common.collect.ImmutableList.toImmutableList;
|
||||
import static java.util.concurrent.TimeUnit.NANOSECONDS;
|
||||
import static org.apache.cassandra.cql3.statements.RequestValidations.invalidRequest;
|
||||
import static org.apache.cassandra.schema.SchemaConstants.VIRTUAL_ACCORD_DEBUG;
|
||||
import static org.apache.cassandra.utils.MonotonicClock.Global.approxTime;
|
||||
|
||||
|
|
@ -95,6 +114,10 @@ public class AccordDebugKeyspace extends VirtualKeyspace
|
|||
public static final String REDUNDANT_BEFORE = "redundant_before";
|
||||
public static final String REJECT_BEFORE = "reject_before";
|
||||
public static final String TXN_BLOCKED_BY = "txn_blocked_by";
|
||||
public static final String TXN = "txn";
|
||||
public static final String JOURNAL = "journal";
|
||||
public static final String TXN_TRACE = "txn_trace";
|
||||
public static final String TXN_TRACES = "txn_traces";
|
||||
|
||||
public static final AccordDebugKeyspace instance = new AccordDebugKeyspace();
|
||||
|
||||
|
|
@ -109,11 +132,14 @@ public class AccordDebugKeyspace extends VirtualKeyspace
|
|||
new ProgressLogTable(),
|
||||
new RedundantBeforeTable(),
|
||||
new RejectBeforeTable(),
|
||||
new TxnBlockedByTable()
|
||||
new TxnBlockedByTable(),
|
||||
new TxnTable(),
|
||||
new JournalTable(),
|
||||
new TxnTraceTable(),
|
||||
new TxnTracesTable()
|
||||
));
|
||||
}
|
||||
|
||||
// TODO (consider): use a different type for the three timestamps in micros
|
||||
public static final class DurabilityServiceTable extends AbstractVirtualTable
|
||||
{
|
||||
private DurabilityServiceTable()
|
||||
|
|
@ -126,10 +152,11 @@ public class AccordDebugKeyspace extends VirtualKeyspace
|
|||
" token_sort blob,\n" +
|
||||
" token_start text,\n" +
|
||||
" token_end text,\n" +
|
||||
" last_started_at_micros bigint,\n" +
|
||||
" cycle_started_at_micros bigint,\n" +
|
||||
" last_started_at bigint,\n" +
|
||||
" cycle_started_at bigint,\n" +
|
||||
" retries int,\n" +
|
||||
" min text,\n" +
|
||||
" requested_by text,\n" +
|
||||
" active text,\n" +
|
||||
" waiting text,\n" +
|
||||
" node_offset int,\n" +
|
||||
|
|
@ -160,6 +187,9 @@ public class AccordDebugKeyspace extends VirtualKeyspace
|
|||
.column("token_end", printToken(view.shard().range.end()))
|
||||
.column("last_started_at", approxTime.translate().toMillisSinceEpoch(view.lastStartedAtMicros() * 1000))
|
||||
.column("cycle_started_at", approxTime.translate().toMillisSinceEpoch(view.cycleStartedAtMicros() * 1000))
|
||||
.column("retries", view.retries())
|
||||
.column("min", Objects.toString(view.min()))
|
||||
.column("requested_by", Objects.toString(view.requestedBy()))
|
||||
.column("active", Objects.toString(view.active()))
|
||||
.column("waiting", Objects.toString(view.waiting()))
|
||||
.column("node_offset", view.nodeOffset())
|
||||
|
|
@ -170,6 +200,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
|
|||
.column("endIndex", view.cycleLength())
|
||||
.column("current_splits", view.currentSplits())
|
||||
.column("stopping", view.stopping())
|
||||
.column("stopped", view.stopped())
|
||||
;
|
||||
}
|
||||
return ds;
|
||||
|
|
@ -476,11 +507,12 @@ public class AccordDebugKeyspace extends VirtualKeyspace
|
|||
" token_sort blob,\n" +
|
||||
" token_start text,\n" +
|
||||
" token_end text,\n" +
|
||||
" command_store_id bigint,\n" +
|
||||
" command_store_id int,\n" +
|
||||
" start_epoch bigint,\n" +
|
||||
" end_epoch bigint,\n" +
|
||||
" gc_before text,\n" +
|
||||
" shard_only_applied text,\n" +
|
||||
" shard_applied text,\n" +
|
||||
" majority_applied text,\n" +
|
||||
" locally_applied text,\n" +
|
||||
" locally_synced text,\n" +
|
||||
" locally_redundant text,\n" +
|
||||
|
|
@ -577,6 +609,358 @@ public class AccordDebugKeyspace extends VirtualKeyspace
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Usage:
|
||||
* collect N events (may be more than N messages)
|
||||
* UPDATE system_accord_debug.txn_trace SET permits = N WHERE txn_id = ? AND event_type = ?
|
||||
* SELECT * FROM system_accord_debug.txn_traces WHERE txn_id = ? AND event_type = ?
|
||||
*/
|
||||
public static final class TxnTraceTable extends AbstractMutableVirtualTable
|
||||
{
|
||||
private TxnTraceTable()
|
||||
{
|
||||
super(parse(VIRTUAL_ACCORD_DEBUG, TXN_TRACE,
|
||||
"Accord Transaction Trace Configuration",
|
||||
"CREATE TABLE %s (\n" +
|
||||
" txn_id text,\n" +
|
||||
" event_type text,\n" +
|
||||
" permits int,\n" +
|
||||
" PRIMARY KEY (txn_id, event_type)" +
|
||||
')', UTF8Type.instance));
|
||||
}
|
||||
|
||||
@Override
|
||||
public DataSet data()
|
||||
{
|
||||
AccordTracing tracing = tracing();
|
||||
SimpleDataSet dataSet = new SimpleDataSet(metadata());
|
||||
tracing.forEach(id -> true, (txnId, eventType, permits, events) -> {
|
||||
dataSet.row(txnId.toString(), eventType.toString()).column("permits", permits);
|
||||
});
|
||||
return dataSet;
|
||||
}
|
||||
|
||||
private AccordTracing tracing()
|
||||
{
|
||||
return ((AccordAgent)AccordService.instance().agent()).tracing();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void applyPartitionDeletion(ColumnValues partitionKey)
|
||||
{
|
||||
TxnId txnId = TxnId.parse(partitionKey.value(0));
|
||||
tracing().erasePermits(txnId);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void applyRowDeletion(ColumnValues partitionKey, ColumnValues clusteringColumns)
|
||||
{
|
||||
TxnId txnId = TxnId.parse(partitionKey.value(0));
|
||||
tracing().erasePermits(txnId, parseEventType(clusteringColumns.value(0)));
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void applyColumnDeletion(ColumnValues partitionKey, ColumnValues clusteringColumns, String columnName)
|
||||
{
|
||||
TxnId txnId = TxnId.parse(partitionKey.value(0));
|
||||
TraceEventType eventType = parseEventType(clusteringColumns.value(0));
|
||||
tracing().erasePermits(txnId, eventType);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void applyColumnUpdate(ColumnValues partitionKey, ColumnValues clusteringColumns, Optional<ColumnValue> columnValue)
|
||||
{
|
||||
TxnId txnId = TxnId.parse(partitionKey.value(0));
|
||||
TraceEventType eventType = parseEventType(clusteringColumns.value(0));
|
||||
if (columnValue.isEmpty()) tracing().erasePermits(txnId, eventType);
|
||||
else tracing().setPermits(txnId, eventType, columnValue.get().value());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void truncate()
|
||||
{
|
||||
tracing().eraseAllEvents();
|
||||
}
|
||||
}
|
||||
|
||||
public static final class TxnTracesTable extends AbstractMutableVirtualTable
|
||||
{
|
||||
private TxnTracesTable()
|
||||
{
|
||||
super(parse(VIRTUAL_ACCORD_DEBUG, TXN_TRACES,
|
||||
"Accord Transaction Traces",
|
||||
"CREATE TABLE %s (\n" +
|
||||
" txn_id text,\n" +
|
||||
" event_type text,\n" +
|
||||
" id_micros bigint,\n" +
|
||||
" at_micros bigint,\n" +
|
||||
" command_store_id int,\n" +
|
||||
" message text,\n" +
|
||||
" PRIMARY KEY (txn_id, event_type, id_micros, at_micros)" +
|
||||
')', UTF8Type.instance));
|
||||
}
|
||||
|
||||
private AccordTracing tracing()
|
||||
{
|
||||
return ((AccordAgent)AccordService.instance().agent()).tracing();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void applyPartitionDeletion(ColumnValues partitionKey)
|
||||
{
|
||||
TxnId txnId = TxnId.parse(partitionKey.value(0));
|
||||
tracing().eraseEvents(txnId);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void applyRangeTombstone(ColumnValues partitionKey, Range<ColumnValues> range)
|
||||
{
|
||||
TxnId txnId = TxnId.parse(partitionKey.value(0));
|
||||
if (!range.hasLowerBound() || range.lowerBoundType() != BoundType.CLOSED) throw invalidRequest("May restrict deletion by at most one event_type");
|
||||
if (range.lowerEndpoint().size() != 1) throw invalidRequest("Deletion restricted by lower bound on id_micros or at_micros is unsupported");
|
||||
if (!range.hasUpperBound() || (range.upperBoundType() != BoundType.CLOSED && range.upperEndpoint().size() == 1)) throw invalidRequest("Range deletion must specify one event_type");
|
||||
if (!range.upperEndpoint().value(0).equals(range.lowerEndpoint().value(0))) throw invalidRequest("May restrict deletion by at most one event_type");
|
||||
if (range.upperEndpoint().size() > 2) throw invalidRequest("Deletion restricted by upper bound on at_micros is unsupported");
|
||||
TraceEventType eventType = parseEventType(range.lowerEndpoint().value(0));
|
||||
if (range.upperEndpoint().size() == 1)
|
||||
{
|
||||
tracing().eraseEvents(txnId, eventType);
|
||||
}
|
||||
else
|
||||
{
|
||||
long before = range.upperEndpoint().value(1);
|
||||
tracing().eraseEventsBefore(txnId, eventType, before);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void truncate()
|
||||
{
|
||||
tracing().eraseAllEvents();
|
||||
}
|
||||
|
||||
@Override
|
||||
public DataSet data()
|
||||
{
|
||||
SimpleDataSet dataSet = new SimpleDataSet(metadata());
|
||||
tracing().forEach(id -> true, (txnId, eventType, permits, events) -> {
|
||||
events.forEach(e -> {
|
||||
e.messages().forEach(m -> {
|
||||
dataSet.row(txnId.toString(), eventType.name(), e.idMicros, NANOSECONDS.toMicros(m.atNanos - e.atNanos))
|
||||
.column("command_store_id", m.commandStoreId)
|
||||
.column("message", m.message);
|
||||
});
|
||||
});
|
||||
});
|
||||
return dataSet;
|
||||
}
|
||||
}
|
||||
|
||||
// TODO (desired): don't report null as "null"
|
||||
public static final class TxnTable extends AbstractVirtualTable implements AbstractVirtualTable.DataSet
|
||||
{
|
||||
static class Entry
|
||||
{
|
||||
final int commandStoreId;
|
||||
final Command command;
|
||||
|
||||
Entry(int commandStoreId, Command command)
|
||||
{
|
||||
this.commandStoreId = commandStoreId;
|
||||
this.command = command;
|
||||
}
|
||||
}
|
||||
private TxnTable()
|
||||
{
|
||||
super(parse(VIRTUAL_ACCORD_DEBUG, TXN,
|
||||
"Accord per-CommandStore Transaction State",
|
||||
"CREATE TABLE %s (\n" +
|
||||
" txn_id text,\n" +
|
||||
" command_store_id int,\n" +
|
||||
" save_status text,\n" +
|
||||
" route text,\n" +
|
||||
" durability text,\n" +
|
||||
" execute_at text,\n" +
|
||||
" executes_at_least text,\n" +
|
||||
" txn text,\n" +
|
||||
" deps text,\n" +
|
||||
" waiting_on text,\n" +
|
||||
" writes text,\n" +
|
||||
" result text,\n" +
|
||||
" participants_owns text,\n" +
|
||||
" participants_touches text,\n" +
|
||||
" participants_has_touched text,\n" +
|
||||
" participants_executes text,\n" +
|
||||
" participants_waits_on text,\n" +
|
||||
" PRIMARY KEY (txn_id, command_store_id)" +
|
||||
')', UTF8Type.instance));
|
||||
}
|
||||
|
||||
@Override
|
||||
public DataSet data()
|
||||
{
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isEmpty()
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Partition getPartition(DecoratedKey partitionKey)
|
||||
{
|
||||
String txnIdStr = UTF8Type.instance.compose(partitionKey.getKey());
|
||||
TxnId txnId = TxnId.parse(txnIdStr);
|
||||
|
||||
List<Entry> commands = new ArrayList<>();
|
||||
AccordService.instance().node().commandStores().forEachCommandStore(store -> {
|
||||
Command command = ((AccordCommandStore)store).loadCommand(txnId);
|
||||
if (command != null)
|
||||
commands.add(new Entry(store.id(), command));
|
||||
});
|
||||
|
||||
if (commands.isEmpty())
|
||||
return null;
|
||||
|
||||
SimpleDataSet ds = new SimpleDataSet(metadata);
|
||||
for (Entry e : commands)
|
||||
{
|
||||
Command command = e.command;
|
||||
ds.row(txnIdStr, e.commandStoreId)
|
||||
.column("save_status", Objects.toString(command.saveStatus()))
|
||||
.column("route", Objects.toString(command.route()))
|
||||
.column("participants_owns", toStr(command, StoreParticipants::owns, StoreParticipants::stillOwns))
|
||||
.column("participants_touches", toStr(command, StoreParticipants::touches, StoreParticipants::stillTouches))
|
||||
.column("participants_has_touched", Objects.toString(command.participants().hasTouched()))
|
||||
.column("participants_executes", toStr(command, StoreParticipants::executes, StoreParticipants::stillExecutes))
|
||||
.column("participants_waits_on", toStr(command, StoreParticipants::waitsOn, StoreParticipants::stillWaitsOn))
|
||||
.column("durability", Objects.toString(command.durability()))
|
||||
.column("execute_at", Objects.toString(command.executeAt()))
|
||||
.column("executes_at_least", Objects.toString(command.executesAtLeast()))
|
||||
.column("txn", Objects.toString(command.partialTxn()))
|
||||
.column("deps", Objects.toString(command.partialDeps()))
|
||||
.column("waiting_on", Objects.toString(command.waitingOn()))
|
||||
.column("writes", Objects.toString(command.writes()))
|
||||
.column("result", Objects.toString(command.result()));
|
||||
}
|
||||
|
||||
return ds.getPartition(partitionKey);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Iterator<Partition> getPartitions(DataRange range)
|
||||
{
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
}
|
||||
|
||||
public static final class JournalTable extends AbstractVirtualTable implements AbstractVirtualTable.DataSet
|
||||
{
|
||||
static class Entry
|
||||
{
|
||||
final int commandStoreId;
|
||||
final long segment;
|
||||
final int position;
|
||||
final CommandChange.Builder builder;
|
||||
|
||||
Entry(int commandStoreId, long segment, int position, CommandChange.Builder builder)
|
||||
{
|
||||
this.commandStoreId = commandStoreId;
|
||||
this.segment = segment;
|
||||
this.position = position;
|
||||
this.builder = builder;
|
||||
}
|
||||
}
|
||||
|
||||
private JournalTable()
|
||||
{
|
||||
super(parse(VIRTUAL_ACCORD_DEBUG, JOURNAL,
|
||||
"Accord per-CommandStore Journal State",
|
||||
"CREATE TABLE %s (\n" +
|
||||
" txn_id text,\n" +
|
||||
" command_store_id int,\n" +
|
||||
" segment bigint,\n" +
|
||||
" segment_position int,\n" +
|
||||
" save_status text,\n" +
|
||||
" route text,\n" +
|
||||
" durability text,\n" +
|
||||
" execute_at text,\n" +
|
||||
" executes_at_least text,\n" +
|
||||
" txn text,\n" +
|
||||
" deps text,\n" +
|
||||
" writes text,\n" +
|
||||
" result text,\n" +
|
||||
" participants_owns text,\n" +
|
||||
" participants_touches text,\n" +
|
||||
" participants_has_touched text,\n" +
|
||||
" participants_executes text,\n" +
|
||||
" participants_waits_on text,\n" +
|
||||
" PRIMARY KEY (txn_id, command_store_id, segment, segment_position)" +
|
||||
')', UTF8Type.instance));
|
||||
}
|
||||
|
||||
@Override
|
||||
public DataSet data()
|
||||
{
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isEmpty()
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Partition getPartition(DecoratedKey partitionKey)
|
||||
{
|
||||
String txnIdStr = UTF8Type.instance.compose(partitionKey.getKey());
|
||||
TxnId txnId = TxnId.parse(txnIdStr);
|
||||
|
||||
List<Entry> entries = new ArrayList<>();
|
||||
AccordService.instance().node().commandStores().forEachCommandStore(store -> {
|
||||
for (AccordJournal.DebugEntry e : ((AccordCommandStore)store).debugCommand(txnId))
|
||||
entries.add(new Entry(store.id(), e.segment, e.position, e.builder));
|
||||
});
|
||||
|
||||
if (entries.isEmpty())
|
||||
return null;
|
||||
|
||||
SimpleDataSet ds = new SimpleDataSet(metadata);
|
||||
for (Entry e : entries)
|
||||
{
|
||||
CommandChange.Builder b = e.builder;
|
||||
StoreParticipants participants = b.participants();
|
||||
if (participants == null) participants = StoreParticipants.empty(txnId);
|
||||
ds.row(txnIdStr, e.commandStoreId, e.segment, e.position)
|
||||
.column("save_status", Objects.toString(b.saveStatus()))
|
||||
.column("route", Objects.toString(participants.route()))
|
||||
.column("participants_owns", toStr(participants, StoreParticipants::owns, StoreParticipants::stillOwns))
|
||||
.column("participants_touches", toStr(participants, StoreParticipants::touches, StoreParticipants::stillTouches))
|
||||
.column("participants_has_touched", Objects.toString(participants.hasTouched()))
|
||||
.column("participants_executes", toStr(participants, StoreParticipants::executes, StoreParticipants::stillExecutes))
|
||||
.column("participants_waits_on", toStr(participants, StoreParticipants::waitsOn, StoreParticipants::stillWaitsOn))
|
||||
.column("durability", Objects.toString(b.durability()))
|
||||
.column("execute_at", Objects.toString(b.executeAt()))
|
||||
.column("executes_at_least", Objects.toString(b.executesAtLeast()))
|
||||
.column("txn", Objects.toString(b.partialTxn()))
|
||||
.column("deps", Objects.toString(b.partialDeps()))
|
||||
.column("writes", Objects.toString(b.writes()))
|
||||
.column("result", Objects.toString(b.result()));
|
||||
}
|
||||
|
||||
return ds.getPartition(partitionKey);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Iterator<Partition> getPartitions(DataRange range)
|
||||
{
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
}
|
||||
|
||||
public static class TxnBlockedByTable extends AbstractVirtualTable
|
||||
{
|
||||
enum Reason { Self, Txn, Key }
|
||||
|
|
@ -716,4 +1100,24 @@ public class AccordDebugKeyspace extends VirtualKeyspace
|
|||
.partitioner(new LocalPartitioner(partitionKeyType))
|
||||
.build();
|
||||
}
|
||||
|
||||
private static String toStr(Command command, Function<StoreParticipants, Participants<?>> a, Function<StoreParticipants, Participants<?>> b)
|
||||
{
|
||||
return toStr(command.participants(), a, b);
|
||||
}
|
||||
|
||||
private static String toStr(StoreParticipants participants, Function<StoreParticipants, Participants<?>> a, Function<StoreParticipants, Participants<?>> b)
|
||||
{
|
||||
Participants<?> av = a.apply(participants);
|
||||
Participants<?> bv = b.apply(participants);
|
||||
if (av == bv || av.equals(bv))
|
||||
return Objects.toString(av);
|
||||
return av + " (" + bv + ')';
|
||||
}
|
||||
|
||||
private static TraceEventType parseEventType(String input)
|
||||
{
|
||||
try { return TraceEventType.valueOf(LocalizeString.toUpperCaseLocalized(input, Locale.ENGLISH)); }
|
||||
catch (Throwable t) { throw invalidRequest("event_type must be one of %s; received %s", TraceEventType.values(), input); }
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -443,6 +443,12 @@ public class AccordCommandStore extends CommandStore
|
|||
return journal.loadCommand(id, txnId, unsafeGetRedundantBefore(), durableBefore());
|
||||
}
|
||||
|
||||
@VisibleForTesting
|
||||
public List<AccordJournal.DebugEntry> debugCommand(TxnId txnId)
|
||||
{
|
||||
return (List<AccordJournal.DebugEntry>) journal.debugCommand(id, txnId);
|
||||
}
|
||||
|
||||
public static Command prepareToCache(Command command)
|
||||
{
|
||||
// TODO (required): validate we don't have duplicate objects
|
||||
|
|
|
|||
|
|
@ -19,13 +19,16 @@ package org.apache.cassandra.service.accord;
|
|||
|
||||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.NavigableMap;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Supplier;
|
||||
import javax.annotation.Nullable;
|
||||
|
||||
import com.google.common.annotations.VisibleForTesting;
|
||||
|
|
@ -228,6 +231,39 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
|
|||
return builder.construct(redundantBefore);
|
||||
}
|
||||
|
||||
public static class DebugEntry implements Supplier<CommandChange.Builder>
|
||||
{
|
||||
public final long segment;
|
||||
public final int position;
|
||||
public final Builder builder;
|
||||
|
||||
public DebugEntry(long segment, int position, Builder builder)
|
||||
{
|
||||
this.segment = segment;
|
||||
this.position = position;
|
||||
this.builder = builder;
|
||||
}
|
||||
|
||||
@Override
|
||||
public CommandChange.Builder get()
|
||||
{
|
||||
return builder;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<DebugEntry> debugCommand(int commandStoreId, TxnId txnId)
|
||||
{
|
||||
JournalKey key = new JournalKey(txnId, JournalKey.Type.COMMAND_DIFF, commandStoreId);
|
||||
List<DebugEntry> result = new ArrayList<>();
|
||||
journalTable.readAll(key, (long segment, int position, JournalKey k, ByteBuffer buffer, int userVersion) -> {
|
||||
Builder builder = new Builder(txnId);
|
||||
new AccordJournalTable.RecordConsumerAdapter<>(builder::deserializeNext).accept(segment, position, k, buffer, userVersion);
|
||||
result.add(new DebugEntry(segment, position, builder));
|
||||
});
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Command.Minimal loadMinimal(int commandStoreId, TxnId txnId, Load load, RedundantBefore redundantBefore, DurableBefore durableBefore)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -192,7 +192,7 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
|
|||
void read(DataInputPlus input, Version userVersion) throws IOException;
|
||||
}
|
||||
|
||||
private static class RecordConsumerAdapter<K> implements RecordConsumer<K>
|
||||
static class RecordConsumerAdapter<K> implements RecordConsumer<K>
|
||||
{
|
||||
protected final Reader reader;
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,324 @@
|
|||
/*
|
||||
* 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.service.accord;
|
||||
|
||||
import java.util.AbstractList;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.EnumMap;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.function.BiFunction;
|
||||
import java.util.function.Predicate;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import accord.api.Tracing;
|
||||
import accord.api.TraceEventType;
|
||||
import accord.local.CommandStore;
|
||||
import accord.primitives.TxnId;
|
||||
import org.apache.cassandra.utils.Clock;
|
||||
import org.apache.cassandra.utils.NoSpamLogger;
|
||||
|
||||
public class AccordTracing
|
||||
{
|
||||
private static final int MAX_EVENTS = 10000;
|
||||
private static final Logger logger = LoggerFactory.getLogger(AccordTracing.class);
|
||||
private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1L, TimeUnit.MINUTES);
|
||||
|
||||
public interface ConsumeState
|
||||
{
|
||||
void accept(TxnId txnId, TraceEventType eventType, int permits, List<Event> events);
|
||||
}
|
||||
|
||||
public static class Message
|
||||
{
|
||||
public final long atNanos;
|
||||
public final int commandStoreId;
|
||||
public final String message;
|
||||
|
||||
Message(int commandStoreId, String message, long atLeastNanos)
|
||||
{
|
||||
this.commandStoreId = commandStoreId;
|
||||
this.message = message;
|
||||
this.atNanos = Math.max(atLeastNanos, Clock.Global.nanoTime());
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString()
|
||||
{
|
||||
return message;
|
||||
}
|
||||
}
|
||||
|
||||
public static class Event implements Tracing, Comparable<Event>
|
||||
{
|
||||
public final long idMicros = uniqueNowMicros();
|
||||
public final long atNanos = Clock.Global.nanoTime();
|
||||
final List<Message> messages = new ArrayList<>();
|
||||
|
||||
@Override
|
||||
public void trace(CommandStore commandStore, String s)
|
||||
{
|
||||
long prevNanos = messages.isEmpty() ? 0 : messages.get(messages.size() - 1).atNanos;
|
||||
int id = commandStore == null ? -1 : commandStore.id();
|
||||
if (s.length() > 1000)
|
||||
s = s.substring(0, 1000);
|
||||
messages.add(new Message(id, s, prevNanos + 1));
|
||||
}
|
||||
|
||||
@Override
|
||||
public int compareTo(Event that)
|
||||
{
|
||||
return Long.compareUnsigned(this.idMicros, that.idMicros);
|
||||
}
|
||||
|
||||
public List<Message> messages()
|
||||
{
|
||||
return Collections.unmodifiableList(messages);
|
||||
}
|
||||
}
|
||||
|
||||
static class TraceState extends AbstractList<Event>
|
||||
{
|
||||
int permits;
|
||||
int size;
|
||||
Event[] events;
|
||||
|
||||
void addInternal(Event event)
|
||||
{
|
||||
if (events == null) events = new Event[10];
|
||||
else if (size == events.length) events = Arrays.copyOf(events, events.length * 2);
|
||||
events[size++] = event;
|
||||
}
|
||||
|
||||
void truncate(int eraseBefore)
|
||||
{
|
||||
System.arraycopy(events, eraseBefore, events, 0, size - eraseBefore);
|
||||
Arrays.fill(events, size - eraseBefore, size, null);
|
||||
size -= eraseBefore;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Event get(int index)
|
||||
{
|
||||
return events[index];
|
||||
}
|
||||
|
||||
@Override
|
||||
public int size()
|
||||
{
|
||||
return size;
|
||||
}
|
||||
}
|
||||
|
||||
private static final AtomicLong lastNowMicros = new AtomicLong();
|
||||
private static long uniqueNowMicros()
|
||||
{
|
||||
long nowMicros = Clock.Global.currentTimeMillis() * 1000;
|
||||
while (true)
|
||||
{
|
||||
long last = lastNowMicros.get();
|
||||
if (last >= nowMicros)
|
||||
return lastNowMicros.incrementAndGet();
|
||||
if (lastNowMicros.compareAndSet(last, nowMicros))
|
||||
return nowMicros;
|
||||
}
|
||||
}
|
||||
|
||||
final Map<TxnId, EnumMap<TraceEventType, TraceState>> stateMap = new ConcurrentHashMap<>();
|
||||
final AtomicInteger count = new AtomicInteger();
|
||||
|
||||
public Tracing trace(TxnId txnId, TraceEventType eventType)
|
||||
{
|
||||
if (!stateMap.containsKey(txnId))
|
||||
return null;
|
||||
|
||||
class Register implements BiFunction<TxnId, EnumMap<TraceEventType, TraceState>, EnumMap<TraceEventType, TraceState>>
|
||||
{
|
||||
Event event;
|
||||
|
||||
@Override
|
||||
public EnumMap<TraceEventType, TraceState> apply(TxnId id, EnumMap<TraceEventType, TraceState> cur)
|
||||
{
|
||||
if (cur == null)
|
||||
return null;
|
||||
|
||||
TraceState curState = cur.get(eventType);
|
||||
if (curState == null || curState.permits == 0)
|
||||
return cur;
|
||||
|
||||
if (count.incrementAndGet() >= MAX_EVENTS)
|
||||
{
|
||||
count.decrementAndGet();
|
||||
noSpamLogger.warn("Too many Accord trace events stored already; delete some to continue tracing");
|
||||
}
|
||||
else
|
||||
{
|
||||
curState.permits--;
|
||||
curState.addInternal(event = new Event());
|
||||
}
|
||||
return cur;
|
||||
}
|
||||
}
|
||||
Register register = new Register();
|
||||
stateMap.compute(txnId, register);
|
||||
return register.event;
|
||||
}
|
||||
|
||||
public void setPermits(TxnId txnId, TraceEventType eventType, int newPermits)
|
||||
{
|
||||
stateMap.compute(txnId, (id, cur) -> {
|
||||
if (newPermits != 0)
|
||||
{
|
||||
if (cur == null)
|
||||
cur = new EnumMap<>(TraceEventType.class);
|
||||
cur.computeIfAbsent(eventType, ignore -> new TraceState()).permits = newPermits;
|
||||
}
|
||||
else if (cur != null)
|
||||
{
|
||||
TraceState curState = cur.get(eventType);
|
||||
if (curState != null)
|
||||
{
|
||||
if (!curState.isEmpty()) curState.permits = 0;
|
||||
else
|
||||
{
|
||||
cur.remove(eventType);
|
||||
if (cur.isEmpty())
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
return cur;
|
||||
});
|
||||
}
|
||||
|
||||
public void erasePermits(TxnId txnId)
|
||||
{
|
||||
stateMap.compute(txnId, (id, cur) -> {
|
||||
if (cur == null)
|
||||
return null;
|
||||
|
||||
Iterator<TraceState> iter = cur.values().iterator();
|
||||
while (iter.hasNext())
|
||||
{
|
||||
TraceState state = iter.next();
|
||||
state.permits = 0;
|
||||
if (state.isEmpty()) iter.remove();
|
||||
}
|
||||
return cur.isEmpty() ? null : cur;
|
||||
});
|
||||
}
|
||||
|
||||
public void erasePermits(TxnId txnId, TraceEventType eventType)
|
||||
{
|
||||
setPermits(txnId, eventType, 0);
|
||||
}
|
||||
|
||||
public void eraseEvents(TxnId txnId)
|
||||
{
|
||||
stateMap.compute(txnId, (id, cur) -> {
|
||||
if (cur == null)
|
||||
return null;
|
||||
|
||||
Iterator<TraceState> iter = cur.values().iterator();
|
||||
while (iter.hasNext())
|
||||
{
|
||||
TraceState state = iter.next();
|
||||
count.addAndGet(-state.size());
|
||||
state.truncate(state.size());
|
||||
if (state.permits == 0) iter.remove();
|
||||
}
|
||||
return cur.isEmpty() ? null : cur;
|
||||
});
|
||||
}
|
||||
|
||||
public void eraseEvents(TxnId txnId, TraceEventType eventType)
|
||||
{
|
||||
stateMap.compute(txnId, (id, cur) -> {
|
||||
if (cur != null)
|
||||
{
|
||||
TraceState state = cur.get(eventType);
|
||||
if (state == null)
|
||||
return cur;
|
||||
|
||||
count.addAndGet(-state.size());
|
||||
state.truncate(state.size());
|
||||
if (state.permits == 0)
|
||||
cur.remove(eventType);
|
||||
if (cur.isEmpty())
|
||||
return null;
|
||||
}
|
||||
return cur;
|
||||
});
|
||||
}
|
||||
|
||||
public void eraseEventsBefore(TxnId txnId, TraceEventType eventType, long timestamp)
|
||||
{
|
||||
stateMap.compute(txnId, (id, cur) -> {
|
||||
if (cur != null)
|
||||
{
|
||||
TraceState state = cur.get(eventType);
|
||||
if (state == null)
|
||||
return cur;
|
||||
|
||||
int i = 0;
|
||||
while (i < state.size() && state.get(i).idMicros < timestamp)
|
||||
++i;
|
||||
state.truncate(i);
|
||||
count.addAndGet(-i);
|
||||
if (cur.isEmpty())
|
||||
return null;
|
||||
}
|
||||
return cur;
|
||||
});
|
||||
}
|
||||
|
||||
public void eraseAllEvents()
|
||||
{
|
||||
stateMap.keySet().forEach(this::eraseEvents);
|
||||
}
|
||||
|
||||
public void eraseAllPermits()
|
||||
{
|
||||
stateMap.keySet().forEach(this::erasePermits);
|
||||
}
|
||||
|
||||
public void forEach(Predicate<TxnId> include, ConsumeState forEach)
|
||||
{
|
||||
stateMap.forEach((txnId, state) -> {
|
||||
if (include.test(txnId))
|
||||
{
|
||||
// ensure lock is held for duration of callback
|
||||
stateMap.compute(txnId, (id, cur) -> {
|
||||
if (cur != null)
|
||||
cur.forEach((event, events) -> forEach.accept(txnId, event, events.permits, Collections.unmodifiableList(events)));
|
||||
return cur;
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
@ -22,6 +22,8 @@ import java.util.concurrent.CancellationException;
|
|||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.BiConsumer;
|
||||
|
||||
import javax.annotation.Nullable;
|
||||
|
||||
import com.google.common.annotations.VisibleForTesting;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
|
@ -31,6 +33,8 @@ import accord.api.EventListener;
|
|||
import accord.api.ProgressLog.BlockedUntil;
|
||||
import accord.api.Result;
|
||||
import accord.api.RoutingKey;
|
||||
import accord.api.Tracing;
|
||||
import accord.api.TraceEventType;
|
||||
import accord.local.Command;
|
||||
import accord.local.Node;
|
||||
import accord.local.SafeCommand;
|
||||
|
|
@ -61,6 +65,7 @@ import org.apache.cassandra.exceptions.RequestTimeoutException;
|
|||
import org.apache.cassandra.metrics.AccordMetrics;
|
||||
import org.apache.cassandra.net.ResponseContext;
|
||||
import org.apache.cassandra.service.accord.AccordService;
|
||||
import org.apache.cassandra.service.accord.AccordTracing;
|
||||
import org.apache.cassandra.service.accord.serializers.TableMetadatasAndKeys;
|
||||
import org.apache.cassandra.service.accord.txn.TxnQuery;
|
||||
import org.apache.cassandra.service.accord.txn.TxnRead;
|
||||
|
|
@ -98,7 +103,7 @@ public class AccordAgent implements Agent
|
|||
if (invoke != null) invoke.accept(txnId, cause);
|
||||
}
|
||||
|
||||
|
||||
private final AccordTracing tracing = new AccordTracing();
|
||||
private final RandomSource random = new DefaultRandom();
|
||||
protected Node.Id self;
|
||||
|
||||
|
|
@ -106,6 +111,17 @@ public class AccordAgent implements Agent
|
|||
{
|
||||
}
|
||||
|
||||
public AccordTracing tracing()
|
||||
{
|
||||
return tracing;
|
||||
}
|
||||
|
||||
@Override
|
||||
public @Nullable Tracing trace(TxnId txnId, TraceEventType eventType)
|
||||
{
|
||||
return tracing.trace(txnId, eventType);
|
||||
}
|
||||
|
||||
public void setNodeId(Node.Id id)
|
||||
{
|
||||
self = id;
|
||||
|
|
|
|||
|
|
@ -51,6 +51,7 @@ import org.apache.cassandra.service.accord.AccordService;
|
|||
import org.apache.cassandra.tcm.ClusterMetadata;
|
||||
import org.apache.cassandra.utils.ByteBufferUtil;
|
||||
import org.apache.cassandra.utils.concurrent.Condition;
|
||||
import org.assertj.core.api.Assertions;
|
||||
import org.awaitility.Awaitility;
|
||||
|
||||
import static accord.api.ProtocolModifiers.Toggles.SendStableMessages.TO_ALL;
|
||||
|
|
@ -65,6 +66,39 @@ public class AccordDebugKeyspaceTest extends CQLTester
|
|||
private static final String QUERY_TXN_BLOCKED_BY =
|
||||
String.format("SELECT * FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_BLOCKED_BY);
|
||||
|
||||
private static final String QUERY_TXN =
|
||||
String.format("SELECT txn_id, save_status FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN);
|
||||
|
||||
private static final String QUERY_JOURNAL =
|
||||
String.format("SELECT txn_id, save_status FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.JOURNAL);
|
||||
|
||||
private static final String SET_TRACE =
|
||||
String.format("UPDATE %s.%s SET permits = ? WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACE);
|
||||
|
||||
private static final String QUERY_TRACE =
|
||||
String.format("SELECT * FROM %s.%s WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACE);
|
||||
|
||||
private static final String UNSET_TRACE1 =
|
||||
String.format("DELETE FROM %s.%s WHERE txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACE);
|
||||
|
||||
private static final String UNSET_TRACE2 =
|
||||
String.format("DELETE FROM %s.%s WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACE);
|
||||
|
||||
private static final String QUERY_TRACES =
|
||||
String.format("SELECT * FROM %s.%s WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
|
||||
|
||||
private static final String ERASE_TRACES1 =
|
||||
String.format("DELETE FROM %s.%s WHERE txn_id = ? AND event_type = ? AND id_micros < ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
|
||||
|
||||
private static final String ERASE_TRACES2 =
|
||||
String.format("DELETE FROM %s.%s WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
|
||||
|
||||
private static final String ERASE_TRACES3 =
|
||||
String.format("DELETE FROM %s.%s WHERE txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
|
||||
|
||||
private static final String QUERY_REDUNDANT_BEFORE =
|
||||
String.format("SELECT * FROM %s.%s", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.REDUNDANT_BEFORE);
|
||||
|
||||
@BeforeClass
|
||||
public static void setUpClass()
|
||||
{
|
||||
|
|
@ -85,6 +119,63 @@ public class AccordDebugKeyspaceTest extends CQLTester
|
|||
{
|
||||
createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
|
||||
assertRows(execute(QUERY_TXN_BLOCKED_BY, TxnId.NONE.toString()));
|
||||
assertRows(execute(QUERY_TXN, TxnId.NONE.toString()));
|
||||
assertRows(execute(QUERY_JOURNAL, TxnId.NONE.toString()));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void tracing()
|
||||
{
|
||||
// simple test to confirm basic tracing functionality works, doesn't validate specific behaviours only requesting/querying/erasing
|
||||
AccordMsgFilter filter = new AccordMsgFilter();
|
||||
MessagingService.instance().outboundSink.add(filter);
|
||||
try
|
||||
{
|
||||
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
|
||||
var accord = accord();
|
||||
DatabaseDescriptor.getAccord().fetch_txn = "1s";
|
||||
TxnId id = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key);
|
||||
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0);
|
||||
|
||||
execute(SET_TRACE, 1, id.toString(), "PROGRESS");
|
||||
assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1));
|
||||
execute(SET_TRACE, 0, id.toString(), "PROGRESS");
|
||||
assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"));
|
||||
execute(SET_TRACE, 1, id.toString(), "PROGRESS");
|
||||
assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1));
|
||||
execute(UNSET_TRACE1, id.toString());
|
||||
assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"));
|
||||
execute(SET_TRACE, 1, id.toString(), "PROGRESS");
|
||||
assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1));
|
||||
execute(UNSET_TRACE2, id.toString(), "PROGRESS");
|
||||
assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"));
|
||||
execute(SET_TRACE, 1, id.toString(), "PROGRESS");
|
||||
assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1));
|
||||
accord.node().coordinate(id, txn);
|
||||
filter.preAccept.awaitThrowUncheckedOnInterrupt();
|
||||
|
||||
filter.apply.awaitThrowUncheckedOnInterrupt();
|
||||
spinUntilSuccess(() -> Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "PROGRESS").size()).isGreaterThan(0));
|
||||
execute(ERASE_TRACES1, id.toString(), "FETCH", Long.MAX_VALUE);
|
||||
execute(ERASE_TRACES2, id.toString(), "FETCH");
|
||||
execute(ERASE_TRACES1, id.toString(), "PROGRESS", Long.MAX_VALUE);
|
||||
Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "PROGRESS").size()).isEqualTo(0);
|
||||
// just check other variants don't fail
|
||||
execute(ERASE_TRACES2, id.toString(), "PROGRESS");
|
||||
execute(ERASE_TRACES3, id.toString());
|
||||
}
|
||||
finally
|
||||
{
|
||||
MessagingService.instance().outboundSink.remove(filter);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void redundantBefore()
|
||||
{
|
||||
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
|
||||
var accord = accord();
|
||||
Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE).size()).isGreaterThan(0);
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -98,6 +189,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
|
|||
|
||||
spinUntilSuccess(() -> assertRows(execute(QUERY_TXN_BLOCKED_BY, id.toString()),
|
||||
row(id.toString(), KEYSPACE, tableName, anyInt(), 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, "Self", any(), null, anyOf(SaveStatus.ReadyToExecute.name(), SaveStatus.Applying.name(), SaveStatus.Applied.name()))));
|
||||
assertRows(execute(QUERY_TXN, id.toString()), row(id.toString(), "Applied"));
|
||||
assertRows(execute(QUERY_JOURNAL, id.toString()), row(id.toString(), "PreAccepted"), row(id.toString(), "Applying"), row(id.toString(), "Applied"), row(id.toString(), "null"));
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
|
|||
Loading…
Reference in New Issue