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();