From d16e8d3653dce8ed767a040c06dbaabc47a9b474 Mon Sep 17 00:00:00 2001 From: Jacek Lewandowski Date: Mon, 18 Sep 2023 12:44:08 +0200 Subject: [PATCH] Do not create sstable files before registering in txn Refactoring prevents the situation where some sstable components, like data or index, are created before the new sstable is registered with lifecycle transaction, which leads to a problem such that there is a short time when incomplete sstable components are present. At the same time, no transaction file is created, which leads to the possibility that the sstable can be recognized as completed by various transaction-aware listers. Patch by Jacek Lewandowski; reviewed by Branimir Lambov, Mike Adamson for CASSANDRA-18737 --- CHANGES.txt | 1 + .../db/compaction/CompactionManager.java | 4 +- .../cassandra/db/compaction/Upgrader.java | 2 +- .../unified/ShardedMultiWriter.java | 2 +- .../writers/CompactionAwareWriter.java | 2 +- .../compress/CompressedSequentialWriter.java | 2 +- .../io/sstable/SimpleSSTableMultiWriter.java | 2 +- .../io/sstable/format/SSTableWriter.java | 97 ++++++--- .../io/sstable/format/SortedTableWriter.java | 76 ++++++- .../io/sstable/format/big/BigTableWriter.java | 129 ++++++----- .../io/sstable/format/bti/BtiTableWriter.java | 204 ++++++++---------- .../cassandra/io/util/SequentialWriter.java | 7 +- .../cassandra/db/SerializationHeaderTest.java | 1 - .../db/lifecycle/RealTransactionsTest.java | 2 +- .../io/sstable/SSTableFlushObserverTest.java | 198 ++++++++++++----- .../io/sstable/SSTableWriterTestBase.java | 2 +- .../cassandra/io/sstable/ScrubTest.java | 1 - .../format/bti/PartitionIndexTest.java | 8 +- 18 files changed, 437 insertions(+), 303 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index fe15bbd4ab..e5f38dbd9a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.0-alpha2 + * Do not create sstable files before registering in txn (CASSANDRA-18737) * Do not log stacktrace on mismatched cache and schema version and checksum error in AutoSavingCache (CASSANDRA-18862) * Remove wrong assertion in SSTableLoader (CASSANDRA-18840) * Fix accessing java.nio.Bits.TOTAL_CAPACITY in Java17 (CASSANDRA-18848) diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index cf6904dda7..44d6f4b833 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -1624,7 +1624,7 @@ public class CompactionManager implements CompactionManagerMBean .setMetadataCollector(new MetadataCollector(cfs.metadata().comparator).sstableLevel(sstable.getSSTableLevel())) .setSerializationHeader(sstable.header) .addDefaultComponents(cfs.indexManager.listIndexGroups()) - .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, cfs.metadata.get()) + .setSecondaryIndexGroups(cfs.indexManager.listIndexGroups()) .build(txn, cfs); } @@ -1664,7 +1664,7 @@ public class CompactionManager implements CompactionManagerMBean .setMetadataCollector(new MetadataCollector(sstables, cfs.metadata().comparator).sstableLevel(minLevel)) .setSerializationHeader(SerializationHeader.make(cfs.metadata(), sstables)) .addDefaultComponents(cfs.indexManager.listIndexGroups()) - .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, cfs.metadata.get()) + .setSecondaryIndexGroups(cfs.indexManager.listIndexGroups()) .build(txn, cfs); } diff --git a/src/java/org/apache/cassandra/db/compaction/Upgrader.java b/src/java/org/apache/cassandra/db/compaction/Upgrader.java index 89e73646bc..9e4c4dd750 100644 --- a/src/java/org/apache/cassandra/db/compaction/Upgrader.java +++ b/src/java/org/apache/cassandra/db/compaction/Upgrader.java @@ -84,7 +84,7 @@ public class Upgrader .setMetadataCollector(sstableMetadataCollector) .setSerializationHeader(SerializationHeader.make(cfs.metadata(), Sets.newHashSet(sstable))) .addDefaultComponents(cfs.indexManager.listIndexGroups()) - .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), transaction, cfs.metadata.get()) + .setSecondaryIndexGroups(cfs.indexManager.listIndexGroups()) .build(transaction, cfs); } diff --git a/src/java/org/apache/cassandra/db/compaction/unified/ShardedMultiWriter.java b/src/java/org/apache/cassandra/db/compaction/unified/ShardedMultiWriter.java index 58ce924c12..a5b5df9e49 100644 --- a/src/java/org/apache/cassandra/db/compaction/unified/ShardedMultiWriter.java +++ b/src/java/org/apache/cassandra/db/compaction/unified/ShardedMultiWriter.java @@ -117,7 +117,7 @@ public class ShardedMultiWriter implements SSTableMultiWriter .setMetadataCollector(metadataCollector) .setSerializationHeader(header) .addDefaultComponents(indexGroups) - .addFlushObserversForSecondaryIndexes(indexGroups, lifecycleNewTracker, cfs.metadata.get()) + .setSecondaryIndexGroups(indexGroups) .build(lifecycleNewTracker, cfs); } diff --git a/src/java/org/apache/cassandra/db/compaction/writers/CompactionAwareWriter.java b/src/java/org/apache/cassandra/db/compaction/writers/CompactionAwareWriter.java index be014ff4b3..64055bba5c 100644 --- a/src/java/org/apache/cassandra/db/compaction/writers/CompactionAwareWriter.java +++ b/src/java/org/apache/cassandra/db/compaction/writers/CompactionAwareWriter.java @@ -319,7 +319,7 @@ public abstract class CompactionAwareWriter extends Transactional.AbstractTransa .setTransientSSTable(isTransient) .setRepairedAt(minRepairedAt) .setPendingRepair(pendingRepair) - .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, cfs.metadata.get()) + .setSecondaryIndexGroups(cfs.indexManager.listIndexGroups()) .addDefaultComponents(cfs.indexManager.listIndexGroups()); } } diff --git a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java index 7a0959490f..4a7f8161cc 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java +++ b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java @@ -202,7 +202,7 @@ public class CompressedSequentialWriter extends SequentialWriter // next chunk should be written right after current + length of the checksum (int) chunkOffset += compressedLength + 4; if (runPostFlush != null) - runPostFlush.run(); + runPostFlush.accept(getLastFlushOffset()); } public CompressionMetadata open(long overrideLength) diff --git a/src/java/org/apache/cassandra/io/sstable/SimpleSSTableMultiWriter.java b/src/java/org/apache/cassandra/io/sstable/SimpleSSTableMultiWriter.java index affc0d86b1..5ace39b501 100644 --- a/src/java/org/apache/cassandra/io/sstable/SimpleSSTableMultiWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SimpleSSTableMultiWriter.java @@ -132,7 +132,7 @@ public class SimpleSSTableMultiWriter implements SSTableMultiWriter .setMetadataCollector(metadataCollector) .setSerializationHeader(header) .addDefaultComponents(indexGroups) - .addFlushObserversForSecondaryIndexes(indexGroups, lifecycleNewTracker, metadata.get()) + .setSecondaryIndexGroups(indexGroups) .build(lifecycleNewTracker, owner); return new SimpleSSTableMultiWriter(writer, lifecycleNewTracker); } diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java index 05140632f9..38efd1955e 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java @@ -18,6 +18,7 @@ package org.apache.cassandra.io.sstable.format; +import java.io.IOError; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -30,6 +31,9 @@ import java.util.function.Supplier; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; +import com.google.common.collect.Sets; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.SerializationHeader; @@ -51,7 +55,6 @@ import org.apache.cassandra.io.sstable.metadata.MetadataComponent; import org.apache.cassandra.io.sstable.metadata.MetadataType; import org.apache.cassandra.io.sstable.metadata.StatsMetadata; import org.apache.cassandra.io.util.MmappedRegionsCache; -import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.utils.Throwables; import org.apache.cassandra.utils.TimeUUID; import org.apache.cassandra.utils.concurrent.Transactional; @@ -65,6 +68,8 @@ import static com.google.common.base.Preconditions.checkNotNull; */ public abstract class SSTableWriter extends SSTable implements Transactional { + private final static Logger logger = LoggerFactory.getLogger(SSTableWriter.class); + protected long repairedAt; protected TimeUUID pendingRepair; protected boolean isTransient; @@ -72,7 +77,7 @@ public abstract class SSTableWriter extends SSTable implements Transactional protected final long keyCount; protected final MetadataCollector metadataCollector; protected final SerializationHeader header; - protected final Collection observers; + protected final List observers; protected final MmappedRegionsCache mmappedRegionsCache; protected final TransactionalProxy txnProxy = txnProxy(); protected final LifecycleNewTracker lifecycleNewTracker; @@ -88,7 +93,7 @@ public abstract class SSTableWriter extends SSTable implements Transactional protected SSTableWriter(Builder builder, LifecycleNewTracker lifecycleNewTracker, SSTable.Owner owner) { super(builder, owner); - checkNotNull(builder.getFlushObservers()); + checkNotNull(builder.getIndexGroups()); checkNotNull(builder.getMetadataCollector()); checkNotNull(builder.getSerializationHeader()); @@ -98,11 +103,60 @@ public abstract class SSTableWriter extends SSTable implements Transactional this.isTransient = builder.isTransientSSTable(); this.metadataCollector = builder.getMetadataCollector(); this.header = builder.getSerializationHeader(); - this.observers = builder.getFlushObservers(); this.mmappedRegionsCache = builder.getMmappedRegionsCache(); this.lifecycleNewTracker = lifecycleNewTracker; + // We need to ensure that no sstable components exist before the lifecycle transaction starts tracking it. + // Otherwise, it means that we either want to overwrite some existing sstable, which is not allowed, or some + // sstable files were created before the sstable is registered in the lifecycle transaction, which may lead + // to a race such that the sstable is listed as completed due to the lack of the transaction file before + // anything is actually written to it. + Set existingComponents = Sets.filter(components, c -> descriptor.fileFor(c).exists()); + assert existingComponents.isEmpty() : String.format("Cannot create a new SSTable in directory %s as component files %s already exist there", + descriptor.directory, + existingComponents); + lifecycleNewTracker.trackNew(this); + + try + { + ArrayList observers = new ArrayList<>(); + this.observers = Collections.unmodifiableList(observers); + for (Index.Group group : builder.getIndexGroups()) + { + SSTableFlushObserver observer = group.getFlushObserver(descriptor, lifecycleNewTracker, metadata.getLocal()); + if (observer != null) + { + observer.begin(); + observers.add(observer); + } + } + } + catch (RuntimeException | IOError ex) + { + handleConstructionFailure(ex); + throw ex; + } + } + + /** + * Constructors of subclasses, if they open any resources, should wrap that in a try-catch block and call this + * method in the 'catch' section after closing any resources opened in the constructor. This method would remove + * the sstable from the transaction and delete the orphaned components, if any were created during the construction. + * The caught exception should be then rethrown so the {@link Builder} can handle it and close any resources opened + * implicitly by the builder. + *

