From ccf12ef84545b5ebd4fc91f1bd04a74f120b06f9 Mon Sep 17 00:00:00 2001 From: Alex Petrov Date: Tue, 1 Jul 2025 07:52:33 +0200 Subject: [PATCH] Add accord journal standalone dump tool Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20738 --- .gitmodules | 2 +- .../apache/cassandra/config/AccordSpec.java | 3 +- .../cassandra/config/DatabaseDescriptor.java | 2 +- .../apache/cassandra/journal/Descriptor.java | 2 +- .../apache/cassandra/journal/DumpUtil.java | 42 +++ .../apache/cassandra/journal/Metadata.java | 9 + .../cassandra/journal/StaticSegment.java | 2 +- .../cassandra/net/InboundMessageHandler.java | 10 +- .../cassandra/schema/SchemaConstants.java | 8 + .../service/accord/AccordJournal.java | 7 +- .../accord/AccordJournalValueSerializers.java | 15 + .../cassandra/tcm/ClusterMetadataService.java | 53 ++- .../tools/StandaloneJournalUtil.java | 353 ++++++++++++++++++ .../utils/btree/FullBTreeSearchIterator.java | 2 - .../accord/WatermarkCollectorTest.java | 3 +- 15 files changed, 500 insertions(+), 13 deletions(-) create mode 100644 src/java/org/apache/cassandra/journal/DumpUtil.java create mode 100644 src/java/org/apache/cassandra/tools/StandaloneJournalUtil.java diff --git a/.gitmodules b/.gitmodules index 616dacf610..8a45ad8f54 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,4 +1,4 @@ [submodule "modules/accord"] path = modules/accord url = https://github.com/apache/cassandra-accord.git - branch = trunk + branch = trunk \ No newline at end of file diff --git a/src/java/org/apache/cassandra/config/AccordSpec.java b/src/java/org/apache/cassandra/config/AccordSpec.java index 22dafaac3d..817532e23d 100644 --- a/src/java/org/apache/cassandra/config/AccordSpec.java +++ b/src/java/org/apache/cassandra/config/AccordSpec.java @@ -196,10 +196,11 @@ public class AccordSpec private volatile long flushCombinedBlockPeriod = Long.MIN_VALUE; public Version version = Version.DOWNGRADE_SAFE_VERSION; - public void setFlushPeriod(DurationSpec newFlushPeriod) + public JournalSpec setFlushPeriod(DurationSpec newFlushPeriod) { flushPeriod = newFlushPeriod; flushCombinedBlockPeriod = Long.MIN_VALUE; + return this; } public void setPeriodicFlushLagBlock(DurationSpec newPeriodicFlushLagBlock) diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 625fe2be10..28a74a39b9 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -2234,7 +2234,7 @@ public class DatabaseDescriptor } catch (FSWriteError e) { - throw new IllegalStateException(e.getCause().getMessage() + "; unable to start server"); + throw new IllegalStateException(e.getCause().getMessage() + "; unable to start server", e); } } diff --git a/src/java/org/apache/cassandra/journal/Descriptor.java b/src/java/org/apache/cassandra/journal/Descriptor.java index cea68c353e..bac6c7029f 100644 --- a/src/java/org/apache/cassandra/journal/Descriptor.java +++ b/src/java/org/apache/cassandra/journal/Descriptor.java @@ -109,7 +109,7 @@ public final class Descriptor implements Comparable return new Descriptor(directory, timestamp, generation, journalVersion, userVersion); } - static Descriptor fromFile(File file) + public static Descriptor fromFile(File file) { return fromName(file.parent(), file.name()); } diff --git a/src/java/org/apache/cassandra/journal/DumpUtil.java b/src/java/org/apache/cassandra/journal/DumpUtil.java new file mode 100644 index 0000000000..d98e758384 --- /dev/null +++ b/src/java/org/apache/cassandra/journal/DumpUtil.java @@ -0,0 +1,42 @@ +/* + * 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.journal; + +import java.util.function.Consumer; + +/** + * Helper file to avoid exposing components outside their package-local visibility scope + */ +public class DumpUtil +{ + public static void dumpMetadata(Descriptor descriptor, Consumer out) + { + if (Component.METADATA.existsFor(descriptor)) + { + out.accept(Metadata.load(descriptor).toString()); + } + else + out.accept("Metadata absent for " + descriptor); + } + + public static StaticSegment open(Descriptor descriptor, KeySupport keySupport) + { + return StaticSegment.open(descriptor, keySupport); + } +} diff --git a/src/java/org/apache/cassandra/journal/Metadata.java b/src/java/org/apache/cassandra/journal/Metadata.java index d921d02537..13198fa77a 100644 --- a/src/java/org/apache/cassandra/journal/Metadata.java +++ b/src/java/org/apache/cassandra/journal/Metadata.java @@ -154,4 +154,13 @@ final class Metadata metadata.persist(descriptor); return metadata; } + + @Override + public String toString() + { + return "Metadata{" + + "fsyncLimit=" + fsyncLimit + + ", recordsCount=" + recordsCount + + '}'; + } } diff --git a/src/java/org/apache/cassandra/journal/StaticSegment.java b/src/java/org/apache/cassandra/journal/StaticSegment.java index 7250659f85..658224e056 100644 --- a/src/java/org/apache/cassandra/journal/StaticSegment.java +++ b/src/java/org/apache/cassandra/journal/StaticSegment.java @@ -289,7 +289,7 @@ public final class StaticSegment extends Segment /** * Iterate over and invoke the supplied callback on every record. */ - void forEachRecord(RecordConsumer consumer) + public void forEachRecord(RecordConsumer consumer) { try (SequentialReader reader = sequentialReader(descriptor, keySupport, fsyncLimit)) { diff --git a/src/java/org/apache/cassandra/net/InboundMessageHandler.java b/src/java/org/apache/cassandra/net/InboundMessageHandler.java index 6c697d4848..16a900cf72 100644 --- a/src/java/org/apache/cassandra/net/InboundMessageHandler.java +++ b/src/java/org/apache/cassandra/net/InboundMessageHandler.java @@ -23,6 +23,7 @@ import java.nio.ByteBuffer; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; +import org.apache.cassandra.utils.ByteBufferUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -167,7 +168,14 @@ public class InboundMessageHandler extends AbstractMessageHandler { Message m = serializer.deserialize(in, header, version); if (in.available() > 0) // bytes remaining after deser: deserializer is busted - throw new InvalidSerializedSizeException(header.verb, size, size - in.available()); + { + Throwable t = new InvalidSerializedSizeException(header.verb, size, size - in.available()); + ByteBuffer clone = bytes.get(); + clone.limit(clone.position() + size); // cap to expected message size + logger.error("Could not deserialize the message: {}", ByteBufferUtil.bytesToHex(clone), t); + throw t; + + } message = m; } catch (IncompatibleSchemaException e) diff --git a/src/java/org/apache/cassandra/schema/SchemaConstants.java b/src/java/org/apache/cassandra/schema/SchemaConstants.java index 03e911d675..efccc997b3 100644 --- a/src/java/org/apache/cassandra/schema/SchemaConstants.java +++ b/src/java/org/apache/cassandra/schema/SchemaConstants.java @@ -92,6 +92,14 @@ public final class SchemaConstants emptyVersion = UUID.nameUUIDFromBytes(Digest.forSchema().digest()); } + /** + * @return whether the table is an Accord Journal tabe + */ + public static boolean isAccordJournal(String keyspaceName, String tableName) + { + return keyspaceName.equals(SchemaConstants.ACCORD_KEYSPACE_NAME) && tableName.equals(AccordKeyspace.JOURNAL); + } + /** * @return whether or not the keyspace is a really system one (w/ LocalStrategy, unmodifiable, hardcoded) */ diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index aa1d8425c0..5ed6983dec 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -415,7 +415,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier return loadDiffs(commandStoreId, txnId, Load.ALL); } - private BUILDER readAll(JournalKey key) + public BUILDER readAll(JournalKey key) { BUILDER builder = (BUILDER) key.type.serializer.mergerFor(); // TODO (expected): this can be further improved to avoid allocating lambdas @@ -425,6 +425,11 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier return builder; } + public void forEachEntry(JournalKey key, AccordJournalTable.Reader reader) + { + journalTable.readAll(key, reader); + } + private RecordPointer appendInternal(JournalKey key, T write) { AccordJournalValueSerializers.FlyweightSerializer serializer = (AccordJournalValueSerializers.FlyweightSerializer) key.type.serializer; diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java index 58b238d31f..c7b3a77c4b 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java @@ -55,6 +55,13 @@ public class AccordJournalValueSerializers void reserialize(JournalKey key, IMAGE from, DataOutputPlus out, Version userVersion) throws IOException; void deserialize(JournalKey key, IMAGE into, DataInputPlus in, Version userVersion) throws IOException; + + default IMAGE deserialize(JournalKey key, DataInputPlus in, Version userVersion) throws IOException + { + IMAGE image = mergerFor(); + deserialize(key, image, in, userVersion); + return image; + } } public static class CommandDiffSerializer @@ -142,6 +149,14 @@ public class AccordJournalValueSerializers hasRead = true; return newValue; } + + @Override + public String toString() + { + return "IdentityAccumulator{" + + initial + + '}'; + } } public static class RedundantBeforeSerializer diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java b/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java index 3272e5772b..23ba4385e4 100644 --- a/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java +++ b/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java @@ -34,6 +34,14 @@ import java.util.function.Supplier; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Sets; +import org.apache.cassandra.schema.*; +import org.apache.cassandra.service.accord.AccordFastPath; +import org.apache.cassandra.service.accord.AccordStaleReplicas; +import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState; +import org.apache.cassandra.tcm.membership.Directory; +import org.apache.cassandra.tcm.ownership.DataPlacements; +import org.apache.cassandra.tcm.ownership.TokenMap; +import org.apache.cassandra.tcm.sequences.LockedRanges; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -51,8 +59,6 @@ import org.apache.cassandra.metrics.TCMMetrics; import org.apache.cassandra.net.IVerbHandler; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessageDelivery; -import org.apache.cassandra.schema.DistributedSchema; -import org.apache.cassandra.schema.ReplicationParams; import org.apache.cassandra.tcm.listeners.SchemaListener; import org.apache.cassandra.tcm.log.Entry; import org.apache.cassandra.tcm.log.LocalLog; @@ -288,6 +294,49 @@ public class ClusterMetadataService ClusterMetadataService.setInstance(cms); } + public static void empty(Keyspaces keyspaces) + { + if (instance != null) + return; + String localDC = DatabaseDescriptor.getLocalDataCenter(); + ClusterMetadata empty = new ClusterMetadata(Epoch.EMPTY, + DatabaseDescriptor.getPartitioner(), + new DistributedSchema(keyspaces), + Directory.EMPTY, + new TokenMap(DatabaseDescriptor.getPartitioner()), + DataPlacements.empty(), + AccordFastPath.EMPTY, + LockedRanges.EMPTY, + InProgressSequences.EMPTY, + ConsensusMigrationState.EMPTY, + Collections.emptyMap(), + AccordStaleReplicas.EMPTY); + + + LocalLog.LogSpec logSpec = LocalLog.logSpec() + .withInitialState(empty) + .loadSSTables(false) + .withDefaultListeners(false) + .sync() + .withStorage(new AtomicLongBackedProcessor.InMemoryStorage()); + LocalLog log = logSpec.createLog(); + ClusterMetadataService cms = new ClusterMetadataService(new UniformRangePlacement(), + MetadataSnapshots.NO_OP, + log, + new AtomicLongBackedProcessor(log), + new LogState.ReplicationHandler(log), + new LogState.LogNotifyHandler(log), + new CurrentEpochRequestHandler(), + null, + null, + null); + + log.readyUnchecked(); + log.bootstrap(FBUtilities.getBroadcastAddressAndPort(), localDC); + ClusterMetadataService.setInstance(cms); + } + + @SuppressWarnings("resource") public static void initializeForClients() { diff --git a/src/java/org/apache/cassandra/tools/StandaloneJournalUtil.java b/src/java/org/apache/cassandra/tools/StandaloneJournalUtil.java new file mode 100644 index 0000000000..871031acee --- /dev/null +++ b/src/java/org/apache/cassandra/tools/StandaloneJournalUtil.java @@ -0,0 +1,353 @@ +/* + * 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.tools; + +import accord.local.RedundantBefore; +import accord.primitives.Timestamp; +import accord.primitives.TxnId; +import io.airlift.airline.Cli; +import io.airlift.airline.Command; +import io.airlift.airline.Option; +import org.apache.cassandra.config.AccordSpec; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.config.DurationSpec; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.io.util.DataInputBuffer; +import org.apache.cassandra.io.util.File; +import org.apache.cassandra.journal.Descriptor; +import org.apache.cassandra.journal.DumpUtil; +import org.apache.cassandra.journal.StaticSegment; +import org.apache.cassandra.schema.Keyspaces; +import org.apache.cassandra.schema.Schema; +import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.schema.Tables; +import org.apache.cassandra.service.accord.AccordJournal; +import org.apache.cassandra.service.accord.AccordKeyspace; +import org.apache.cassandra.service.accord.JournalKey; +import org.apache.cassandra.service.accord.serializers.Version; +import org.apache.cassandra.tcm.ClusterMetadataService; + +import java.io.IOException; +import java.nio.file.FileSystems; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.PathMatcher; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; + +import static accord.api.Journal.Load.ALL; +import static com.google.common.base.Throwables.getStackTraceAsString; +import static org.apache.cassandra.config.DatabaseDescriptor.setAccordJournalDirectory; + +/** + * Standalone Accord Journal Util. Useful for inspecting journals on remote nodes in cases of failures + * As a convenience, you can build the dtest-jar and scp it to the server, and run the tool: + + java --add-exports java.base/jdk.internal.misc=ALL-UNNAMED \ + --add-exports java.base/jdk.internal.ref=ALL-UNNAMED \ + --add-exports java.base/sun.nio.ch=ALL-UNNAMED \ + --add-exports java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED \ + --add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED \ + --add-exports java.rmi/sun.rmi.server=ALL-UNNAMED \ + --add-exports java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED \ + --add-exports java.sql/java.sql=ALL-UNNAMED \ + \ + --add-opens java.base/java.lang.module=ALL-UNNAMED \ + --add-opens java.base/java.net=ALL-UNNAMED \ + --add-opens java.base/jdk.internal.loader=ALL-UNNAMED \ + --add-opens java.base/jdk.internal.ref=ALL-UNNAMED \ + --add-opens java.base/jdk.internal.reflect=ALL-UNNAMED \ + --add-opens java.base/jdk.internal.math=ALL-UNNAMED \ + --add-opens java.base/jdk.internal.module=ALL-UNNAMED \ + --add-opens java.base/jdk.internal.util.jar=ALL-UNNAMED \ + --add-opens jdk.management/com.sun.management.internal=ALL-UNNAMED \ + --add-opens jdk.management/com.sun.beans.introspect=ALL-UNNAMED \ + --add-opens jdk.management.jfr/jdk.management.jfr=ALL-UNNAMED \ + --add-opens java.desktop/com.sun.beans.introspect=ALL-UNNAMED \ + -cp ./dtest-5.1.jar \ + org.apache.cassandra.tools.StandaloneJournalUtil \ + dump_journal --construct --skip-errors --sstables /tmp/journals/journal-c3412dbac2923051b7914f7b0b79a166/ --journal-segments /tmp/journals/accord_journal/ + + * +**/ +public class StandaloneJournalUtil +{ + private static final Output output = Output.CONSOLE; + + public static void main(String... args) + { + Util.initDatabaseDescriptor(); + DatabaseDescriptor.setPartitioner("org.apache.cassandra.dht.Murmur3Partitioner"); + AccordKeyspace.TABLES = Tables.of(AccordKeyspace.journalMetadata("journal", false)); + ClusterMetadataService.empty(Keyspaces.of(AccordKeyspace.metadata())); + Cli.CliBuilder builder = Cli.builder("util"); + + builder.withDescription("Dump journal").withCommand(DumpSegments.class).withCommand(DumpJournal.class); + + Cli parser = builder.build(); + + int status = 0; + try + { + Runnable parse = parser.parse(args); + parse.run(); + } + catch (Throwable throwable) + { + err(throwable); + status = 2; + } + + System.exit(status); + } + + protected static void err(Throwable e) + { + output.err.println("error: " + e.getMessage()); + output.err.println("-- StackTrace --"); + output.err.println(getStackTraceAsString(e)); + } + + @Command(name = "dump_segments", description = "Dump journal segments") + public static class DumpSegments implements Runnable + { + @Option(name = {"-d", "--dir"}, description = "Directory to find journal segments") + public String dir; + + @Option(name = {"-p", "--pattern"}, description = "Kind to filter by") + public String pattern; + + @Option(name = {"-k", "--kind"}, description = "Kind to filter by") + public String kind; + + @Option(name = {"-t", "--txnid"}, description = "Transaction id to filter by") + public String txnId; + + @Option(name = {"-m", "--metadata-only"}, description = "Only dump metadata file contents") + public boolean metadataOnly; + + public void run() + { + PathMatcher matcher = FileSystems.getDefault().getPathMatcher("glob:" + pattern); + try + { + Files.list(Path.of(dir)).filter(Files::isRegularFile).filter(p -> matcher.matches(p.getFileName())).sorted().forEach(path -> { + Descriptor descriptor = Descriptor.fromFile(new File(path)); + DumpUtil.dumpMetadata(descriptor, output.out::println); + if (metadataOnly) + return; + + StaticSegment segment = DumpUtil.open(descriptor, JournalKey.SUPPORT); + segment.forEachRecord((segment1, position, key, buffer, userVersion) -> { + if (kind != null && key.type != JournalKey.Type.valueOf(kind)) + return; + + if (txnId != null && !TxnId.fromString(txnId).equals(key.id)) + return; + + try (DataInputBuffer in = new DataInputBuffer(buffer, false)) + { + Object v = key.type.serializer.deserialize(key, in, Version.V1); + output.out.printf("%s: %s%n", key, v); + } + catch (Throwable t) + { + t.printStackTrace(output.err); + } + }); + }); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + + } + } + + @Command(name = "dump_journal", description = "Dump journal") + public static class DumpJournal implements Runnable + { + @Option(name = {"-s", "--sstables"}, description = "Path to sstables") + public String sstables; + + @Option(name = {"-j", "--journal-segments"}, description = "Path to journal segments") + public String journalSegments; + + @Option(name = {"-k", "--kind"}, description = "Kind to filter by") + public String kind; + + @Option(name = {"-t", "--txnid"}, description = "Transaction id to filter by") + public String txnId; + + @Option(name = {"--since"}, description = "Filter transactions since this timestamp (inclusive)") + public String since; + + @Option(name = {"--until"}, description = "Filter transactions until this timestamp (inclusive)") + public String until; + + @Option(name = {"-e", "--skip-errors"}, description = "Skip errors: 'true' to skip all, or comma-separated exception class names to skip specific types") + public String skipErrors; + + @Option(name = {"-c", "--construct"}, description = "Construct entry") + public boolean construct; + + + public void run() + { + if (sstables == null && journalSegments == null) + throw new IllegalArgumentException("Either --sstables or --journal-segments must be provided"); + + Timestamp txnId = this.txnId != null ? TxnId.fromString(this.txnId) : null; + Timestamp sinceTimestamp = this.since != null ? TxnId.fromString(this.since) : null; + Timestamp untilTimestamp = this.until != null ? TxnId.fromString(this.until) : null; + + boolean skipAllErrors; + Set skipExceptionTypes = new HashSet<>(); + + if (skipErrors != null && !skipErrors.trim().isEmpty()) + { + String trimmed = skipErrors.trim(); + if ("true".equalsIgnoreCase(trimmed)) + { + skipAllErrors = true; + } + else + { + skipAllErrors = false; + String[] types = trimmed.split(","); + for (String type : types) + { + skipExceptionTypes.add(type.trim()); + } + } + } + else + { + skipAllErrors = false; + } + + if (journalSegments == null) + { + try + { + journalSegments = Files.createTempDirectory("dump_journal").getFileName().toString(); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + setAccordJournalDirectory(journalSegments); + Keyspace.setInitialized(); + AccordJournal journal = new AccordJournal(new AccordSpec.JournalSpec().setFlushPeriod(new DurationSpec.IntMillisecondsBound("1500ms")), new File(journalSegments).parent(), Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME).getColumnFamilyStore(AccordKeyspace.JOURNAL)); + + Keyspace ks = Schema.instance.getKeyspaceInstance("system_accord"); + ColumnFamilyStore cfs = ks.getColumnFamilyStore("journal"); + if (sstables != null) + cfs.importNewSSTables(Collections.singleton(sstables), false, false, false, false, false, false, true); + + Map cache = new HashMap<>(); + journal.start(null); + journal.forEach(key -> processKey(cache, journal, key, txnId, sinceTimestamp, untilTimestamp, skipAllErrors, skipExceptionTypes)); + } + + private void processKey(Map redundantBeforeCache, AccordJournal journal, JournalKey key, Timestamp txnId, Timestamp minTimestamp, Timestamp maxTimestamp, boolean skipAllErrors, Set skipExceptionTypes) + { + if (kind != null && key.type != JournalKey.Type.valueOf(kind)) + return; + + if (txnId != null && !txnId.equals(key.id)) + return; + + // Apply since filtering (inclusive) + if ((minTimestamp != null && key.id.compareTo(minTimestamp) < 0) || + (maxTimestamp != null && key.id.compareTo(maxTimestamp) > 0)) + return; + + try + { + switch (key.type) + { + case COMMAND_DIFF: + { + AtomicInteger counter = new AtomicInteger(); + output.out.println(key + " " + key.id.toStandardString()); + + output.out.println("Individual entries:"); + journal.forEachEntry(key, (in, userVersion) -> { + AccordJournal.Builder builder = new AccordJournal.Builder(key.id, ALL); + builder.deserializeNext(in, userVersion); + output.out.println(String.format("\t%s", builder.toString("\n\t\t"))); + counter.getAndIncrement(); + }); + + if (construct) + { + AccordJournal.Builder builder = new AccordJournal.Builder(key.id, ALL); + journal.forEachEntry(key, builder::deserializeNext); + output.out.println("Reconstructed\n\t\t" + builder.construct(redundantBeforeCache.computeIfAbsent(key.commandStoreId, k -> journal.loadRedundantBefore(key.commandStoreId)))); + } + + break; + } + case REDUNDANT_BEFORE: + case DURABLE_BEFORE: + case SAFE_TO_READ: + case BOOTSTRAP_BEGAN_AT: + case RANGES_FOR_EPOCH: + case TOPOLOGY_UPDATE: + { + Object image = journal.readAll(key); + output.out.println(image); + } + } + } + catch (Throwable t) + { + boolean shouldSkip = false; + + if (skipAllErrors) + { + shouldSkip = true; + } + else if (!skipExceptionTypes.isEmpty()) + { + String exceptionClassName = t.getClass().getSimpleName(); + String fullExceptionClassName = t.getClass().getName(); + + shouldSkip = skipExceptionTypes.contains(exceptionClassName) || + skipExceptionTypes.contains(fullExceptionClassName); + } + + if (!shouldSkip) + throw t; + + output.out.println(String.format("Got error reading key %s", key)); + t.printStackTrace(); + } + } + } +} diff --git a/src/java/org/apache/cassandra/utils/btree/FullBTreeSearchIterator.java b/src/java/org/apache/cassandra/utils/btree/FullBTreeSearchIterator.java index 546fad8fa3..a4cafde0ee 100644 --- a/src/java/org/apache/cassandra/utils/btree/FullBTreeSearchIterator.java +++ b/src/java/org/apache/cassandra/utils/btree/FullBTreeSearchIterator.java @@ -22,8 +22,6 @@ import static org.apache.cassandra.utils.btree.BTree.size; import java.util.Comparator; import java.util.NoSuchElementException; -import accord.utils.Invariants; - public class FullBTreeSearchIterator extends TreeCursor implements BTreeSearchIterator { private final boolean forwards; diff --git a/test/unit/org/apache/cassandra/service/accord/WatermarkCollectorTest.java b/test/unit/org/apache/cassandra/service/accord/WatermarkCollectorTest.java index 7b7ec0f5e5..78e96d2475 100644 --- a/test/unit/org/apache/cassandra/service/accord/WatermarkCollectorTest.java +++ b/test/unit/org/apache/cassandra/service/accord/WatermarkCollectorTest.java @@ -30,7 +30,6 @@ import accord.utils.AccordGens; import accord.utils.Gen; import accord.utils.Gens; import accord.utils.Invariants; -import org.agrona.collections.Int2ObjectHashMap; import org.agrona.collections.Long2LongHashMap; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.dht.IPartitioner; @@ -105,4 +104,4 @@ public class WatermarkCollectorTest return map; }; } -} \ No newline at end of file +}