+ * See {@link SortedTableWriter#SortedTableWriter(SortedTableWriter.Builder, LifecycleNewTracker, Owner)} as of CASSANDRA-18737. + * + * @param ex the exception thrown during the construction + */ + protected void handleConstructionFailure(Throwable ex) + { + logger.warn("Failed to open " + descriptor + " for writing", ex); + for (int i = observers.size()-1; i >= 0; i--) + observers.get(i).abort(ex); + descriptor.getFormat().deleteOrphanedComponents(descriptor, components); + lifecycleNewTracker.untrackNew(this); } @Override @@ -382,7 +436,7 @@ public abstract class SSTableWriter extends SSTable implements Transactional private TimeUUID pendingRepair; private boolean transientSSTable; private SerializationHeader serializationHeader; - private Collection flushObservers; + private List indexGroups; public B setMetadataCollector(MetadataCollector metadataCollector) { @@ -420,12 +474,6 @@ public abstract class SSTableWriter extends SSTable implements Transactional return (B) this; } - public B setFlushObservers(Collection flushObservers) - { - this.flushObservers = ImmutableList.copyOf(flushObservers); - return (B) this; - } - public B addDefaultComponents(Collection indexGroups) { checkNotNull(getTableMetadataRef()); @@ -460,26 +508,11 @@ public abstract class SSTableWriter extends SSTable implements Transactional return components; } - public B addFlushObserversForSecondaryIndexes(Collection indexGroups, LifecycleNewTracker tracker, TableMetadata metadata) + public B setSecondaryIndexGroups(Collection indexGroups) { - if (indexGroups == null) - return (B) this; - - Collection current = this.flushObservers != null ? this.flushObservers : Collections.emptyList(); - List observers = new ArrayList<>(indexGroups.size() + current.size()); - observers.addAll(current); - - for (Index.Group group : indexGroups) - { - SSTableFlushObserver observer = group.getFlushObserver(descriptor, tracker, metadata); - if (observer != null) - { - observer.begin(); - observers.add(observer); - } - } - - return setFlushObservers(observers); + checkNotNull(indexGroups); + this.indexGroups = ImmutableList.copyOf(indexGroups); + return (B) this; } public MetadataCollector getMetadataCollector() @@ -512,9 +545,9 @@ public abstract class SSTableWriter extends SSTable implements Transactional return serializationHeader; } - public Collection getFlushObservers() + public List getIndexGroups() { - return flushObservers; + return indexGroups == null ? Collections.emptyList() : indexGroups; } public abstract MmappedRegionsCache getMmappedRegionsCache(); diff --git a/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java index cf3918d05b..a2228a986e 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java @@ -23,7 +23,9 @@ import java.nio.BufferOverflowException; import java.util.Collection; import java.util.Set; import java.util.function.Consumer; +import java.util.function.Supplier; +import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -71,27 +73,48 @@ import static com.google.common.base.Preconditions.checkNotNull; /** * A generic implementation of a writer which assumes the existence of some partition index and bloom filter. */ -public abstract class SortedTableWriter

extends SSTableWriter +public abstract class SortedTableWriter

extends SSTableWriter { private final static Logger logger = LoggerFactory.getLogger(SortedTableWriter.class); // TODO dataWriter is not needed to be directly accessible - we can access everything we need for the dataWriter // from a partition writer protected final SequentialWriter dataWriter; + protected final I indexWriter; protected final P partitionWriter; private final FileHandle.Builder dataFileBuilder = new FileHandle.Builder(descriptor.fileFor(Components.DATA)); private DecoratedKey lastWrittenKey; private DataPosition dataMark; private long lastEarlyOpenLength; - public SortedTableWriter(Builder builder, LifecycleNewTracker lifecycleNewTracker, SSTable.Owner owner) + public SortedTableWriter(Builder builder, LifecycleNewTracker lifecycleNewTracker, SSTable.Owner owner) { super(builder, lifecycleNewTracker, owner); - checkNotNull(builder.getDataWriter()); - checkNotNull(builder.getPartitionWriter()); - this.dataWriter = builder.getDataWriter(); - this.partitionWriter = builder.getPartitionWriter(); + SequentialWriter dataWriter = null; + I indexWriter = null; + P partitionWriter = null; + try + { + dataWriter = builder.openDataWriter(); + checkNotNull(dataWriter); + + indexWriter = builder.openIndexWriter(dataWriter); + checkNotNull(indexWriter); + + partitionWriter = builder.openPartitionWriter(dataWriter, indexWriter); + checkNotNull(partitionWriter); + + this.dataWriter = dataWriter; + this.indexWriter = indexWriter; + this.partitionWriter = partitionWriter; + } + catch (RuntimeException | Error ex) + { + Throwables.closeNonNullAndAddSuppressed(ex, partitionWriter, indexWriter, dataWriter); + handleConstructionFailure(ex); + throw ex; + } } /** @@ -264,6 +287,7 @@ public abstract class SortedTableWriter

ex public void mark() { dataMark = dataWriter.mark(); + indexWriter.mark(); } @Override @@ -271,6 +295,29 @@ public abstract class SortedTableWriter

ex { dataWriter.resetAndTruncate(dataMark); partitionWriter.reset(); + indexWriter.resetAndTruncate(); + } + + @Override + protected SSTableWriter.TransactionalProxy txnProxy() + { + return new TransactionalProxy(() -> FBUtilities.immutableListWithFilteredNulls(indexWriter, dataWriter)); + } + + protected class TransactionalProxy extends SSTableWriter.TransactionalProxy + { + public TransactionalProxy(Supplier> transactionals) + { + super(transactionals); + } + + @Override + protected Throwable doPostCleanup(Throwable accumulate) + { + accumulate = Throwables.close(accumulate, partitionWriter); + accumulate = super.doPostCleanup(accumulate); + return accumulate; + } } @Override @@ -385,7 +432,7 @@ public abstract class SortedTableWriter

ex protected final IFilter bf; - protected AbstractIndexWriter(Builder b) + protected AbstractIndexWriter(Builder b) { this.descriptor = b.descriptor; this.metadata = b.getTableMetadataRef(); @@ -409,6 +456,10 @@ public abstract class SortedTableWriter

ex } } + public abstract void mark(); + + public abstract void resetAndTruncate(); + protected void doPrepare() { flushBf(); @@ -428,8 +479,9 @@ public abstract class SortedTableWriter

ex } public abstract static class Builder

, - B extends Builder> extends SSTableWriter.Builder + I extends AbstractIndexWriter, + W extends SortedTableWriter, + B extends Builder> extends SSTableWriter.Builder { public Builder(Descriptor descriptor) @@ -450,8 +502,10 @@ public abstract class SortedTableWriter

ex return (B) this; } - public abstract SequentialWriter getDataWriter(); + protected abstract SequentialWriter openDataWriter(); - public abstract P getPartitionWriter(); + protected abstract I openIndexWriter(SequentialWriter dataWriter); + + protected abstract P openPartitionWriter(SequentialWriter dataWriter, I indexWriter); } } diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java index 6a837946c1..37c6f6119c 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java @@ -24,15 +24,14 @@ import java.util.HashMap; import java.util.Map; import java.util.function.Consumer; -import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableSet; - import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.DeletionTime; +import org.apache.cassandra.db.compaction.OperationType; import org.apache.cassandra.db.lifecycle.ILifecycleTransaction; import org.apache.cassandra.db.lifecycle.LifecycleNewTracker; import org.apache.cassandra.index.Index; @@ -44,7 +43,6 @@ import org.apache.cassandra.io.sstable.SSTable; import org.apache.cassandra.io.sstable.format.DataComponent; import org.apache.cassandra.io.sstable.format.IndexComponent; import org.apache.cassandra.io.sstable.format.SSTableReader; -import org.apache.cassandra.io.sstable.format.SSTableWriter; import org.apache.cassandra.io.sstable.format.SortedTableWriter; import org.apache.cassandra.io.sstable.format.big.BigFormat.Components; import org.apache.cassandra.io.sstable.indexsummary.IndexSummary; @@ -59,20 +57,19 @@ import org.apache.cassandra.io.util.SequentialWriter; import org.apache.cassandra.service.CacheService; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.EstimatedHistogram; -import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.IFilter; import org.apache.cassandra.utils.JVMStabilityInspector; import org.apache.cassandra.utils.Throwables; import static com.google.common.base.Preconditions.checkNotNull; +import static com.google.common.base.Preconditions.checkState; import static org.apache.cassandra.io.util.FileHandle.Builder.NO_LENGTH_OVERRIDE; import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis; -public class BigTableWriter extends SortedTableWriter +public class BigTableWriter extends SortedTableWriter { private static final Logger logger = LoggerFactory.getLogger(BigTableWriter.class); - private final IndexWriter indexWriter; private final RowIndexEntry.IndexSerializer rowIndexEntrySerializer; private final Map cachedKeys = new HashMap<>(); private final boolean shouldMigrateKeyCache; @@ -80,30 +77,15 @@ public class BigTableWriter extends SortedTableWriter public BigTableWriter(Builder builder, LifecycleNewTracker lifecycleNewTracker, SSTable.Owner owner) { super(builder, lifecycleNewTracker, owner); - checkNotNull(builder.getRowIndexEntrySerializer()); - checkNotNull(builder.getIndexWriter()); this.rowIndexEntrySerializer = builder.getRowIndexEntrySerializer(); - this.indexWriter = builder.getIndexWriter(); + checkNotNull(this.rowIndexEntrySerializer); + this.shouldMigrateKeyCache = DatabaseDescriptor.shouldMigrateKeycacheOnCompaction() && lifecycleNewTracker instanceof ILifecycleTransaction && !((ILifecycleTransaction) lifecycleNewTracker).isOffline(); } - @Override - public void mark() - { - super.mark(); - indexWriter.mark(); - } - - @Override - public void resetAndTruncate() - { - super.resetAndTruncate(); - indexWriter.resetAndTruncate(); - } - @Override protected void onStartPartition(DecoratedKey key) { @@ -145,7 +127,7 @@ public class BigTableWriter extends SortedTableWriter return entry; } - @SuppressWarnings({"resource", "RedundantSuppression"}) + @SuppressWarnings({ "resource", "RedundantSuppression" }) private BigTableReader openInternal(IndexSummaryBuilder.ReadableBoundary boundary, SSTableReader.OpenReason openReason) { assert boundary == null || (boundary.indexLength > 0 && boundary.dataLength > 0); @@ -251,16 +233,10 @@ public class BigTableWriter extends SortedTableWriter return openInternal(null, openReason); } - @Override - protected SSTableWriter.TransactionalProxy txnProxy() - { - return new SSTableWriter.TransactionalProxy(() -> FBUtilities.immutableListWithFilteredNulls(indexWriter, dataWriter)); - } - /** * Encapsulates writing the index and filter for an SSTable. The state of this object is not valid until it has been closed. */ - static class IndexWriter extends SortedTableWriter.AbstractIndexWriter + protected static class IndexWriter extends SortedTableWriter.AbstractIndexWriter { private final RowIndexEntry.IndexSerializer rowIndexEntrySerializer; @@ -271,7 +247,7 @@ public class BigTableWriter extends SortedTableWriter private DecoratedKey first; private DecoratedKey last; - protected IndexWriter(Builder b) + protected IndexWriter(Builder b, SequentialWriter dataWriter) { super(b); this.rowIndexEntrySerializer = b.getRowIndexEntrySerializer(); @@ -279,10 +255,8 @@ public class BigTableWriter extends SortedTableWriter builder = IndexComponent.fileBuilder(Components.PRIMARY_INDEX, b).withMmappedRegionsCache(b.getMmappedRegionsCache()); summary = new IndexSummaryBuilder(b.getKeyCount(), b.getTableMetadataRef().getLocal().params.minIndexInterval, Downsampling.BASE_SAMPLING_LEVEL); // register listeners to be alerted when the data files are flushed - writer.setPostFlushListener(() -> summary.markIndexSynced(writer.getLastFlushOffset())); - @SuppressWarnings({"resource", "RedundantSuppression"}) - SequentialWriter dataWriter = b.getDataWriter(); - dataWriter.setPostFlushListener(() -> summary.markDataSynced(dataWriter.getLastFlushOffset())); + writer.setPostFlushListener(summary::markIndexSynced); + dataWriter.setPostFlushListener(summary::markDataSynced); } // finds the last (-offset) decorated key that can be guaranteed to occur fully in the flushed portion of the index file @@ -316,11 +290,13 @@ public class BigTableWriter extends SortedTableWriter summary.maybeAddEntry(key, indexStart, indexEnd, dataEnd); } + @Override public void mark() { mark = writer.mark(); } + @Override public void resetAndTruncate() { // we can't un-set the bloom filter addition, but extra keys in there are harmless. @@ -372,13 +348,17 @@ public class BigTableWriter extends SortedTableWriter } } - public static class Builder extends SortedTableWriter.Builder + public static class Builder extends SortedTableWriter.Builder { private RowIndexEntry.IndexSerializer rowIndexEntrySerializer; - private IndexWriter indexWriter; - private SequentialWriter dataWriter; - private BigFormatPartitionWriter partitionWriter; private MmappedRegionsCache mmappedRegionsCache; + private OperationType operationType; + + // Writers are expected to be opened only once during construction of the sstable. The following flags are used + // to ensure that. + private boolean indexWriterOpened; + private boolean dataWriterOpened; + private boolean partitionWriterOpened; public Builder(Descriptor descriptor) { @@ -405,30 +385,51 @@ public class BigTableWriter extends SortedTableWriter } @Override - public SequentialWriter getDataWriter() + protected SequentialWriter openDataWriter() { - return ensuringInBuildInternalContext(dataWriter); + checkState(!dataWriterOpened, "Data writer has been already opened."); + + SequentialWriter dataWriter = DataComponent.buildWriter(descriptor, + getTableMetadataRef().getLocal(), + getIOOptions().writerOptions, + getMetadataCollector(), + ensuringInBuildInternalContext(operationType), + getIOOptions().flushCompression); + this.dataWriterOpened = true; + return dataWriter; } @Override - public BigFormatPartitionWriter getPartitionWriter() + protected IndexWriter openIndexWriter(SequentialWriter dataWriter) { - return ensuringInBuildInternalContext(partitionWriter); + checkNotNull(dataWriter); + checkState(!indexWriterOpened, "Index writer has been already opened."); + + IndexWriter indexWriter = new IndexWriter(this, dataWriter); + this.indexWriterOpened = true; + return indexWriter; } - public RowIndexEntry.IndexSerializer getRowIndexEntrySerializer() + @Override + protected BigFormatPartitionWriter openPartitionWriter(SequentialWriter dataWriter, IndexWriter indexWriter) + { + checkNotNull(dataWriter); + checkNotNull(indexWriter); + checkState(!partitionWriterOpened, "Partition writer has been already opened."); + + BigFormatPartitionWriter partitionWriter = new BigFormatPartitionWriter(getSerializationHeader(), dataWriter, descriptor.version, getRowIndexEntrySerializer().indexInfoSerializer()); + this.partitionWriterOpened = true; + return partitionWriter; + } + + RowIndexEntry.IndexSerializer getRowIndexEntrySerializer() { return ensuringInBuildInternalContext(rowIndexEntrySerializer); } - public IndexWriter getIndexWriter() - { - return ensuringInBuildInternalContext(indexWriter); - } - private T ensuringInBuildInternalContext(T value) { - Preconditions.checkState(value != null, "This getter can be used only during the lifetime of the sstable constructor. Do not use it directly."); + checkState(value != null, "The requested resource has not been initialized yet."); return value; } @@ -437,30 +438,24 @@ public class BigTableWriter extends SortedTableWriter { try { - mmappedRegionsCache = new MmappedRegionsCache(); - rowIndexEntrySerializer = new RowIndexEntry.Serializer(descriptor.version, getSerializationHeader(), owner != null ? owner.getMetrics() : null); - dataWriter = DataComponent.buildWriter(descriptor, - getTableMetadataRef().getLocal(), - getIOOptions().writerOptions, - getMetadataCollector(), - lifecycleNewTracker.opType(), - getIOOptions().flushCompression); - indexWriter = new IndexWriter(this); - partitionWriter = new BigFormatPartitionWriter(getSerializationHeader(), dataWriter, descriptor.version, rowIndexEntrySerializer.indexInfoSerializer()); + this.operationType = lifecycleNewTracker.opType(); + this.mmappedRegionsCache = new MmappedRegionsCache(); + this.rowIndexEntrySerializer = new RowIndexEntry.Serializer(descriptor.version, getSerializationHeader(), owner != null ? owner.getMetrics() : null); return new BigTableWriter(this, lifecycleNewTracker, owner); } catch (RuntimeException | Error ex) { - Throwables.closeAndAddSuppressed(ex, partitionWriter, indexWriter, dataWriter, mmappedRegionsCache); + Throwables.closeNonNullAndAddSuppressed(ex, mmappedRegionsCache); throw ex; } finally { - rowIndexEntrySerializer = null; - indexWriter = null; - dataWriter = null; - partitionWriter = null; - mmappedRegionsCache = null; + this.rowIndexEntrySerializer = null; + this.mmappedRegionsCache = null; + this.operationType = null; + this.partitionWriterOpened = false; + this.indexWriterOpened = false; + this.dataWriterOpened = false; } } } diff --git a/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableWriter.java index 0e462125a9..ebe17d6c0f 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableWriter.java @@ -23,14 +23,13 @@ import java.util.function.Consumer; import java.util.function.Supplier; import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Preconditions; -import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.DeletionTime; +import org.apache.cassandra.db.compaction.OperationType; import org.apache.cassandra.db.lifecycle.LifecycleNewTracker; import org.apache.cassandra.index.Index; import org.apache.cassandra.io.FSReadError; @@ -50,44 +49,25 @@ import org.apache.cassandra.io.util.MmappedRegionsCache; import org.apache.cassandra.io.util.SequentialWriter; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.Clock; -import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.IFilter; import org.apache.cassandra.utils.JVMStabilityInspector; import org.apache.cassandra.utils.Throwables; -import org.apache.cassandra.utils.concurrent.Transactional; +import static com.google.common.base.Preconditions.checkNotNull; +import static com.google.common.base.Preconditions.checkState; import static org.apache.cassandra.io.util.FileHandle.Builder.NO_LENGTH_OVERRIDE; /** * Writes SSTables in BTI format (see {@link BtiFormat}), which can be read by {@link BtiTableReader}. */ @VisibleForTesting -public class BtiTableWriter extends SortedTableWriter +public class BtiTableWriter extends SortedTableWriter { private static final Logger logger = LoggerFactory.getLogger(BtiTableWriter.class); - private final BtiFormatPartitionWriter partitionWriter; - private final IndexWriter iwriter; - public BtiTableWriter(Builder builder, LifecycleNewTracker lifecycleNewTracker, SSTable.Owner owner) { super(builder, lifecycleNewTracker, owner); - this.iwriter = builder.getIndexWriter(); - this.partitionWriter = builder.getPartitionWriter(); - } - - @Override - public void mark() - { - super.mark(); - iwriter.mark(); - } - - @Override - public void resetAndTruncate() - { - super.resetAndTruncate(); - iwriter.resetAndTruncate(); } @Override @@ -97,11 +77,11 @@ public class BtiTableWriter extends SortedTableWriter finishResult, partitionLevelDeletion, partitionWriter.getRowIndexBlockCount()); - iwriter.append(key, entry); + indexWriter.append(key, entry); return entry; } - @SuppressWarnings({"resource", "RedundantSuppression"}) + @SuppressWarnings({ "resource", "RedundantSuppression" }) private BtiTableReader openInternal(OpenReason openReason, boolean isFinal, Supplier partitionIndexSupplier) { IFilter filter = null; @@ -118,9 +98,9 @@ public class BtiTableWriter extends SortedTableWriter builder.setStatsMetadata(statsMetadata()); partitionIndex = partitionIndexSupplier.get(); - rowIndexFile = iwriter.rowIndexFHBuilder.complete(); + rowIndexFile = indexWriter.rowIndexFHBuilder.complete(); dataFile = openDataFile(isFinal ? NO_LENGTH_OVERRIDE : dataWriter.getLastFlushOffset(), builder.getStatsMetadata()); - filter = iwriter.getFilterCopy(); + filter = indexWriter.getFilterCopy(); return builder.setPartitionIndex(partitionIndex) .setFirst(partitionIndex.firstKey()) @@ -142,9 +122,9 @@ public class BtiTableWriter extends SortedTableWriter public void openEarly(Consumer callWhenReady) { long dataLength = dataWriter.position(); - iwriter.buildPartial(dataLength, partitionIndex -> + indexWriter.buildPartial(dataLength, partitionIndex -> { - iwriter.rowIndexFHBuilder.withLengthOverride(iwriter.rowIndexWriter.getLastFlushOffset()); + indexWriter.rowIndexFHBuilder.withLengthOverride(indexWriter.rowIndexWriter.getLastFlushOffset()); BtiTableReader reader = openInternal(OpenReason.EARLY, false, () -> partitionIndex); callWhenReady.accept(reader); }); @@ -154,10 +134,10 @@ public class BtiTableWriter extends SortedTableWriter public SSTableReader openFinalEarly() { // we must ensure the data is completely flushed to disk - iwriter.complete(); // This will be called by completedPartitionIndex() below too, but we want it done now to + indexWriter.complete(); // This will be called by completedPartitionIndex() below too, but we want it done now to // ensure outstanding openEarly actions are not triggered. dataWriter.sync(); - iwriter.rowIndexWriter.sync(); + indexWriter.rowIndexWriter.sync(); // Note: Nothing must be written to any of the files after this point, as the chunk cache could pick up and // retain a partially-written page. @@ -165,42 +145,20 @@ public class BtiTableWriter extends SortedTableWriter } @Override - @SuppressWarnings({"resource", "RedundantSuppression"}) + @SuppressWarnings({ "resource", "RedundantSuppression" }) protected SSTableReader openFinal(OpenReason openReason) { if (maxDataAge < 0) maxDataAge = Clock.Global.currentTimeMillis(); - return openInternal(openReason, true, iwriter::completedPartitionIndex); - } - - @Override - protected TransactionalProxy txnProxy() - { - return new TransactionalProxy(() -> FBUtilities.immutableListWithFilteredNulls(iwriter, dataWriter)); - } - - private class TransactionalProxy extends SortedTableWriter.TransactionalProxy - { - public TransactionalProxy(Supplier> transactionals) - { - super(transactionals); - } - - @Override - protected Throwable doPostCleanup(Throwable accumulate) - { - accumulate = Throwables.close(accumulate, partitionWriter); - accumulate = super.doPostCleanup(accumulate); - return accumulate; - } + return openInternal(openReason, true, indexWriter::completedPartitionIndex); } /** * Encapsulates writing the index and filter for an SSTable. The state of this object is not valid until it has been closed. */ - static class IndexWriter extends SortedTableWriter.AbstractIndexWriter + protected static class IndexWriter extends SortedTableWriter.AbstractIndexWriter { final SequentialWriter rowIndexWriter; private final FileHandle.Builder rowIndexFHBuilder; @@ -211,7 +169,7 @@ public class BtiTableWriter extends SortedTableWriter private DataPosition riMark; private DataPosition piMark; - IndexWriter(Builder b) + IndexWriter(Builder b, SequentialWriter dataWriter) { super(b); rowIndexWriter = new SequentialWriter(descriptor.fileFor(Components.ROW_INDEX), b.getIOOptions().writerOptions); @@ -220,11 +178,9 @@ public class BtiTableWriter extends SortedTableWriter partitionIndexFHBuilder = IndexComponent.fileBuilder(Components.PARTITION_INDEX, b).withMmappedRegionsCache(b.getMmappedRegionsCache()); partitionIndex = new PartitionIndexBuilder(partitionIndexWriter, partitionIndexFHBuilder); // register listeners to be alerted when the data files are flushed - partitionIndexWriter.setPostFlushListener(() -> partitionIndex.markPartitionIndexSynced(partitionIndexWriter.getLastFlushOffset())); - rowIndexWriter.setPostFlushListener(() -> partitionIndex.markRowIndexSynced(rowIndexWriter.getLastFlushOffset())); - @SuppressWarnings({"resource", "RedundantSuppression"}) - SequentialWriter dataWriter = b.getDataWriter(); - dataWriter.setPostFlushListener(() -> partitionIndex.markDataSynced(dataWriter.getLastFlushOffset())); + partitionIndexWriter.setPostFlushListener(partitionIndex::markPartitionIndexSynced); + rowIndexWriter.setPostFlushListener(partitionIndex::markRowIndexSynced); + dataWriter.setPostFlushListener(partitionIndex::markDataSynced); } public long append(DecoratedKey key, AbstractRowIndexEntry indexEntry) throws IOException @@ -336,50 +292,20 @@ public class BtiTableWriter extends SortedTableWriter } } - public static class Builder extends SortedTableWriter.Builder + public static class Builder extends SortedTableWriter.Builder { - private SequentialWriter dataWriter; - private BtiFormatPartitionWriter partitionWriter; - private IndexWriter indexWriter; private MmappedRegionsCache mmappedRegionsCache; + private OperationType operationType; + + private boolean dataWriterOpened; + private boolean partitionWriterOpened; + private boolean indexWriterOpened; public Builder(Descriptor descriptor) { super(descriptor); } - // The following getters for the resources opened by buildInternal method can be only used during the lifetime of - // that method - that is, during the construction of the sstable. - - @Override - public MmappedRegionsCache getMmappedRegionsCache() - { - return ensuringInBuildInternalContext(mmappedRegionsCache); - } - - @Override - public SequentialWriter getDataWriter() - { - return ensuringInBuildInternalContext(dataWriter); - } - - @Override - public BtiFormatPartitionWriter getPartitionWriter() - { - return ensuringInBuildInternalContext(partitionWriter); - } - - public IndexWriter getIndexWriter() - { - return ensuringInBuildInternalContext(indexWriter); - } - - private T ensuringInBuildInternalContext(T value) - { - Preconditions.checkState(value != null, "This getter can be used only during the lifetime of the sstable constructor. Do not use it directly."); - return value; - } - @Override public Builder addDefaultComponents(Collection indexGroups) { @@ -390,40 +316,82 @@ public class BtiTableWriter extends SortedTableWriter return this; } + // The following getters for the resources opened by buildInternal method can be only used during the lifetime of + // that method - that is, during the construction of the sstable. + + @Override + public MmappedRegionsCache getMmappedRegionsCache() + { + return ensuringInBuildInternalContext(mmappedRegionsCache); + } + + @Override + protected SequentialWriter openDataWriter() + { + checkState(!dataWriterOpened, "Data writer has been already opened."); + + return DataComponent.buildWriter(descriptor, + getTableMetadataRef().getLocal(), + getIOOptions().writerOptions, + getMetadataCollector(), + ensuringInBuildInternalContext(operationType), + getIOOptions().flushCompression); + } + + @Override + protected IndexWriter openIndexWriter(SequentialWriter dataWriter) + { + checkNotNull(dataWriter); + checkState(!indexWriterOpened, "Index writer has been already opened."); + + IndexWriter indexWriter = new IndexWriter(this, dataWriter); + indexWriterOpened = true; + return indexWriter; + } + + @Override + protected BtiFormatPartitionWriter openPartitionWriter(SequentialWriter dataWriter, IndexWriter indexWriter) + { + checkNotNull(dataWriter); + checkNotNull(indexWriter); + checkState(!partitionWriterOpened, "Partition writer has been already opened."); + + BtiFormatPartitionWriter partitionWriter = new BtiFormatPartitionWriter(getSerializationHeader(), + getTableMetadataRef().getLocal().comparator, + dataWriter, + indexWriter.rowIndexWriter, + descriptor.version); + partitionWriterOpened = true; + return partitionWriter; + } + + private T ensuringInBuildInternalContext(T value) + { + checkState(value != null, "The requested resource has not been initialized yet."); + return value; + } + @Override protected BtiTableWriter buildInternal(LifecycleNewTracker lifecycleNewTracker, Owner owner) { try { - mmappedRegionsCache = new MmappedRegionsCache(); - dataWriter = DataComponent.buildWriter(descriptor, - getTableMetadataRef().getLocal(), - getIOOptions().writerOptions, - getMetadataCollector(), - lifecycleNewTracker.opType(), - getIOOptions().flushCompression); - - indexWriter = new IndexWriter(this); - partitionWriter = new BtiFormatPartitionWriter(getSerializationHeader(), - getTableMetadataRef().getLocal().comparator, - dataWriter, - indexWriter.rowIndexWriter, - descriptor.version); - + this.mmappedRegionsCache = new MmappedRegionsCache(); + this.operationType = lifecycleNewTracker.opType(); return new BtiTableWriter(this, lifecycleNewTracker, owner); } catch (RuntimeException | Error ex) { - Throwables.closeAndAddSuppressed(ex, partitionWriter, indexWriter, dataWriter, mmappedRegionsCache); + Throwables.closeAndAddSuppressed(ex, mmappedRegionsCache); throw ex; } finally { - partitionWriter = null; - indexWriter = null; - dataWriter = null; mmappedRegionsCache = null; + partitionWriterOpened = false; + indexWriterOpened = false; + dataWriterOpened = false; } } } diff --git a/src/java/org/apache/cassandra/io/util/SequentialWriter.java b/src/java/org/apache/cassandra/io/util/SequentialWriter.java index cb70aaccaf..69643be987 100644 --- a/src/java/org/apache/cassandra/io/util/SequentialWriter.java +++ b/src/java/org/apache/cassandra/io/util/SequentialWriter.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.nio.channels.FileChannel; import java.nio.file.StandardOpenOption; +import java.util.function.LongConsumer; import org.apache.cassandra.io.FSReadError; import org.apache.cassandra.io.FSWriteError; @@ -60,7 +61,7 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr protected long lastFlushOffset; - protected Runnable runPostFlush; + protected LongConsumer runPostFlush; private final TransactionalProxy txnProxy = txnProxy(); @@ -229,7 +230,7 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr resetBuffer(); } - public void setPostFlushListener(Runnable runPostFlush) + public void setPostFlushListener(LongConsumer runPostFlush) { assert this.runPostFlush == null; this.runPostFlush = runPostFlush; @@ -252,7 +253,7 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr throw new FSWriteError(e, getPath()); } if (runPostFlush != null) - runPostFlush.run(); + runPostFlush.accept(getLastFlushOffset()); } @Override diff --git a/test/unit/org/apache/cassandra/db/SerializationHeaderTest.java b/test/unit/org/apache/cassandra/db/SerializationHeaderTest.java index 2b8db525f1..661e3651d3 100644 --- a/test/unit/org/apache/cassandra/db/SerializationHeaderTest.java +++ b/test/unit/org/apache/cassandra/db/SerializationHeaderTest.java @@ -100,7 +100,6 @@ public class SerializationHeaderTest .setTableMetadataRef(TableMetadataRef.forOfflineTools(schema)) .setKeyCount(1) .setSerializationHeader(header) - .setFlushObservers(Collections.emptyList()) .setMetadataCollector(new MetadataCollector(schema.comparator)) .addDefaultComponents(Collections.emptySet()) .build(txn, null)) diff --git a/test/unit/org/apache/cassandra/db/lifecycle/RealTransactionsTest.java b/test/unit/org/apache/cassandra/db/lifecycle/RealTransactionsTest.java index 6d447ddb25..28aab8556e 100644 --- a/test/unit/org/apache/cassandra/db/lifecycle/RealTransactionsTest.java +++ b/test/unit/org/apache/cassandra/db/lifecycle/RealTransactionsTest.java @@ -163,7 +163,7 @@ public class RealTransactionsTest extends SchemaLoader rewriter.switchWriter(desc.getFormat().getWriterFactory().builder(desc) .setTableMetadataRef(metadata) .setSerializationHeader(SerializationHeader.make(cfs.metadata(), txn.originals())) - .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, metadata.get()) + .setSecondaryIndexGroups(cfs.indexManager.listIndexGroups()) .setMetadataCollector(new MetadataCollector(cfs.metadata().comparator)) .addDefaultComponents(cfs.indexManager.listIndexGroups()) .build(txn, cfs)); diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableFlushObserverTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableFlushObserverTest.java index fbeabb95cb..4f3ff5a6fe 100644 --- a/test/unit/org/apache/cassandra/io/sstable/SSTableFlushObserverTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/SSTableFlushObserverTest.java @@ -23,12 +23,16 @@ import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.Iterator; +import java.util.List; +import java.util.function.Supplier; +import java.util.stream.Stream; import com.google.common.collect.ArrayListMultimap; import com.google.common.collect.Multimap; import org.apache.commons.lang3.tuple.ImmutableTriple; import org.apache.commons.lang3.tuple.Triple; import org.junit.Assert; +import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; @@ -50,6 +54,7 @@ import org.apache.cassandra.db.rows.Cell; import org.apache.cassandra.db.rows.EncodingStats; import org.apache.cassandra.db.rows.Row; import org.apache.cassandra.db.rows.Unfiltered; +import org.apache.cassandra.index.Index; import org.apache.cassandra.io.FSWriteError; import org.apache.cassandra.io.sstable.format.SSTableFormat; import org.apache.cassandra.io.sstable.format.SSTableReader; @@ -60,102 +65,169 @@ import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.schema.ColumnMetadata; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.schema.TableMetadataRef; +import org.apache.cassandra.utils.concurrent.Transactional.AbstractTransactional.State; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatRuntimeException; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; public class SSTableFlushObserverTest { + + private TableMetadata cfm; + private File directory; + private SSTableFormat sstableFormat; + private static Supplier idGen; + @BeforeClass public static void initDD() { DatabaseDescriptor.daemonInitialization(); CommitLog.instance.start(); + idGen = SSTableIdFactory.instance.defaultBuilder().generator(Stream.empty()); } private static final String KS_NAME = "test"; private static final String CF_NAME = "flush_observer"; - @Test - public void testFlushObserver() throws Exception + @Before + public void setUp() throws Exception { - TableMetadata cfm = - TableMetadata.builder(KS_NAME, CF_NAME) + cfm = TableMetadata.builder(KS_NAME, CF_NAME) .addPartitionKeyColumn("id", UTF8Type.instance) .addRegularColumn("first_name", UTF8Type.instance) .addRegularColumn("age", Int32Type.instance) .addRegularColumn("height", LongType.instance) .build(); - - LifecycleTransaction transaction = LifecycleTransaction.offline(OperationType.COMPACTION); - FlushObserver observer = new FlushObserver(); - String sstableDirectory = DatabaseDescriptor.getAllDataFileLocations()[0]; - File directory = new File(sstableDirectory + File.pathSeparator() + KS_NAME + File.pathSeparator() + CF_NAME); + directory = new File(sstableDirectory + File.pathSeparator() + KS_NAME + File.pathSeparator() + CF_NAME); directory.deleteOnExit(); if (!directory.exists() && !directory.tryCreateDirectories()) throw new FSWriteError(new IOException("failed to create tmp directory"), directory.absolutePath()); - SSTableFormat sstableFormat = DatabaseDescriptor.getSelectedSSTableFormat(); - Descriptor descriptor = new Descriptor(sstableFormat.getLatestVersion(), - directory, - cfm.keyspace, - cfm.name, - new SequenceBasedSSTableId(0)); + sstableFormat = DatabaseDescriptor.getSelectedSSTableFormat(); + } - SSTableWriter writer = descriptor.getFormat().getWriterFactory().builder(descriptor) - .setKeyCount(10) - .setTableMetadataRef(TableMetadataRef.forOfflineTools(cfm)) - .setMetadataCollector(new MetadataCollector(cfm.comparator).sstableLevel(0)) - .setSerializationHeader(new SerializationHeader(true, cfm, cfm.regularAndStaticColumns(), EncodingStats.NO_STATS)) - .setFlushObservers(Collections.singletonList(observer)) - .addDefaultComponents(Collections.emptySet()) - .build(transaction, null); - - SSTableReader reader; - Multimap> expected = ArrayListMultimap.create(); - - try + @Test + public void testFlushObserver() throws Exception + { + try (LifecycleTransaction transaction = LifecycleTransaction.offline(OperationType.COMPACTION)) { - final long now = System.currentTimeMillis(); + FlushObserver observer = new FlushObserver(); - ByteBuffer key = UTF8Type.instance.fromString("key1"); - expected.putAll(key, Arrays.asList(BufferCell.live(getColumn(cfm, "age"), now, Int32Type.instance.decompose(27)), - BufferCell.live(getColumn(cfm, "first_name"), now, UTF8Type.instance.fromString("jack")), - BufferCell.live(getColumn(cfm, "height"), now, LongType.instance.decompose(183L)))); + Descriptor descriptor = new Descriptor(sstableFormat.getLatestVersion(), + directory, + cfm.keyspace, + cfm.name, + idGen.get()); + Index.Group indexGroup = mock(Index.Group.class); + when(indexGroup.getFlushObserver(any(), any(), any())).thenReturn(observer); + SSTableWriter.Builder writerBuilder = descriptor.getFormat().getWriterFactory().builder(descriptor) + .setKeyCount(10) + .setTableMetadataRef(TableMetadataRef.forOfflineTools(cfm)) + .setMetadataCollector(new MetadataCollector(cfm.comparator).sstableLevel(0)) + .setSerializationHeader(new SerializationHeader(true, cfm, cfm.regularAndStaticColumns(), EncodingStats.NO_STATS)) + .setSecondaryIndexGroups(Collections.singleton(indexGroup)) + .addDefaultComponents(Collections.emptySet()); + assertThat(observer.beginCalled).isFalse(); + assertThat(observer.isComplete).isFalse(); - writer.append(new RowIterator(cfm, key.duplicate(), Collections.singletonList(buildRow(expected.get(key))))); + SSTableWriter writer = writerBuilder.build(transaction, null); - key = UTF8Type.instance.fromString("key2"); - expected.putAll(key, Arrays.asList(BufferCell.live(getColumn(cfm, "age"), now, Int32Type.instance.decompose(30)), - BufferCell.live(getColumn(cfm, "first_name"), now, UTF8Type.instance.fromString("jim")), - BufferCell.live(getColumn(cfm, "height"), now, LongType.instance.decompose(180L)))); + assertThat(observer.beginCalled).isTrue(); + assertThat(observer.isComplete).isFalse(); - writer.append(new RowIterator(cfm, key, Collections.singletonList(buildRow(expected.get(key))))); + SSTableReader reader; + Multimap> expected = ArrayListMultimap.create(); - key = UTF8Type.instance.fromString("key3"); - expected.putAll(key, Arrays.asList(BufferCell.live(getColumn(cfm, "age"), now, Int32Type.instance.decompose(30)), - BufferCell.live(getColumn(cfm, "first_name"), now, UTF8Type.instance.fromString("ken")), - BufferCell.live(getColumn(cfm, "height"), now, LongType.instance.decompose(178L)))); + try + { + final long now = System.currentTimeMillis(); - writer.append(new RowIterator(cfm, key, Collections.singletonList(buildRow(expected.get(key))))); + ByteBuffer key = UTF8Type.instance.fromString("key1"); + expected.putAll(key, Arrays.asList(BufferCell.live(getColumn(cfm, "age"), now, Int32Type.instance.decompose(27)), + BufferCell.live(getColumn(cfm, "first_name"), now, UTF8Type.instance.fromString("jack")), + BufferCell.live(getColumn(cfm, "height"), now, LongType.instance.decompose(183L)))); - reader = writer.finish(true); + writer.append(new RowIterator(cfm, key.duplicate(), Collections.singletonList(buildRow(expected.get(key))))); + + key = UTF8Type.instance.fromString("key2"); + expected.putAll(key, Arrays.asList(BufferCell.live(getColumn(cfm, "age"), now, Int32Type.instance.decompose(30)), + BufferCell.live(getColumn(cfm, "first_name"), now, UTF8Type.instance.fromString("jim")), + BufferCell.live(getColumn(cfm, "height"), now, LongType.instance.decompose(180L)))); + + writer.append(new RowIterator(cfm, key, Collections.singletonList(buildRow(expected.get(key))))); + + key = UTF8Type.instance.fromString("key3"); + expected.putAll(key, Arrays.asList(BufferCell.live(getColumn(cfm, "age"), now, Int32Type.instance.decompose(30)), + BufferCell.live(getColumn(cfm, "first_name"), now, UTF8Type.instance.fromString("ken")), + BufferCell.live(getColumn(cfm, "height"), now, LongType.instance.decompose(178L)))); + + writer.append(new RowIterator(cfm, key, Collections.singletonList(buildRow(expected.get(key))))); + + reader = writer.finish(true); + } + finally + { + FileUtils.closeQuietly(writer); + } + + Assert.assertTrue(observer.isComplete); + Assert.assertEquals(expected.size(), observer.rows.size()); + + for (Triple e : observer.rows.keySet()) + { + ByteBuffer key = e.getLeft(); + long indexPosition = e.getRight(); + + DecoratedKey indexKey = reader.keyAtPositionFromSecondaryIndex(indexPosition); + Assert.assertEquals(0, UTF8Type.instance.compare(key, indexKey.getKey())); + Assert.assertEquals(expected.get(key), observer.rows.get(e)); + } } - finally + } + + @Test + public void testFailedInitialization() throws Exception + { + try (LifecycleTransaction transaction = LifecycleTransaction.offline(OperationType.COMPACTION)) { - FileUtils.closeQuietly(writer); - } + FlushObserver observer1 = new FlushObserver(); + FlushObserver observer2 = new FlushObserver(); + Index.Group indexGroup1 = mock(Index.Group.class); + Index.Group indexGroup2 = mock(Index.Group.class); + when(indexGroup1.getFlushObserver(any(), any(), any())).thenReturn(observer1); + when(indexGroup2.getFlushObserver(any(), any(), any())).thenReturn(observer2); + observer2.failOnBegin = true; - Assert.assertTrue(observer.isComplete); - Assert.assertEquals(expected.size(), observer.rows.size()); + Descriptor descriptor = new Descriptor(sstableFormat.getLatestVersion(), + directory, + cfm.keyspace, + cfm.name, + idGen.get()); + assertThatRuntimeException().isThrownBy(() -> descriptor.getFormat().getWriterFactory().builder(descriptor) + .setKeyCount(10) + .setTableMetadataRef(TableMetadataRef.forOfflineTools(cfm)) + .setMetadataCollector(new MetadataCollector(cfm.comparator).sstableLevel(0)) + .setSerializationHeader(new SerializationHeader(true, cfm, cfm.regularAndStaticColumns(), EncodingStats.NO_STATS)) + .setSecondaryIndexGroups(List.of(indexGroup1, indexGroup2)) + .addDefaultComponents(Collections.emptySet()) + .build(transaction, null) + ).withMessage("Failed to initialize"); - for (Triple e : observer.rows.keySet()) - { - ByteBuffer key = e.getLeft(); - long indexPosition = e.getRight(); + assertThat(observer1.beginCalled).isTrue(); + assertThat(observer1.isComplete).isFalse(); + assertThat(observer1.abortCalled).isTrue(); - DecoratedKey indexKey = reader.keyAtPositionFromSecondaryIndex(indexPosition); - Assert.assertEquals(0, UTF8Type.instance.compare(key, indexKey.getKey())); - Assert.assertEquals(expected.get(key), observer.rows.get(e)); + assertThat(observer2.beginCalled).isTrue(); + assertThat(observer2.isComplete).isFalse(); + assertThat(observer2.abortCalled).isFalse(); + + assertThat(transaction.state()).isEqualTo(State.IN_PROGRESS); + assertThat(transaction.originals()).isEmpty(); } } @@ -190,10 +262,16 @@ public class SSTableFlushObserverTest private Triple currentKey; private boolean isComplete; + private boolean beginCalled; + private boolean failOnBegin; + private boolean abortCalled; @Override public void begin() { + beginCalled = true; + if (failOnBegin) + throw new RuntimeException("Failed to initialize"); } @Override @@ -220,6 +298,12 @@ public class SSTableFlushObserverTest { staticRow.forEach((c) -> staticRows.put(currentKey, (Cell) c)); } + + @Override + public void abort(Throwable accumulate) + { + abortCalled = true; + } } private static Row buildRow(Collection> cells) diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTestBase.java b/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTestBase.java index 0cdd8ac18b..061591500e 100644 --- a/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTestBase.java +++ b/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTestBase.java @@ -163,7 +163,7 @@ public class SSTableWriterTestBase extends SchemaLoader .setPendingRepair(pendingRepair) .setTransientSSTable(isTransient) .setSerializationHeader(new SerializationHeader(true, cfs.metadata(), cfs.metadata().regularAndStaticColumns(), EncodingStats.NO_STATS)) - .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, cfs.metadata.get()) + .setSecondaryIndexGroups(cfs.indexManager.listIndexGroups()) .setMetadataCollector(new MetadataCollector(cfs.metadata().comparator)) .addDefaultComponents(cfs.indexManager.listIndexGroups()) .build(txn, cfs); diff --git a/test/unit/org/apache/cassandra/io/sstable/ScrubTest.java b/test/unit/org/apache/cassandra/io/sstable/ScrubTest.java index bd5b4a5c18..e795254903 100644 --- a/test/unit/org/apache/cassandra/io/sstable/ScrubTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/ScrubTest.java @@ -839,7 +839,6 @@ public class ScrubTest .setTableMetadataRef(cfs.metadata) .setMetadataCollector(collector) .setSerializationHeader(header) - .setFlushObservers(Collections.emptyList()) .addDefaultComponents(Collections.emptySet()) .build(txn, cfs); diff --git a/test/unit/org/apache/cassandra/io/sstable/format/bti/PartitionIndexTest.java b/test/unit/org/apache/cassandra/io/sstable/format/bti/PartitionIndexTest.java index 8f24ced86f..6ceee331f6 100644 --- a/test/unit/org/apache/cassandra/io/sstable/format/bti/PartitionIndexTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/format/bti/PartitionIndexTest.java @@ -429,7 +429,7 @@ public class PartitionIndexTest PartitionIndexBuilder builder = new PartitionIndexBuilder(writer, fhBuilder) ) { - writer.setPostFlushListener(() -> builder.markPartitionIndexSynced(writer.getLastFlushOffset())); + writer.setPostFlushListener(builder::markPartitionIndexSynced); for (int i = 0; i < COUNT; i++) { DecoratedKey key = generateRandomLengthKey(); @@ -506,7 +506,7 @@ public class PartitionIndexTest for (i = 0; i < 3; ++i) builder.addEntry(list.get(i), i); - writer.setPostFlushListener(() -> builder.markPartitionIndexSynced(writer.getLastFlushOffset())); + writer.setPostFlushListener(builder::markPartitionIndexSynced); AtomicInteger callCount = new AtomicInteger(); final int addedSize = i; @@ -669,7 +669,7 @@ public class PartitionIndexTest PartitionIndexBuilder builder = new PartitionIndexBuilder(writer, fhBuilder) ) { - writer.setPostFlushListener(() -> builder.markPartitionIndexSynced(writer.getLastFlushOffset())); + writer.setPostFlushListener(builder::markPartitionIndexSynced); for (int i = 0; i < COUNT; i++) { DecoratedKey key = generateRandomKey(); @@ -708,7 +708,7 @@ public class PartitionIndexTest PartitionIndexBuilder builder = new PartitionIndexBuilder(writer, fhBuilder) ) { - writer.setPostFlushListener(() -> builder.markPartitionIndexSynced(writer.getLastFlushOffset())); + writer.setPostFlushListener(builder::markPartitionIndexSynced); for (int i = 0; i < 1000; i++) { DecoratedKey key = generateRandomKey();