This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch cassandra-5.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cassandra-5.0 by this push:
new 93b5aec70d Isolate SAI index rebuilds from ZCS
93b5aec70d is described below
commit 93b5aec70d161b97069bf396f6a1c470ce09fa68
Author: Pranav Shenoy <[email protected]>
AuthorDate: Wed Aug 26 21:13:20 2026 -0700
Isolate SAI index rebuilds from ZCS
patch by Pranav Shenoy; reviewed by Caleb Rackliffe and David Capwell for
CASSANDRA-21520
---
CHANGES.txt | 1 +
.../cassandra/db/compaction/CompactionManager.java | 8 +-
.../db/streaming/CassandraOutgoingFile.java | 41 ++-
.../db/streaming/CassandraStreamManager.java | 13 +-
.../cassandra/db/streaming/ComponentContext.java | 9 +-
.../cassandra/index/SecondaryIndexBuilder.java | 8 +
.../index/sai/StorageAttachedIndexBuilder.java | 70 ++++-
.../sai/StorageAttachedIndexBuildingSupport.java | 66 ++++-
.../cassandra/io/sstable/format/SSTableReader.java | 10 +
.../sstable/format/SSTableStreamRebuildState.java | 96 +++++++
.../apache/cassandra/streaming/OutgoingStream.java | 7 +
...ndexDeleteDuringEntireSSTableStreamingTest.java | 234 +++++++++++++++++
...dexRebuildDuringEntireSSTableStreamingTest.java | 243 ++++++++++++++++++
.../test/sai/StreamingDuringIndexRebuildTest.java | 155 +++++++++++
.../CassandraStreamManagerDoubleReleaseTest.java | 271 ++++++++++++++++++++
.../db/streaming/ComponentContextTest.java | 121 +++++++++
.../db/streaming/TableStreamManagerTest.java | 258 +++++++++++++++++++
.../index/sai/StreamRebuildCoordinationTest.java | 190 ++++++++++++++
.../format/SSTableStreamRebuildStateTest.java | 285 +++++++++++++++++++++
19 files changed, 2061 insertions(+), 25 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 1cdbf799c1..5ce20dd5f0 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
5.0.10
+ * Isolate SAI index rebuilds from ZCS (CASSANDRA-21520)
* Bound and chunk the BTI partition index preload (CASSANDRA-21632)
* Restore volatile on CommitLogSegment.cdcState dropped in merge
(CASSANDRA-21665)
* Fix compressed scan read-ahead buffer Block leak and chunk-cache pollution
by scans (CASSANDRA-21671)
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index a20dc32a6e..830c483ecf 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@ -1917,7 +1917,13 @@ public class CompactionManager implements
CompactionManagerMBean, ICompactionMan
}
};
- return secondaryIndexExecutor.submitIfRunning(runnable, "index build");
+ Future<?> future = secondaryIndexExecutor.submitIfRunning(runnable,
"index build");
+ if (future.isCancelled())
+ {
+ // Let the builder release any resources it reserved during
construction.
+ builder.onNotExecuted();
+ }
+ return future;
}
/**
diff --git
a/src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java
b/src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java
index 9fc04e7e0d..8950eddf1a 100644
--- a/src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java
+++ b/src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java
@@ -22,6 +22,8 @@ import java.io.IOException;
import java.util.List;
import java.util.Objects;
+import javax.annotation.concurrent.NotThreadSafe;
+
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
@@ -40,6 +42,7 @@ import org.apache.cassandra.utils.concurrent.Ref;
/**
* used to transfer the part(or whole) of a SSTable data file
*/
+@NotThreadSafe
public class CassandraOutgoingFile implements OutgoingStream
{
private final Ref<SSTableReader> ref;
@@ -50,6 +53,9 @@ public class CassandraOutgoingFile implements OutgoingStream
private final StreamOperation operation;
private final CassandraStreamHeader header;
+ // guards against releasing the SSTable streaming status more than once
+ private boolean streamRebuildStatusReleased = false;
+
public CassandraOutgoingFile(StreamOperation operation, Ref<SSTableReader>
ref,
List<SSTableReader.PartitionPositionBounds>
sections, List<Range<Token>> normalizedRanges,
long estimatedKeys)
@@ -64,9 +70,22 @@ public class CassandraOutgoingFile implements OutgoingStream
SSTableReader sstable = ref.get();
this.filename = sstable.getFilename();
- this.shouldStreamEntireSSTable = computeShouldStreamEntireSSTables();
- ComponentManifest manifest = ComponentManifest.create(sstable);
- this.header = makeHeader(sstable, operation, sections, estimatedKeys,
shouldStreamEntireSSTable, manifest);
+
+ // If the SSTable is currently being rebuilt, fall back to legacy
streaming.
+ boolean entire = computeShouldStreamEntireSSTables() &&
sstable.streamRebuildState().tryBeginStreaming();
+ this.shouldStreamEntireSSTable = entire;
+ try
+ {
+ ComponentManifest manifest = ComponentManifest.create(sstable);
+ this.header = makeHeader(sstable, operation, sections,
estimatedKeys, entire, manifest);
+ }
+ catch (RuntimeException | Error e)
+ {
+ // Release the status we just acquired if construction fails.
+ if (entire)
+ sstable.streamRebuildState().endStreaming();
+ throw e;
+ }
}
private static CassandraStreamHeader makeHeader(SSTableReader sstable,
@@ -204,9 +223,25 @@ public class CassandraOutgoingFile implements
OutgoingStream
@Override
public void finish()
{
+ releaseStreamRebuildStatus();
ref.release();
}
+ /**
+ * Releases the entire-sstable streaming status this stream reserved at
construction, exactly once, without
+ * releasing the SSTable reference. Used both by {@link #finish()} and by
callers that must clean up a
+ * constructed-but-never-transferred stream.
+ */
+ @Override
+ public void releaseStreamRebuildStatus()
+ {
+ if (shouldStreamEntireSSTable && !streamRebuildStatusReleased)
+ {
+ streamRebuildStatusReleased = true;
+ ref.get().streamRebuildState().endStreaming();
+ }
+ }
+
public boolean equals(Object o)
{
if (this == o) return true;
diff --git
a/src/java/org/apache/cassandra/db/streaming/CassandraStreamManager.java
b/src/java/org/apache/cassandra/db/streaming/CassandraStreamManager.java
index 6940f11b57..dcee46b7e4 100644
--- a/src/java/org/apache/cassandra/db/streaming/CassandraStreamManager.java
+++ b/src/java/org/apache/cassandra/db/streaming/CassandraStreamManager.java
@@ -90,6 +90,9 @@ public class CassandraStreamManager implements
TableStreamManager
public Collection<OutgoingStream> createOutgoingStreams(StreamSession
session, RangesAtEndpoint replicas, TimeUUID pendingRepair, PreviewKind
previewKind)
{
Refs<SSTableReader> refs = new Refs<>();
+ // Declared outside the try so the catch can release the SSTable
streaming status reserved by any
+ // stream already constructed before a failure.
+ List<OutgoingStream> streams = new ArrayList<>();
try
{
final List<Range<PartitionPosition>> keyRanges = new
ArrayList<>(replicas.size());
@@ -141,18 +144,17 @@ public class CassandraStreamManager implements
TableStreamManager
List<Range<Token>> normalizedFullRanges =
Range.normalize(replicas.onlyFull().ranges());
List<Range<Token>> normalizedAllRanges =
Range.normalize(replicas.ranges());
//Create outgoing file streams for ranges possibly skipping
repaired ranges in sstables
- List<OutgoingStream> streams = new ArrayList<>(refs.size());
- for (SSTableReader sstable : refs)
+ for (SSTableReader sstable : new ArrayList<>(refs))
{
List<Range<Token>> ranges = sstable.isRepaired() ?
normalizedFullRanges : normalizedAllRanges;
List<SSTableReader.PartitionPositionBounds> sections =
sstable.getPositionsForRanges(ranges);
- Ref<SSTableReader> ref = refs.get(sstable);
if (sections.isEmpty())
{
- ref.release();
+ refs.release(sstable);
continue;
}
+ Ref<SSTableReader> ref = refs.get(sstable);
streams.add(new
CassandraOutgoingFile(session.getStreamOperation(), ref, sections, ranges,
sstable.estimatedKeysForRanges(ranges)));
}
@@ -161,6 +163,9 @@ public class CassandraStreamManager implements
TableStreamManager
}
catch (Throwable t)
{
+ // Release the SSTable streaming status held by any
already-constructed stream, so a planning failure cannot leak it.
+ for (OutgoingStream stream : streams)
+ stream.releaseStreamRebuildStatus();
refs.release();
throw t;
}
diff --git a/src/java/org/apache/cassandra/db/streaming/ComponentContext.java
b/src/java/org/apache/cassandra/db/streaming/ComponentContext.java
index c03e7b4c34..2743743b23 100644
--- a/src/java/org/apache/cassandra/db/streaming/ComponentContext.java
+++ b/src/java/org/apache/cassandra/db/streaming/ComponentContext.java
@@ -78,8 +78,13 @@ public class ComponentContext implements AutoCloseable
@SuppressWarnings("resource") // file channel will be closed by Caller
FileChannel channel = toTransfer.newReadChannel();
- assert size == channel.size() : String.format("Entire sstable
streaming expects %s file size to be %s but got %s.",
- component, size,
channel.size());
+ if (size != channel.size())
+ {
+ long actual = channel.size();
+ FileUtils.closeQuietly(channel);
+ throw new IllegalStateException(String.format("Entire SSTable
streaming expects %s file size to be %s but got %s.",
+ component, size,
actual));
+ }
return channel;
}
diff --git a/src/java/org/apache/cassandra/index/SecondaryIndexBuilder.java
b/src/java/org/apache/cassandra/index/SecondaryIndexBuilder.java
index 73dc3345a2..e963648e82 100644
--- a/src/java/org/apache/cassandra/index/SecondaryIndexBuilder.java
+++ b/src/java/org/apache/cassandra/index/SecondaryIndexBuilder.java
@@ -30,4 +30,12 @@ public abstract class SecondaryIndexBuilder extends
CompactionInfo.Holder
{
return false;
}
+
+ /**
+ * Invoked when this builder was created but its {@link #build()} will
never run (e.g. because an executor
+ * rejects the submission during shutdown). Implementations must release
any resources they reserved during construction.
+ */
+ public void onNotExecuted()
+ {
+ }
}
diff --git
a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuilder.java
b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuilder.java
index c7ee124063..03a1d40118 100644
--- a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuilder.java
+++ b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuilder.java
@@ -77,6 +77,9 @@ public class StorageAttachedIndexBuilder extends
SecondaryIndexBuilder
private final boolean isFullRebuild;
private final boolean isInitialBuild;
+ // The reserved SSTables whose rebuild status this builder still owns and
must release.
+ private final Set<SSTableReader> outstandingReservations;
+
private final SortedMap<SSTableReader, Set<StorageAttachedIndex>> sstables;
private long bytesProcessed = 0;
@@ -86,6 +89,15 @@ public class StorageAttachedIndexBuilder extends
SecondaryIndexBuilder
SortedMap<SSTableReader,
Set<StorageAttachedIndex>> sstables,
boolean isFullRebuild,
boolean isInitialBuild)
+ {
+ this(group, sstables, isFullRebuild, isInitialBuild, false);
+ }
+
+ StorageAttachedIndexBuilder(StorageAttachedIndexGroup group,
+ SortedMap<SSTableReader,
Set<StorageAttachedIndex>> sstables,
+ boolean isFullRebuild,
+ boolean isInitialBuild,
+ boolean ownsStreamRebuildStatus)
{
this.group = group;
this.metadata = group.metadata();
@@ -93,6 +105,7 @@ public class StorageAttachedIndexBuilder extends
SecondaryIndexBuilder
this.tracker = group.table().getTracker();
this.isFullRebuild = isFullRebuild;
this.isInitialBuild = isInitialBuild;
+ this.outstandingReservations = ownsStreamRebuildStatus ? new
HashSet<>(sstables.keySet()) : Collections.emptySet();
this.totalSizeInBytes =
sstables.keySet().stream().mapToLong(SSTableReader::uncompressedLength).sum();
}
@@ -103,21 +116,56 @@ public class StorageAttachedIndexBuilder extends
SecondaryIndexBuilder
isInitialBuild ? "initial" :
"non-initial",
isFullRebuild ? "full" :
"partial")));
- for (Map.Entry<SSTableReader, Set<StorageAttachedIndex>> e :
sstables.entrySet())
+ try
{
- SSTableReader sstable = e.getKey();
- Set<StorageAttachedIndex> indexes = e.getValue();
-
- Set<StorageAttachedIndex> existing = validateIndexes(indexes,
sstable.descriptor);
- if (existing.isEmpty())
+ for (Map.Entry<SSTableReader, Set<StorageAttachedIndex>> e :
sstables.entrySet())
{
- logger.debug(logMessage("{} dropped during index build"),
indexes);
- continue;
- }
+ SSTableReader sstable = e.getKey();
+ Set<StorageAttachedIndex> indexes = e.getValue();
- if (indexSSTable(sstable, existing))
- return;
+ Set<StorageAttachedIndex> existing = validateIndexes(indexes,
sstable.descriptor);
+ if (existing.isEmpty())
+ {
+ logger.debug(logMessage("{} dropped during index build"),
indexes);
+ // Nothing to build for this sstable, so release its
reserved status right away.
+ releaseStreamRebuildStatus(sstable);
+ continue;
+ }
+
+ if (indexSSTable(sstable, existing))
+ return;
+
+ // This SSTable's index build completed, so release its
streaming status immediately, which allows
+ // other streaming operations to operate on it.
+ releaseStreamRebuildStatus(sstable);
+ }
}
+ finally
+ {
+ // Release the rebuild status for any sstables not already
released above.
+ releaseStreamRebuildStatus();
+ }
+ }
+
+ @Override
+ public void onNotExecuted()
+ {
+ // If build() will never run, release here to avoid leaving the
reserved sstables stuck in the REBUILDING state.
+ releaseStreamRebuildStatus();
+ }
+
+ private void releaseStreamRebuildStatus()
+ {
+ outstandingReservations.forEach(sstable ->
sstable.streamRebuildState().endRebuild());
+ outstandingReservations.clear();
+ }
+
+ private void releaseStreamRebuildStatus(SSTableReader sstable)
+ {
+ // Only release a status this builder actually still owns, so we never
clear a reservation that was
+ // released earlier and possibly re-taken by another rebuild or stream.
+ if (outstandingReservations.remove(sstable))
+ sstable.streamRebuildState().endRebuild();
}
private String logMessage(String message)
diff --git
a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuildingSupport.java
b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuildingSupport.java
index 7a13b4b186..0a816c9c71 100644
---
a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuildingSupport.java
+++
b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuildingSupport.java
@@ -21,6 +21,9 @@ package org.apache.cassandra.index.sai;
import java.util.Collection;
import java.util.Comparator;
import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;
import java.util.TreeMap;
@@ -32,9 +35,13 @@ import org.apache.cassandra.index.SecondaryIndexBuilder;
import org.apache.cassandra.index.sai.disk.format.IndexDescriptor;
import org.apache.cassandra.io.sstable.SSTableIdFactory;
import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.io.sstable.format.SSTableStreamRebuildState;
class StorageAttachedIndexBuildingSupport implements Index.IndexBuildingSupport
{
+ /** Maximum number of blocked sstable names to log when a rebuild is
rejected, to keep the message bounded. */
+ private static final int MAX_LOGGED_SSTABLES = 16;
+
@Override
public SecondaryIndexBuilder getIndexBuildTask(ColumnFamilyStore cfs,
Set<Index> indexes,
@@ -46,6 +53,8 @@ class StorageAttachedIndexBuildingSupport implements
Index.IndexBuildingSupport
assert group != null : "Index group does not exist for table " +
cfs.keyspace + '.' + cfs.name;
+ // First resolve, without mutating anything, which SSTables each index
will rebuild.
+ Map<StorageAttachedIndex, Collection<SSTableReader>> targets = new
LinkedHashMap<>();
indexes.stream()
.filter((i) -> i instanceof StorageAttachedIndex)
.forEach((i) ->
@@ -62,11 +71,60 @@ class StorageAttachedIndexBuildingSupport implements
Index.IndexBuildingSupport
.collect(Collectors.toList());
}
- group.dropIndexSSTables(ss, sai);
-
- ss.forEach(sstable ->
sstables.computeIfAbsent(sstable, ignore -> new HashSet<>()).add(sai));
+ targets.put(sai, ss);
});
- return new StorageAttachedIndexBuilder(group, sstables, isFullRebuild,
false);
+ // Reserve the per-sstable rebuild status for every unique target
sstable BEFORE deleting any index
+ // components. A zero-copy stream reserves the same status when its
outgoing file is
+ // constructed, so this ensures a rebuild cannot delete/rewrite SAI
components underneath an in-flight
+ // stream. The returned builder owns these reservations and releases
them when it finishes.
+ Set<SSTableReader> reserved = new LinkedHashSet<>();
+ for (Collection<SSTableReader> ss : targets.values())
+ {
+ for (SSTableReader sstable : ss)
+ {
+ if (reserved.contains(sstable))
+ continue;
+ if (sstable.streamRebuildState().tryBeginRebuild())
+ {
+ reserved.add(sstable);
+ }
+ else
+ {
+ reserved.forEach(s -> s.streamRebuildState().endRebuild());
+
+ // For debugging, list the SSTable that are blocking this
rebuild because they are not in a NORMAL
+ // state (a zero-copy stream is in flight). Cap the list
so the message stays bounded even when a
+ // large number of sstables are affected.
+ String blockedSSTables = targets.values().stream()
+
.flatMap(Collection::stream)
+ .distinct()
+ .filter(s ->
s.streamRebuildState().state() != SSTableStreamRebuildState.State.NORMAL)
+ .map(s ->
s.descriptor.toString())
+
.limit(MAX_LOGGED_SSTABLES)
+
.collect(Collectors.joining(", "));
+ throw new RuntimeException(String.format(
+ "Cannot build SAI index while entire SSTable streaming
is in progress. " +
+ "Blocked sstable(s) (up to %d shown): %s",
MAX_LOGGED_SSTABLES, blockedSSTables));
+ }
+ }
+ }
+
+ // Now it is safe to drop existing components and assemble the build
map.
+ try
+ {
+ targets.forEach((sai, ss) ->
+ {
+ group.dropIndexSSTables(ss, sai);
+ ss.forEach(sstable ->
sstables.computeIfAbsent(sstable, ignore -> new HashSet<>()).add(sai));
+ });
+ }
+ catch (RuntimeException | Error e)
+ {
+ reserved.forEach(s -> s.streamRebuildState().endRebuild());
+ throw e;
+ }
+
+ return new StorageAttachedIndexBuilder(group, sstables, isFullRebuild,
false, true);
}
}
diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
index 854a1b486e..9561e9b2db 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
@@ -588,6 +588,14 @@ public abstract class SSTableReader extends SSTable
implements UnfilteredSource,
}
}
+ /**
+ * @return the thread-safe, per-descriptor coordination status between SAI
index rebuilds and entire-SSTable streaming
+ */
+ public SSTableStreamRebuildState streamRebuildState()
+ {
+ return tidy.global.streamRebuildState;
+ }
+
public void setReplaced()
{
synchronized (tidy.global)
@@ -1629,6 +1637,8 @@ public abstract class SSTableReader extends SSTable
implements UnfilteredSource,
private WeakReference<ScheduledFuture<?>> readMeterSyncFuture = NULL;
// shared state managing if the logical sstable has been compacted;
this is used in cleanup
private volatile Runnable obsoletion;
+ // Coordination state for SAI index rebuilds and ZCS for this SSTable.
+ final SSTableStreamRebuildState streamRebuildState = new
SSTableStreamRebuildState();
GlobalTidy(final SSTableReader reader)
{
diff --git
a/src/java/org/apache/cassandra/io/sstable/format/SSTableStreamRebuildState.java
b/src/java/org/apache/cassandra/io/sstable/format/SSTableStreamRebuildState.java
new file mode 100644
index 0000000000..f7c81c46ae
--- /dev/null
+++
b/src/java/org/apache/cassandra/io/sstable/format/SSTableStreamRebuildState.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.io.sstable.format;
+
+import javax.annotation.concurrent.ThreadSafe;
+
+import com.google.common.annotations.VisibleForTesting;
+
+/**
+ * Coordination state between ZCS and SAI index rebuilds for a single logical
SSTable.
+ *
+ * <p>Multiple entire-SSTable streams of the same SSTable can run concurrently
({@code ZCS_STREAMING} with a
+ * reference count), but a rebuild is exclusive ({@code REBUILDING}) and
mutually exclusive with streaming.
+ */
+@ThreadSafe
+public class SSTableStreamRebuildState
+{
+ public enum State
+ {
+ NORMAL,
+ REBUILDING,
+ ZCS_STREAMING
+ }
+
+ private State state = State.NORMAL;
+ private int zcsStreamCount = 0;
+
+ /**
+ * Attempt to begin an entire-SSTable stream.
+ *
+ * @return true if streaming may proceed (caller must later call {@link
#endStreaming()})
+ * false if a rebuild is active and the caller should fall back to
legacy streaming
+ */
+ public synchronized boolean tryBeginStreaming()
+ {
+ if (state == State.REBUILDING)
+ return false;
+ state = State.ZCS_STREAMING;
+ zcsStreamCount++;
+ return true;
+ }
+
+ public synchronized void endStreaming()
+ {
+ // Return to NORMAL when the last stream ends
+ if (zcsStreamCount > 0 && --zcsStreamCount == 0)
+ state = State.NORMAL;
+ }
+
+ /**
+ * Attempt to begin a rebuild.
+ *
+ * @return true if the rebuild may proceed (caller must later call {@link
#endRebuild()})
+ * false if any stream is in progress or another rebuild is
already running
+ */
+ public synchronized boolean tryBeginRebuild()
+ {
+ if (state != State.NORMAL)
+ return false;
+ state = State.REBUILDING;
+ return true;
+ }
+
+ public synchronized void endRebuild()
+ {
+ if (state == State.REBUILDING)
+ state = State.NORMAL;
+ }
+
+ public synchronized State state()
+ {
+ return state;
+ }
+
+ @VisibleForTesting
+ public synchronized int zcsStreamCount()
+ {
+ return zcsStreamCount;
+ }
+}
diff --git a/src/java/org/apache/cassandra/streaming/OutgoingStream.java
b/src/java/org/apache/cassandra/streaming/OutgoingStream.java
index cc42ab6b82..30c902b5f1 100644
--- a/src/java/org/apache/cassandra/streaming/OutgoingStream.java
+++ b/src/java/org/apache/cassandra/streaming/OutgoingStream.java
@@ -42,6 +42,13 @@ public interface OutgoingStream
*/
void finish();
+ /**
+ * Release any coordination status this stream reserved at construction.
+ */
+ default void releaseStreamRebuildStatus()
+ {
+ }
+
long getRepairedAt();
TimeUUID getPendingRepair();
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexDeleteDuringEntireSSTableStreamingTest.java
b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexDeleteDuringEntireSSTableStreamingTest.java
new file mode 100644
index 0000000000..6cb8b864f5
--- /dev/null
+++
b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexDeleteDuringEntireSSTableStreamingTest.java
@@ -0,0 +1,234 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.sai;
+
+import java.nio.channels.FileChannel;
+import java.util.StringJoiner;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.AllArguments;
+import net.bytebuddy.implementation.bind.annotation.FieldValue;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+
+import org.junit.Test;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.streaming.CassandraEntireSSTableStreamWriter;
+import org.apache.cassandra.db.streaming.ComponentContext;
+import org.apache.cassandra.db.streaming.ComponentManifest;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.io.sstable.Component;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.utils.concurrent.CountDownLatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
+import static org.apache.cassandra.distributed.api.Feature.NETWORK;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Verifies that if a streamed SAI component changes size underneath an
in-flight ZCS stream, the sender must fail the
+ * stream rather than ship bytes that disagree with the already-sent {@link
ComponentManifest}. The failed streaming
+ * transaction must be rolled back on the receiver, which must not expose any
partial or size-mismatched data.
+ */
+public class IndexDeleteDuringEntireSSTableStreamingTest extends TestBaseImpl
+{
+ private static final String TABLE = "tbl";
+ private static final String INDEX = "sai_idx";
+ private static final int ROWS = 200;
+
+ @Test
+ public void testIndexDeleteWhileEntireSSTableStreaming() throws Exception
+ {
+ try (Cluster cluster = init(Cluster.build(2)
+ .withDataDirCount(1)
+ .withConfig(c -> c.with(NETWORK,
GOSSIP)
+
.set("stream_entire_sstables", true)
+
.set("autocompaction_on_startup_enabled", false))
+ // Only the sender (node1) needs
the streaming pause installed.
+ .withInstanceInitializer((cl, num)
-> {
+ if (num == 1)
+ BBHelper.install(cl);
+ })
+ .start()))
+ {
+ cluster.disableAutoCompaction(KEYSPACE);
+ cluster.schemaChange(withKeyspace("CREATE TABLE %s." + TABLE + "
(pk int PRIMARY KEY, v text)"));
+ cluster.schemaChange(withKeyspace("CREATE INDEX " + INDEX + " ON
%s." + TABLE + "(v) USING 'sai'"));
+ SAIUtil.waitForIndexQueryable(cluster, KEYSPACE, INDEX);
+
+ IInvokableInstance node1 = cluster.get(1); // sender
+ IInvokableInstance node2 = cluster.get(2); // receiver
+
+ // Populate the sender only and flush to a single SSTable that
carries the SAI components.
+ // node2 owns the same range but has no data yet.
+ for (int i = 0; i < ROWS; i++)
+ node1.executeInternal(withKeyspace("INSERT INTO %s." + TABLE +
"(pk, v) VALUES (?, ?)"), i, "v" + i);
+ node1.flush(KEYSPACE);
+
+ // Arm the sender-side pause so only the upcoming ZCS stream is
intercepted.
+ node1.runOnInstance(() -> BBHelper.armed.set(true));
+
+ // Kick off the streaming asynchronously so node2 rebuilds its
data from node1. The call blocks until
+ // streaming completes, and streaming is about to block on the
sender, so run it off-thread.
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try
+ {
+ Future<NodeToolResult> streaming =
+ executor.submit(() -> node2.nodetoolResult("rebuild",
"--keyspace", KEYSPACE));
+
+ // Wait until the sender has serialized and flushed the
manifest and is paused before opening the
+ // outgoing component channels.
+ node1.runOnInstance(() ->
BBHelper.manifestSent.awaitUninterruptibly(2, TimeUnit.MINUTES));
+
+ // While the stream is paused mid-flight (status is already
ZCS_STREAMING for this SSTable), a
+ // concurrent SAI rebuild must fail fast rather than mutate
components underneath the in-flight
+ // stream. Directly truncate one of the streamed SAI component
files on the sender. SAI components are
+ // not part of mutableComponents(), so they are not
hard-linked and are streamed from the live file.
+ // When the writer opens the channel the on-disk size no
longer matches the size advertised in the
+ // manifest, and the production size check in
ComponentContext.channel() must fail the stream rather
+ // than ship corrupt bytes.
+
node1.runOnInstance(IndexDeleteDuringEntireSSTableStreamingTest::truncateStreamedSaiComponent);
+
+ // Let the paused stream continue now that a component has
been mutated underneath it.
+ node1.runOnInstance(() -> BBHelper.proceed.decrement());
+
+ NodeToolResult result = streaming.get(3, TimeUnit.MINUTES);
+ result.asserts().failure();
+ assertThat(node1.logs()
+ .grep("Entire SSTable streaming expects .*
file size to be .* but got")
+ .getResult())
+ .describedAs("Expected the ComponentContext size check to
fail the stream")
+ .isNotEmpty();
+ }
+ finally
+ {
+ executor.shutdownNow();
+ }
+
+ for (int i = 0; i < ROWS; i++)
+ {
+ assertThat(node2.executeInternal(withKeyspace("SELECT pk FROM
%s." + TABLE + " WHERE pk = ?"), i))
+ .describedAs("Receiver must not retain data from the
failed stream (pk=%d)", i)
+ .isEmpty();
+ }
+ }
+ }
+
+ private static void truncateStreamedSaiComponent()
+ {
+ ColumnFamilyStore cfs =
Keyspace.open(KEYSPACE).getColumnFamilyStore(TABLE);
+ SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
+ for (Component component : sstable.getStreamingComponents())
+ {
+ // Only mutate a component that is streamed from the live file
(not a hard-linked mutable component and
+ // not a primary component), i.e. a SAI index component, so the
size divergence hits ComponentContext.
+ if
(sstable.descriptor.getFormat().primaryComponents().contains(component))
+ continue;
+ if
(sstable.descriptor.getFormat().mutableComponents().contains(component))
+ continue;
+ File file = sstable.descriptor.fileFor(component);
+ if (!file.exists() || file.length() == 0)
+ continue;
+ try (FileChannel channel = file.newReadWriteChannel())
+ {
+ channel.truncate(Math.max(0, channel.size() - 1));
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ return;
+ }
+ throw new IllegalStateException("No streamed SAI component available
to mutate");
+ }
+
+ public static class BBHelper
+ {
+ // Per-instance state, resolved in the sender's classloader.
+ static final AtomicBoolean armed = new AtomicBoolean(false);
+ static final AtomicBoolean paused = new AtomicBoolean(false);
+ static final CountDownLatch manifestSent =
CountDownLatch.newCountDownLatch(1);
+ static final CountDownLatch proceed =
CountDownLatch.newCountDownLatch(1);
+
+ @SuppressWarnings("unused")
+ public static void write(@FieldValue("manifest") ComponentManifest
manifest, @SuperCall Callable<Void> zuper) throws Exception
+ {
+ if (armed.get() && paused.compareAndSet(false, true))
+ {
+ StringJoiner sj = new StringJoiner(", ", "{", "}");
+ for (Component component : manifest.components())
+ sj.add(component.name + '=' + manifest.sizeOf(component));
+ System.out.println("[ZCS-VERIFY] manifest sent
(component=advertisedSize): " + sj);
+
+ manifestSent.decrement();
+ proceed.awaitUninterruptibly(2, TimeUnit.MINUTES);
+ }
+ zuper.call();
+ }
+
+ @SuppressWarnings("unused")
+ public static FileChannel channel(@AllArguments Object[] args,
@SuperCall Callable<FileChannel> zuper) throws Exception
+ {
+ Component component = (Component) args[1];
+ long advertised = (Long) args[2];
+ try
+ {
+ FileChannel channel = zuper.call();
+ System.out.println("[ZCS-VERIFY] streaming component " +
component.name +
+ ": advertisedSize=" + advertised + ",
actualSize=" + channel.size());
+ return channel;
+ }
+ catch (Throwable t)
+ {
+ System.out.println("[ZCS-VERIFY] streaming component " +
component.name +
+ ": advertisedSize=" + advertised + " ->
FAILED TO OPEN: " + t);
+ throw t;
+ }
+ }
+
+ public static void install(ClassLoader classLoader)
+ {
+ new ByteBuddy().rebase(CassandraEntireSSTableStreamWriter.class)
+ .method(named("write").and(takesArguments(1)))
+ .intercept(MethodDelegation.to(BBHelper.class))
+ .make()
+ .load(classLoader,
ClassLoadingStrategy.Default.INJECTION);
+ new ByteBuddy().rebase(ComponentContext.class)
+ .method(named("channel").and(takesArguments(3)))
+ .intercept(MethodDelegation.to(BBHelper.class))
+ .make()
+ .load(classLoader,
ClassLoadingStrategy.Default.INJECTION);
+ }
+ }
+}
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexRebuildDuringEntireSSTableStreamingTest.java
b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexRebuildDuringEntireSSTableStreamingTest.java
new file mode 100644
index 0000000000..a143721b8d
--- /dev/null
+++
b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexRebuildDuringEntireSSTableStreamingTest.java
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.sai;
+
+import java.nio.channels.FileChannel;
+import java.util.Collections;
+import java.util.StringJoiner;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.AllArguments;
+import net.bytebuddy.implementation.bind.annotation.FieldValue;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+
+import org.junit.Test;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.streaming.CassandraEntireSSTableStreamWriter;
+import org.apache.cassandra.db.streaming.ComponentContext;
+import org.apache.cassandra.db.streaming.ComponentManifest;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.io.sstable.Component;
+import org.apache.cassandra.utils.concurrent.CountDownLatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
+import static org.apache.cassandra.distributed.api.Feature.NETWORK;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Verifies that a concurrent SAI rebuild fails fast while an entire-SSTable
(ZCS) stream is in flight, and that the
+ * stream completes with correct data on the receiver.
+ */
+public class IndexRebuildDuringEntireSSTableStreamingTest extends TestBaseImpl
+{
+ private static final String TABLE = "tbl";
+ private static final String INDEX = "sai_idx";
+ private static final int ROWS = 200;
+
+ @Test
+ public void testIndexRebuildWhileEntireSSTableStreaming() throws Exception
+ {
+ try (Cluster cluster = init(Cluster.build(2)
+ .withDataDirCount(1)
+ .withConfig(c -> c.with(NETWORK,
GOSSIP)
+
.set("stream_entire_sstables", true)
+
.set("autocompaction_on_startup_enabled", false))
+ // Only the sender (node1) needs
the streaming pause installed.
+ .withInstanceInitializer((cl, num)
-> {
+ if (num == 1)
+ BBHelper.install(cl);
+ })
+ .start()))
+ {
+ cluster.disableAutoCompaction(KEYSPACE);
+ cluster.schemaChange(withKeyspace("CREATE TABLE %s." + TABLE + "
(pk int PRIMARY KEY, v text)"));
+ cluster.schemaChange(withKeyspace("CREATE INDEX " + INDEX + " ON
%s." + TABLE + "(v) USING 'sai'"));
+ SAIUtil.waitForIndexQueryable(cluster, KEYSPACE, INDEX);
+
+ IInvokableInstance node1 = cluster.get(1); // sender
+ IInvokableInstance node2 = cluster.get(2); // receiver
+
+ // Populate the sender only and flush to a single SSTable that
carries the SAI components.
+ // node2 owns the same range (RF == cluster size) but has no data
yet.
+ for (int i = 0; i < ROWS; i++)
+ node1.executeInternal(withKeyspace("INSERT INTO %s." + TABLE +
"(pk, v) VALUES (?, ?)"), i, "v" + i);
+ node1.flush(KEYSPACE);
+
+ // Arm the sender-side pause so only the upcoming ZCS stream is
intercepted.
+ node1.runOnInstance(() -> BBHelper.armed.set(true));
+
+ // Kick off the streaming asynchronously so node2 rebuilds its
data from node1. The call blocks until
+ // streaming completes, and streaming is about to block on the
sender, so run it off-thread.
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try
+ {
+ Future<NodeToolResult> streaming =
+ executor.submit(() -> node2.nodetoolResult("rebuild",
"--keyspace", KEYSPACE));
+
+ // Wait until the sender has serialized and flushed the
manifest and is paused before opening the
+ // outgoing component channels.
+ node1.runOnInstance(() ->
BBHelper.manifestSent.awaitUninterruptibly(2, TimeUnit.MINUTES));
+
+ // While the stream is paused mid-flight (status is already
ZCS_STREAMING for this sstable), a
+ // concurrent SAI rebuild must fail fast rather than mutate
components underneath the in-flight stream.
+ boolean rebuildRejected =
node1.callOnInstance(IndexRebuildDuringEntireSSTableStreamingTest::rebuildAndReturnRejected);
+ assertThat(rebuildRejected)
+ .describedAs("SAI rebuild must fail fast while ZCS is in
progress")
+ .isTrue();
+ assertThat(node1.logs()
+ .grep("while entire SSTable streaming is in
progress")
+ .getResult())
+ .describedAs("Expected the rebuild to be rejected due to
active streaming")
+ .isNotEmpty();
+
+ // Let the paused stream continue; its components were never
mutated, so it completes cleanly.
+ node1.runOnInstance(() -> BBHelper.proceed.decrement());
+
+ NodeToolResult result = streaming.get(3, TimeUnit.MINUTES);
+ result.asserts().success();
+ }
+ finally
+ {
+ executor.shutdownNow();
+ }
+
+ for (int i = 0; i < ROWS; i++)
+ {
+ Object[][] byPrimaryKey =
node2.executeInternal(withKeyspace("SELECT pk FROM %s." + TABLE + " WHERE pk =
?"), i);
+ assertThat(byPrimaryKey.length)
+ .describedAs("Receiver must retain streamed data (pk=%d)",
i)
+ .isEqualTo(1);
+ Object[][] byIndex =
node2.executeInternal(withKeyspace("SELECT pk FROM %s." + TABLE + " WHERE v =
?"), "v" + i);
+ assertThat(byIndex.length)
+ .describedAs("Receiver SAI index must return streamed row
(v%d)", i)
+ .isEqualTo(1);
+ assertThat(byIndex[0][0]).isEqualTo(i);
+ }
+
+ // A rejected full rebuild leaves the index marked for rebuild.
The sender's data is intact, and once
+ // streaming has completed and released the status a fresh rebuild
succeeds.
+ for (int i = 0; i < ROWS; i++)
+ {
+ Object[][] onSender =
node1.executeInternal(withKeyspace("SELECT pk FROM %s." + TABLE + " WHERE pk =
?"), i);
+ assertThat(onSender.length).describedAs("Sender data must be
intact (pk=%d)", i).isEqualTo(1);
+ }
+
+ node1.runOnInstance(() -> {
+ ColumnFamilyStore cfs =
Keyspace.open(KEYSPACE).getColumnFamilyStore(TABLE);
+
cfs.indexManager.rebuildIndexesBlocking(Collections.singleton(INDEX));
+ });
+ SAIUtil.waitForIndexQueryable(cluster, KEYSPACE, INDEX);
+
+ for (int i = 0; i < ROWS; i++)
+ {
+ Object[][] onSender =
node1.executeInternal(withKeyspace("SELECT pk FROM %s." + TABLE + " WHERE v =
?"), "v" + i);
+ assertThat(onSender.length).describedAs("Sender index must be
queryable after re-rebuild (v%d)", i).isEqualTo(1);
+ assertThat(onSender[0][0]).isEqualTo(i);
+ }
+ }
+ }
+
+ private static boolean rebuildAndReturnRejected()
+ {
+ try
+ {
+ ColumnFamilyStore cfs =
Keyspace.open(KEYSPACE).getColumnFamilyStore(TABLE);
+
cfs.indexManager.rebuildIndexesBlocking(Collections.singleton(INDEX));
+ return false;
+ }
+ catch (Throwable t)
+ {
+
org.slf4j.LoggerFactory.getLogger(IndexRebuildDuringEntireSSTableStreamingTest.class).error(t.getMessage(),
t);
+ return true;
+ }
+ }
+
+ public static class BBHelper
+ {
+ static final AtomicBoolean armed = new AtomicBoolean(false);
+ static final AtomicBoolean paused = new AtomicBoolean(false);
+ static final CountDownLatch manifestSent =
CountDownLatch.newCountDownLatch(1);
+ static final CountDownLatch proceed =
CountDownLatch.newCountDownLatch(1);
+
+ @SuppressWarnings("unused")
+ public static void write(@FieldValue("manifest") ComponentManifest
manifest, @SuperCall Callable<Void> zuper) throws Exception
+ {
+ if (armed.get() && paused.compareAndSet(false, true))
+ {
+ StringJoiner sj = new StringJoiner(", ", "{", "}");
+ for (Component component : manifest.components())
+ sj.add(component.name + '=' + manifest.sizeOf(component));
+ System.out.println("[ZCS-VERIFY] manifest sent
(component=advertisedSize): " + sj);
+
+ manifestSent.decrement();
+ proceed.awaitUninterruptibly(2, TimeUnit.MINUTES);
+ }
+ zuper.call();
+ }
+
+ @SuppressWarnings("unused")
+ public static FileChannel channel(@AllArguments Object[] args,
@SuperCall Callable<FileChannel> zuper) throws Exception
+ {
+ Component component = (Component) args[1];
+ long advertised = (Long) args[2];
+ try
+ {
+ FileChannel channel = zuper.call();
+ System.out.println("[ZCS-VERIFY] streaming component " +
component.name +
+ ": advertisedSize=" + advertised + ",
actualSize=" + channel.size());
+ return channel;
+ }
+ catch (Throwable t)
+ {
+ System.out.println("[ZCS-VERIFY] streaming component " +
component.name +
+ ": advertisedSize=" + advertised + " ->
MISMATCH: " + t.getMessage());
+ throw t;
+ }
+ }
+
+ public static void install(ClassLoader classLoader)
+ {
+ new ByteBuddy().rebase(CassandraEntireSSTableStreamWriter.class)
+ .method(named("write").and(takesArguments(1)))
+ .intercept(MethodDelegation.to(BBHelper.class))
+ .make()
+ .load(classLoader,
ClassLoadingStrategy.Default.INJECTION);
+ new ByteBuddy().rebase(ComponentContext.class)
+ .method(named("channel").and(takesArguments(3)))
+ .intercept(MethodDelegation.to(BBHelper.class))
+ .make()
+ .load(classLoader,
ClassLoadingStrategy.Default.INJECTION);
+ }
+ }
+}
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/sai/StreamingDuringIndexRebuildTest.java
b/test/distributed/org/apache/cassandra/distributed/test/sai/StreamingDuringIndexRebuildTest.java
new file mode 100644
index 0000000000..61b8688a8c
--- /dev/null
+++
b/test/distributed/org/apache/cassandra/distributed/test/sai/StreamingDuringIndexRebuildTest.java
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.sai;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+
+import org.junit.Test;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.streaming.CassandraEntireSSTableStreamWriter;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
+import static org.apache.cassandra.distributed.api.Feature.NETWORK;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * When an SAI index rebuild is in progress on the sender, a new
entire-SSTable (ZCS) stream of the same SSTable must
+ * degrade to legacy streaming, so the rebuild and the stream never
mutate/ship the SAI components concurrently.
+ */
+public class StreamingDuringIndexRebuildTest extends TestBaseImpl
+{
+ private static final String TABLE = "tbl";
+ private static final String INDEX = "sai_idx";
+ private static final int ROWS = 200;
+
+ @Test
+ public void testStreamingFallsBackToLegacyDuringRebuild() throws Exception
+ {
+ try (Cluster cluster = init(Cluster.build(2)
+ .withDataDirCount(1)
+ .withConfig(c -> c.with(NETWORK,
GOSSIP)
+
.set("stream_entire_sstables", true)
+
.set("autocompaction_on_startup_enabled", false))
+ // Only the sender (node1) needs
the entire-SSTable writer counter installed.
+ .withInstanceInitializer((cl, num)
-> {
+ if (num == 1)
+ BBHelper.install(cl);
+ })
+ .start()))
+ {
+ cluster.disableAutoCompaction(KEYSPACE);
+ cluster.schemaChange(withKeyspace("CREATE TABLE %s." + TABLE + "
(pk int PRIMARY KEY, v text)"));
+ cluster.schemaChange(withKeyspace("CREATE INDEX " + INDEX + " ON
%s." + TABLE + "(v) USING 'sai'"));
+ SAIUtil.waitForIndexQueryable(cluster, KEYSPACE, INDEX);
+
+ IInvokableInstance node1 = cluster.get(1); // sender
+ IInvokableInstance node2 = cluster.get(2); // receiver
+
+ for (int i = 0; i < ROWS; i++)
+ node1.executeInternal(withKeyspace("INSERT INTO %s." + TABLE +
"(pk, v) VALUES (?, ?)"), i, "v" + i);
+ node1.flush(KEYSPACE);
+
+ // Model an in-progress SAI rebuild on the sender by reserving the
per-SSTable rebuild status for every
+ // live SSTable. While this is held, an entire-SSTable stream of
the same sstable must fall back to legacy.
+ node1.runOnInstance(() -> {
+ ColumnFamilyStore cfs =
Keyspace.open(KEYSPACE).getColumnFamilyStore(TABLE);
+ cfs.getLiveSSTables().forEach(sstable -> {
+ if (!sstable.streamRebuildState().tryBeginRebuild())
+ throw new IllegalStateException("Could not reserve
rebuild status for " + sstable.descriptor);
+ });
+ });
+
+ node1.runOnInstance(() -> BBHelper.armed.set(true));
+
+ try
+ {
+ // node2 rebuilds its data from node1. The sender must NOT use
ZCS while a rebuild holds the SSTable, so
+ // it degrades to legacy streaming.
+ NodeToolResult result = node2.nodetoolResult("rebuild",
"--keyspace", KEYSPACE);
+ result.asserts().success();
+
+ int entireWrites = node1.callOnInstance(() ->
BBHelper.entireSSTableWrites.get());
+ assertThat(entireWrites)
+ .describedAs("ZCS must not run while a rebuild holds the
sstable")
+ .isZero();
+ }
+ finally
+ {
+ // Release the reserved status so the sender returns to normal.
+ node1.runOnInstance(() -> {
+ ColumnFamilyStore cfs =
Keyspace.open(KEYSPACE).getColumnFamilyStore(TABLE);
+ cfs.getLiveSSTables().forEach(sstable ->
sstable.streamRebuildState().endRebuild());
+ });
+ }
+
+ // The receiver built its SAI index locally from the legacy stream
and must expose the streamed rows.
+ SAIUtil.waitForIndexQueryable(cluster, KEYSPACE, INDEX);
+ for (int i = 0; i < ROWS; i++)
+ {
+ Object[][] byPrimaryKey =
node2.executeInternal(withKeyspace("SELECT pk FROM %s." + TABLE + " WHERE pk =
?"), i);
+ assertThat(byPrimaryKey.length)
+ .describedAs("Receiver must retain streamed data (pk=%d)",
i)
+ .isEqualTo(1);
+ Object[][] byIndex =
node2.executeInternal(withKeyspace("SELECT pk FROM %s." + TABLE + " WHERE v =
?"), "v" + i);
+ assertThat(byIndex.length)
+ .describedAs("Receiver SAI index must return streamed row
(v%d)", i)
+ .isEqualTo(1);
+ assertThat(byIndex[0][0]).isEqualTo(i);
+ }
+ }
+ }
+
+ public static class BBHelper
+ {
+ static final AtomicBoolean armed = new AtomicBoolean(false);
+ static final AtomicInteger entireSSTableWrites = new AtomicInteger(0);
+
+ @SuppressWarnings("unused")
+ public static void write(@SuperCall Callable<Void> zuper) throws
Exception
+ {
+ if (armed.get())
+ entireSSTableWrites.incrementAndGet();
+ zuper.call();
+ }
+
+ public static void install(ClassLoader classLoader)
+ {
+ new ByteBuddy().rebase(CassandraEntireSSTableStreamWriter.class)
+ .method(named("write").and(takesArguments(1)))
+ .intercept(MethodDelegation.to(BBHelper.class))
+ .make()
+ .load(classLoader,
ClassLoadingStrategy.Default.INJECTION);
+ }
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/db/streaming/CassandraStreamManagerDoubleReleaseTest.java
b/test/unit/org/apache/cassandra/db/streaming/CassandraStreamManagerDoubleReleaseTest.java
new file mode 100644
index 0000000000..643abc6355
--- /dev/null
+++
b/test/unit/org/apache/cassandra/db/streaming/CassandraStreamManagerDoubleReleaseTest.java
@@ -0,0 +1,271 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.db.streaming;
+
+import java.io.IOException;
+import java.net.UnknownHostException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Sets;
+
+import org.jboss.byteman.contrib.bmunit.BMRule;
+import org.jboss.byteman.contrib.bmunit.BMRules;
+import org.jboss.byteman.contrib.bmunit.BMUnitRunner;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.Util;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.RangesAtEndpoint;
+import org.apache.cassandra.locator.Replica;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.schema.CompactionParams;
+import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.streaming.OutgoingStream;
+import org.apache.cassandra.streaming.PreviewKind;
+import org.apache.cassandra.streaming.StreamOperation;
+import org.apache.cassandra.streaming.StreamSession;
+import org.apache.cassandra.streaming.StreamingChannel;
+import org.apache.cassandra.streaming.async.NettyStreamingConnectionFactory;
+import org.apache.cassandra.utils.TimeUUID;
+import org.apache.cassandra.utils.concurrent.Ref;
+
+import static
org.apache.cassandra.service.ActiveRepairService.NO_PENDING_REPAIR;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests that an empty-sections SSTable reference is not double-released if a
subsequent stream
+ * construction failure occurs in {@link
CassandraStreamManager#createOutgoingStreams}.
+ *
+ * <p>When an SSTable has no sections for the requested ranges, its reference
must be released and
+ * removed from {@code refs}. If it is only released without removal, a
failure when creating a later
+ * stream causes the catch block's {@code refs.release()} to release it a
second time, throwing an
+ * {@link IllegalStateException}.
+ */
+@RunWith(BMUnitRunner.class)
+public class CassandraStreamManagerDoubleReleaseTest
+{
+ private static final String table = "tbl";
+ private static final StreamingChannel.Factory connectionFactory = new
NettyStreamingConnectionFactory();
+
+ private static final int N = 1;
+ private static final int RETRY_BUDGET = 20;
+
+ private static final String DOUBLE_RELEASE_MSG = "Attempted to release a
reference that has already been released";
+ private static final String INJECTED_MSG = "injected component manifest
failure";
+
+ // Set once getPositionsForRanges returns empty, ensuring failure is
injected only after an empty-sections ref was released.
+ private static final AtomicBoolean emptySeen = new AtomicBoolean(false);
+
+ // Byteman callbacks
+ public static void markEmptySeen() { emptySeen.set(true); }
+ public static boolean shouldThrow() { return emptySeen.get(); }
+
+ private String keyspace;
+ private TableMetadata tbm;
+ private ColumnFamilyStore cfs;
+
+ @BeforeClass
+ public static void setupClass() throws Exception
+ {
+ SchemaLoader.prepareServer();
+ }
+
+ @Before
+ public void createKeyspace()
+ {
+ keyspace = String.format("ks_%s", System.currentTimeMillis());
+ tbm = CreateTableStatement.parse(String.format("CREATE TABLE %s (k INT
PRIMARY KEY, v INT)", table), keyspace)
+
.compaction(CompactionParams.stcs(Collections.emptyMap()))
+ .build();
+ SchemaLoader.createKeyspace(keyspace, KeyspaceParams.simple(1), tbm);
+ cfs = Schema.instance.getColumnFamilyStoreInstance(tbm.id);
+ }
+
+ @Test
+ @BMRules(rules = {
+ @BMRule(name = "record empty getPositionsForRanges result",
+ targetClass =
"org.apache.cassandra.io.sstable.format.SSTableReader",
+ targetMethod = "getPositionsForRanges",
+ targetLocation = "AT EXIT",
+ condition = "$!.isEmpty()",
+ action =
"org.apache.cassandra.db.streaming.CassandraStreamManagerDoubleReleaseTest.markEmptySeen()"),
+ @BMRule(name = "throw from ComponentManifest.create once an
empty-sections ref was released",
+ targetClass =
"org.apache.cassandra.db.streaming.ComponentManifest",
+ targetMethod = "create",
+ targetLocation = "AT ENTRY",
+ condition =
"org.apache.cassandra.db.streaming.CassandraStreamManagerDoubleReleaseTest.shouldThrow()",
+ action = "throw new java.lang.RuntimeException(\"injected
component manifest failure\")")
+ })
+ public void emptySectionsRefIsNotDoubleReleasedOnExceptionPath() throws
Exception
+ {
+ // Refs iteration order is determined by HashMap bucket layout.
Alternating sstable creation order
+ // ensures an empty-sections sstable is processed before a failing
sstable within two attempts.
+ for (int attempt = 0; attempt < RETRY_BUDGET; attempt++)
+ {
+ RangesAtEndpoint replicas = freshSSTables(attempt);
+ emptySeen.set(false);
+
+ Collection<OutgoingStream> streams = null;
+ try
+ {
+ streams =
cfs.getStreamManager().createOutgoingStreams(session(NO_PENDING_REPAIR),
+
replicas,
+
NO_PENDING_REPAIR,
+
PreviewKind.NONE);
+ // The failing sstable was processed before the empty one, so
no exception was injected; retry.
+ releaseStreams(streams);
+ }
+ catch (IllegalStateException e)
+ {
+ if (e.getMessage() != null &&
e.getMessage().contains(DOUBLE_RELEASE_MSG))
+ fail("Empty-sections Ref was double-released on exception
path: " + e);
+ throw e;
+ }
+ catch (RuntimeException e)
+ {
+ // Injected failure propagated cleanly without
double-releasing any refs.
+ if (e.getMessage() != null &&
e.getMessage().contains(INJECTED_MSG))
+ return;
+ throw e;
+ }
+ }
+
+ fail("Could not process an empty-sections sstable before a failing
sstable within " + RETRY_BUDGET + " attempts");
+ }
+
+ /**
+ * Rebuilds the CFS with N repaired (empty-sections) sstables and N
unrepaired (non-empty)
+ * sstables, and returns a transient-only replica set. A repaired sstable
is streamed against
+ * {@code replicas.onlyFull()} (empty here) so its sections are empty; an
unrepaired sstable is
+ * streamed against {@code replicas.ranges()} (the whole ring) so its
sections are non-empty.
+ */
+ private RangesAtEndpoint freshSSTables(int attempt) throws Exception
+ {
+ cfs.truncateBlocking();
+
+ long repairedAt = System.currentTimeMillis();
+ Runnable makeRepaired = () -> {
+ for (int i = 0; i < N; i++)
+ {
+ final int k = 100 + i;
+ SSTableReader repaired = createSSTable(() ->
QueryProcessor.executeInternal(
+ String.format("INSERT INTO %s.%s (k, v) VALUES (%d, %d)",
keyspace, table, k, k)));
+ try
+ {
+ mutateRepaired(repaired, repairedAt);
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+ };
+ Runnable makeUnrepaired = () -> {
+ for (int i = 0; i < N; i++)
+ {
+ final int k = 200 + i;
+ createSSTable(() -> QueryProcessor.executeInternal(
+ String.format("INSERT INTO %s.%s (k, v) VALUES (%d, %d)",
keyspace, table, k, k)));
+ }
+ };
+
+ if (attempt % 2 == 0)
+ {
+ makeRepaired.run();
+ makeUnrepaired.run();
+ }
+ else
+ {
+ makeUnrepaired.run();
+ makeRepaired.run();
+ }
+
+ IPartitioner partitioner = DatabaseDescriptor.getPartitioner();
+ Range<Token> wholeRing = new Range<>(partitioner.getMinimumToken(),
partitioner.getMinimumToken());
+ InetAddressAndPort local = InetAddressAndPort.getByName("127.0.0.1");
+ // transient replica (full=false): onlyFull() is empty => repaired
sstables get empty sections
+ return RangesAtEndpoint.of(new Replica(local, wholeRing, false));
+ }
+
+ private SSTableReader createSSTable(Runnable queryable)
+ {
+ Set<SSTableReader> before = cfs.getLiveSSTables();
+ queryable.run();
+ Util.flush(cfs);
+ Set<SSTableReader> after = cfs.getLiveSSTables();
+ return Iterables.getOnlyElement(Sets.difference(after, before));
+ }
+
+ private static void mutateRepaired(SSTableReader sstable, long repairedAt)
throws IOException
+ {
+ Descriptor descriptor = sstable.descriptor;
+ descriptor.getMetadataSerializer().mutateRepairMetadata(descriptor,
repairedAt, NO_PENDING_REPAIR, false);
+ sstable.reloadSSTableMetadata();
+ }
+
+ private static void releaseStreams(Collection<OutgoingStream> streams)
+ {
+ if (streams == null)
+ return;
+ for (OutgoingStream stream : streams)
+ {
+ Ref<SSTableReader> ref =
CassandraOutgoingFile.fromStream(stream).getRef();
+ ref.release();
+ }
+ }
+
+ private static StreamSession session(TimeUUID pendingRepair)
+ {
+ try
+ {
+ return new StreamSession(StreamOperation.REPAIR,
+ InetAddressAndPort.getByName("127.0.0.1"),
+ connectionFactory,
+ null,
+ MessagingService.current_version,
+ false,
+ 0,
+ pendingRepair,
+ PreviewKind.NONE);
+ }
+ catch (UnknownHostException e)
+ {
+ throw new AssertionError(e);
+ }
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/db/streaming/ComponentContextTest.java
b/test/unit/org/apache/cassandra/db/streaming/ComponentContextTest.java
new file mode 100644
index 0000000000..e081b1eaad
--- /dev/null
+++ b/test/unit/org/apache/cassandra/db/streaming/ComponentContextTest.java
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.db.streaming;
+
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.Util;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.RowUpdateBuilder;
+import org.apache.cassandra.db.compaction.CompactionManager;
+import org.apache.cassandra.io.sstable.Component;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class ComponentContextTest
+{
+ private static final String KEYSPACE = "ComponentContextTest";
+ private static final String CF_STANDARD = "Standard1";
+
+ private static SSTableReader sstable;
+
+ @BeforeClass
+ public static void defineSchemaAndPrepareSSTable()
+ {
+ SchemaLoader.prepareServer();
+ SchemaLoader.createKeyspace(KEYSPACE,
+ KeyspaceParams.simple(1),
+ SchemaLoader.standardCFMD(KEYSPACE,
CF_STANDARD));
+
+ ColumnFamilyStore store =
Keyspace.open(KEYSPACE).getColumnFamilyStore(CF_STANDARD);
+ CompactionManager.instance.disableAutoCompaction();
+ for (int j = 0; j < 10; j++)
+ {
+ new RowUpdateBuilder(store.metadata(), j, String.valueOf(j))
+ .clustering("0")
+ .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
+ .build()
+ .applyUnsafe();
+ }
+ Util.flush(store);
+ CompactionManager.instance.performMaximal(store, false);
+
+ sstable = store.getLiveSSTables().iterator().next();
+ }
+
+ @Test
+ public void matchingSizeReturnsChannel() throws IOException
+ {
+ try (ComponentContext context = ComponentContext.create(sstable))
+ {
+ Component component = firstStreamableComponent(context);
+ long advertised = context.manifest().sizeOf(component);
+ try (FileChannel channel = context.channel(sstable.descriptor,
component, advertised))
+ {
+ assertEquals("advertised size must match the streamed channel
size", advertised, channel.size());
+ }
+ }
+ }
+
+ @Test
+ public void sizeMismatchThrowsIllegalStateExceptionNotAssertionError()
throws IOException
+ {
+ try (ComponentContext context = ComponentContext.create(sstable))
+ {
+ Component component = firstStreamableComponent(context);
+ long advertised = context.manifest().sizeOf(component);
+ long wrongSize = advertised + 1; // pretend the on-disk file no
longer matches the advertised manifest
+
+ try
+ {
+ context.channel(sstable.descriptor, component, wrongSize);
+ fail("Expected an IllegalStateException when the advertised
size does not match the on-disk size");
+ }
+ catch (AssertionError e)
+ {
+ // The whole point of the hardening: this must NOT be an
AssertionError, since assertions are
+ // disabled in production and the corrupt bytes would
otherwise be shipped silently.
+ throw new AssertionError("size mismatch must fail via
IllegalStateException, not AssertionError", e);
+ }
+ catch (IllegalStateException e)
+ {
+ assertTrue("exception message should describe the size
mismatch",
+ e.getMessage().contains("file size to be"));
+ assertTrue("exception message should include the actual
on-disk size",
+
e.getMessage().contains(String.valueOf(advertised)));
+ }
+ }
+ }
+
+ private static Component firstStreamableComponent(ComponentContext context)
+ {
+ return context.manifest().components().iterator().next();
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/db/streaming/TableStreamManagerTest.java
b/test/unit/org/apache/cassandra/db/streaming/TableStreamManagerTest.java
new file mode 100644
index 0000000000..263c95f157
--- /dev/null
+++ b/test/unit/org/apache/cassandra/db/streaming/TableStreamManagerTest.java
@@ -0,0 +1,258 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.db.streaming;
+
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.Util;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.RowUpdateBuilder;
+import org.apache.cassandra.db.compaction.CompactionManager;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.io.sstable.format.SSTableStreamRebuildState.State;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.RangesAtEndpoint;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.streaming.OutgoingStream;
+import org.apache.cassandra.streaming.PreviewKind;
+import org.apache.cassandra.streaming.StreamOperation;
+import org.apache.cassandra.streaming.StreamSession;
+import org.apache.cassandra.streaming.StreamingChannel;
+import org.apache.cassandra.streaming.async.NettyStreamingConnectionFactory;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+import static
org.apache.cassandra.service.ActiveRepairService.NO_PENDING_REPAIR;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class TableStreamManagerTest
+{
+ private static final String KEYSPACE = "TableStreamManagerTest";
+ private static final String CF = "Standard1";
+ private static final StreamingChannel.Factory connectionFactory = new
NettyStreamingConnectionFactory();
+
+ private static ColumnFamilyStore store;
+ private boolean previousStreamEntireSSTables;
+
+ @BeforeClass
+ public static void defineSchema()
+ {
+ SchemaLoader.prepareServer();
+ SchemaLoader.createKeyspace(KEYSPACE,
+ KeyspaceParams.simple(1),
+ SchemaLoader.standardCFMD(KEYSPACE, CF));
+ store = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF);
+ CompactionManager.instance.disableAutoCompaction();
+ }
+
+ @Before
+ public void setUp()
+ {
+ previousStreamEntireSSTables =
DatabaseDescriptor.streamEntireSSTables();
+ DatabaseDescriptor.setStreamEntireSSTables(true);
+ store.disableAutoCompaction();
+ store.truncateBlocking();
+ }
+
+ @After
+ public void tearDown()
+ {
+
DatabaseDescriptor.setStreamEntireSSTables(previousStreamEntireSSTables);
+ store.truncateBlocking();
+ }
+
+ private static StreamSession session()
+ {
+ try
+ {
+ return new StreamSession(StreamOperation.BOOTSTRAP,
+ InetAddressAndPort.getByName("127.0.0.1"),
+ connectionFactory,
+ null,
+ MessagingService.current_version,
+ false,
+ 0,
+ null,
+ PreviewKind.NONE);
+ }
+ catch (UnknownHostException e)
+ {
+ throw new AssertionError(e);
+ }
+ }
+
+ private RangesAtEndpoint fullRange()
+ {
+ Token min = store.getPartitioner().getMinimumToken();
+ return RangesAtEndpoint.toDummyList(Collections.singleton(new
Range<>(min, min)));
+ }
+
+ private static void insertAndFlush(int from, int to)
+ {
+ for (int j = from; j < to; j++)
+ {
+ new RowUpdateBuilder(store.metadata(), j, String.valueOf(j))
+ .clustering("0")
+ .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
+ .build()
+ .applyUnsafe();
+ }
+ Util.flush(store);
+ }
+
+ @Test
+ public void entireStreamReservesAndFinishReleasesStatus()
+ {
+ insertAndFlush(0, 10);
+ assertEquals(1, store.getLiveSSTables().size());
+
+ Collection<OutgoingStream> streams = store.getStreamManager()
+
.createOutgoingStreams(session(), fullRange(), NO_PENDING_REPAIR,
PreviewKind.NONE);
+ try
+ {
+ assertEquals(1, streams.size());
+ CassandraOutgoingFile cof =
CassandraOutgoingFile.fromStream(streams.iterator().next());
+
+ // Entire-sstable streaming must have engaged (multiple files
advertised) and reserved the ZCS status.
+ assertTrue("expected entire-sstable streaming to engage for a
full-range stream", cof.getNumFiles() > 1);
+ SSTableReader sstable = cof.getRef().get();
+ assertEquals(State.ZCS_STREAMING,
sstable.streamRebuildState().state());
+ assertEquals(1, sstable.streamRebuildState().zcsStreamCount());
+
+ // finish() releases the ZCS status exactly once and the sstable
returns to NORMAL.
+ cof.finish();
+ assertEquals(State.NORMAL, sstable.streamRebuildState().state());
+ assertEquals(0, sstable.streamRebuildState().zcsStreamCount());
+
+ // releaseStreamRebuildStatus() is idempotent - a second call must
not over-release.
+ cof.releaseStreamRebuildStatus();
+ assertEquals(State.NORMAL, sstable.streamRebuildState().state());
+ assertEquals(0, sstable.streamRebuildState().zcsStreamCount());
+ }
+ finally
+ {
+ releaseAll(streams);
+ }
+ }
+
+ @Test
+ public void planningFailurePartwayReleasesAlreadyReservedStreams()
+ {
+ // Two sstables so the manager's construction loop runs more than once.
+ insertAndFlush(0, 10);
+ insertAndFlush(10, 20);
+ assertEquals(2, store.getLiveSSTables().size());
+
+ // Spy the session so the FIRST outgoing stream is built normally
(reserving its ZCS status), then the
+ // SECOND construction fails - getStreamOperation() is only called
from inside the construction loop, once
+ // per sstable, so this deterministically injects a partway planning
failure.
+ StreamSession spy = Mockito.spy(session());
+ RuntimeException boom = new RuntimeException("injected planning
failure");
+ Mockito.doReturn(StreamOperation.BOOTSTRAP)
+ .doThrow(boom)
+ .when(spy).getStreamOperation();
+
+ try
+ {
+ store.getStreamManager().createOutgoingStreams(spy, fullRange(),
NO_PENDING_REPAIR, PreviewKind.NONE);
+ fail("Expected the injected planning failure to propagate");
+ }
+ catch (RuntimeException e)
+ {
+ assertSame(boom, e);
+ }
+
+ // The catch block must have released the ZCS status reserved by the
already-constructed first stream, so
+ // no live sstable is left stuck in ZCS_STREAMING.
+ for (SSTableReader sstable : store.getLiveSSTables())
+ {
+ assertEquals("planning failure must not leak the ZCS streaming
status for " + sstable.descriptor,
+ State.NORMAL, sstable.streamRebuildState().state());
+ assertEquals(0, sstable.streamRebuildState().zcsStreamCount());
+ }
+ }
+
+ @Test
+ public void streamingFallsBackToLegacyWhenRebuildInProgress()
+ {
+ insertAndFlush(0, 10);
+ assertEquals(1, store.getLiveSSTables().size());
+ SSTableReader sstable = store.getLiveSSTables().iterator().next();
+
+ // Model an in-progress SAI rebuild by reserving the rebuild status.
The stream must degrade to legacy and
+ // must NOT reserve (or later release) any streaming status.
+ assertTrue(sstable.streamRebuildState().tryBeginRebuild());
+
+ Collection<OutgoingStream> streams = store.getStreamManager()
+
.createOutgoingStreams(session(), fullRange(), NO_PENDING_REPAIR,
PreviewKind.NONE);
+ try
+ {
+ assertEquals(1, streams.size());
+ CassandraOutgoingFile cof =
CassandraOutgoingFile.fromStream(streams.iterator().next());
+ assertEquals("stream must fall back to legacy (single file) while
a rebuild holds the sstable",
+ 1, cof.getNumFiles());
+ // Status is still owned by the rebuild, untouched by the stream.
+ assertEquals(State.REBUILDING,
sstable.streamRebuildState().state());
+
+ cof.finish(); // must be a no-op for the rebuild status
+ assertEquals(State.REBUILDING,
sstable.streamRebuildState().state());
+ }
+ finally
+ {
+ releaseAll(streams);
+ sstable.streamRebuildState().endRebuild();
+ }
+ }
+
+ private static void releaseAll(Collection<OutgoingStream> streams)
+ {
+ List<OutgoingStream> list = new ArrayList<>(streams);
+ for (OutgoingStream stream : list)
+ {
+ CassandraOutgoingFile cof =
CassandraOutgoingFile.fromStream(stream);
+ cof.releaseStreamRebuildStatus();
+ try
+ {
+ cof.getRef().release();
+ }
+ catch (Throwable ignore)
+ {
+ // ref may already be released via finish() in the test body
+ }
+ }
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/index/sai/StreamRebuildCoordinationTest.java
b/test/unit/org/apache/cassandra/index/sai/StreamRebuildCoordinationTest.java
new file mode 100644
index 0000000000..f68c5773b3
--- /dev/null
+++
b/test/unit/org/apache/cassandra/index/sai/StreamRebuildCoordinationTest.java
@@ -0,0 +1,190 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.index.sai;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.junit.Test;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.index.SecondaryIndexBuilder;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.io.sstable.format.SSTableStreamRebuildState.State;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class StreamRebuildCoordinationTest extends SAITester
+{
+ private StorageAttachedIndex createIndexedTableWithSSTables(int
sstableCount)
+ {
+ createTable("CREATE TABLE %s (pk int PRIMARY KEY, v text)");
+ String indexName = createIndex("CREATE INDEX ON %s(v) USING 'sai'");
+ ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+ cfs.disableAutoCompaction();
+
+ int pk = 0;
+ for (int s = 0; s < sstableCount; s++)
+ {
+ for (int i = 0; i < 20; i++, pk++)
+ execute("INSERT INTO %s (pk, v) VALUES (?, ?)", pk, "v" + pk);
+ flush();
+ }
+ waitForTableIndexesQueryable();
+
+ assertEquals("expected one sstable per flush", sstableCount,
cfs.getLiveSSTables().size());
+ return (StorageAttachedIndex)
cfs.indexManager.getIndexByName(indexName);
+ }
+
+ @Test
+ public void onNotExecutedReleasesReservedRebuildStatus()
+ {
+ StorageAttachedIndex sai = createIndexedTableWithSSTables(2);
+ ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+ List<SSTableReader> sstables = new ArrayList<>(cfs.getLiveSSTables());
+
+ // Real rebuild path: reserves the per-sstable rebuild status for
every target sstable before returning.
+ SecondaryIndexBuilder builder = sai.getBuildTaskSupport()
+ .getIndexBuildTask(cfs,
Collections.singleton(sai), sstables, true);
+
+ for (SSTableReader sstable : sstables)
+ assertEquals("rebuild status must be reserved for " +
sstable.descriptor,
+ State.REBUILDING,
sstable.streamRebuildState().state());
+
+ // build() will never run (executor rejected the submission).
onNotExecuted must release everything, so the
+ // sstables do not stay stuck in REBUILDING and block future
entire-sstable streaming.
+ builder.onNotExecuted();
+
+ for (SSTableReader sstable : sstables)
+ {
+ assertEquals("onNotExecuted must release the reserved rebuild
status for " + sstable.descriptor,
+ State.NORMAL, sstable.streamRebuildState().state());
+ // And a stream may now proceed.
+ assertTrue(sstable.streamRebuildState().tryBeginStreaming());
+ sstable.streamRebuildState().endStreaming();
+ }
+ }
+
+ @Test
+ public void secondRebuildRejectedWhileFirstRebuildInFlight()
+ {
+ StorageAttachedIndex sai = createIndexedTableWithSSTables(2);
+ ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+ List<SSTableReader> sstables = new ArrayList<>(cfs.getLiveSSTables());
+
+ // First rebuild is "in flight": it reserved the per-sstable rebuild
status for every target sstable but has
+ // not run/released yet (the returned builder still owns the
reservations).
+ SecondaryIndexBuilder first = sai.getBuildTaskSupport()
+ .getIndexBuildTask(cfs,
Collections.singleton(sai), sstables, true);
+ for (SSTableReader sstable : sstables)
+ assertEquals(State.REBUILDING,
sstable.streamRebuildState().state());
+
+ try
+ {
+ // A second rebuild of the same sstables must fail fast
(tryBeginRebuild returns false because they are
+ // already REBUILDING) rather than mutating index components
underneath the in-flight rebuild.
+ assertThatThrownBy(() -> sai.getBuildTaskSupport()
+ .getIndexBuildTask(cfs,
Collections.singleton(sai), sstables, true))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Cannot build SAI index");
+
+ // The failed second attempt must not disturb the first rebuild's
reservations: every sstable is still
+ // REBUILDING and owned by the first builder (the second attempt
rolled back only what it reserved).
+ for (SSTableReader sstable : sstables)
+ {
+ assertEquals("first rebuild's reservation must be intact for "
+ sstable.descriptor,
+ State.REBUILDING,
sstable.streamRebuildState().state());
+ assertFalse("no third rebuild may begin while the first is in
flight",
+ sstable.streamRebuildState().tryBeginRebuild());
+ assertFalse("streaming must remain blocked while the first
rebuild is in flight",
+ sstable.streamRebuildState().tryBeginStreaming());
+ }
+ }
+ finally
+ {
+ first.onNotExecuted();
+ }
+
+ // Once the first rebuild releases, a fresh rebuild can reserve
everything cleanly again.
+ for (SSTableReader sstable : sstables)
+ assertEquals(State.NORMAL, sstable.streamRebuildState().state());
+ SecondaryIndexBuilder third = sai.getBuildTaskSupport()
+ .getIndexBuildTask(cfs,
Collections.singleton(sai), sstables, true);
+ for (SSTableReader sstable : sstables)
+ assertEquals(State.REBUILDING,
sstable.streamRebuildState().state());
+ third.onNotExecuted();
+ }
+
+ @Test
+ public void partialReservationRollbackWhenAnSSTableIsBeingStreamed()
+ {
+ StorageAttachedIndex sai = createIndexedTableWithSSTables(3);
+ ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+
+ List<SSTableReader> live = new ArrayList<>(cfs.getLiveSSTables());
+ assertEquals(3, live.size());
+
+ // Simulate an in-flight entire-sstable stream of ONE sstable, and
make sure it is reserved LAST so the
+ // rebuild reserves the other two first and must then roll them back
when it hits the streamed one.
+ SSTableReader streamed = live.get(0);
+ List<SSTableReader> rollbackCandidates = new
ArrayList<>(live.subList(1, live.size()));
+ assertThat(rollbackCandidates).hasSize(2);
+ assertTrue(streamed.streamRebuildState().tryBeginStreaming());
+
+ List<SSTableReader> ordered = new ArrayList<>(rollbackCandidates);
+ ordered.add(streamed); // streamed sstable reserved last
+
+ try
+ {
+ assertThatThrownBy(() -> sai.getBuildTaskSupport()
+ .getIndexBuildTask(cfs,
Collections.singleton(sai), ordered, true))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Cannot build SAI index while entire
SSTable streaming is in progress");
+
+ // The two sstables reserved before the failure must be fully
rolled back to NORMAL...
+ for (SSTableReader sstable : rollbackCandidates)
+ assertEquals("reservation must be rolled back for " +
sstable.descriptor,
+ State.NORMAL,
sstable.streamRebuildState().state());
+
+ // ...and the streamed sstable must be untouched (still
streaming), not corrupted into REBUILDING.
+ assertEquals(State.ZCS_STREAMING,
streamed.streamRebuildState().state());
+ assertEquals(1, streamed.streamRebuildState().zcsStreamCount());
+ }
+ finally
+ {
+ streamed.streamRebuildState().endStreaming();
+ }
+
+ // With the stream released, a rebuild can now reserve every sstable
cleanly - proving the earlier partial
+ // rollback left no lingering reservations.
+ List<SSTableReader> all = new ArrayList<>(cfs.getLiveSSTables());
+ SecondaryIndexBuilder builder = sai.getBuildTaskSupport()
+ .getIndexBuildTask(cfs,
Collections.singleton(sai), all, true);
+ for (SSTableReader sstable : all)
+ assertEquals(State.REBUILDING,
sstable.streamRebuildState().state());
+ builder.onNotExecuted();
+ for (SSTableReader sstable : all)
+ assertEquals(State.NORMAL, sstable.streamRebuildState().state());
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/io/sstable/format/SSTableStreamRebuildStateTest.java
b/test/unit/org/apache/cassandra/io/sstable/format/SSTableStreamRebuildStateTest.java
new file mode 100644
index 0000000000..9717bffcf3
--- /dev/null
+++
b/test/unit/org/apache/cassandra/io/sstable/format/SSTableStreamRebuildStateTest.java
@@ -0,0 +1,285 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.io.sstable.format;
+
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.Test;
+
+import org.apache.cassandra.io.sstable.format.SSTableStreamRebuildState.State;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+public class SSTableStreamRebuildStateTest
+{
+ @Test
+ public void startsNormal()
+ {
+ SSTableStreamRebuildState s = new SSTableStreamRebuildState();
+ assertEquals(State.NORMAL, s.state());
+ assertEquals(0, s.zcsStreamCount());
+ }
+
+ @Test
+ public void multipleStreamsAllowedAndCounted()
+ {
+ SSTableStreamRebuildState s = new SSTableStreamRebuildState();
+ assertTrue(s.tryBeginStreaming());
+ assertTrue(s.tryBeginStreaming());
+ assertEquals(State.ZCS_STREAMING, s.state());
+ assertEquals(2, s.zcsStreamCount());
+
+ s.endStreaming();
+ assertEquals(State.ZCS_STREAMING, s.state());
+ assertEquals(1, s.zcsStreamCount());
+
+ s.endStreaming();
+ assertEquals(State.NORMAL, s.state());
+ assertEquals(0, s.zcsStreamCount());
+ }
+
+ @Test
+ public void rebuildBlockedWhileStreaming()
+ {
+ SSTableStreamRebuildState s = new SSTableStreamRebuildState();
+ assertTrue(s.tryBeginStreaming());
+ assertFalse(s.tryBeginRebuild());
+ assertEquals(State.ZCS_STREAMING, s.state());
+ }
+
+ @Test
+ public void streamingBlockedWhileRebuilding()
+ {
+ SSTableStreamRebuildState s = new SSTableStreamRebuildState();
+ assertTrue(s.tryBeginRebuild());
+ assertFalse(s.tryBeginStreaming());
+ assertEquals(State.REBUILDING, s.state());
+ assertEquals(0, s.zcsStreamCount());
+ }
+
+ @Test
+ public void rebuildExcludesSecondRebuild()
+ {
+ SSTableStreamRebuildState s = new SSTableStreamRebuildState();
+ assertTrue(s.tryBeginRebuild());
+ assertFalse(s.tryBeginRebuild());
+ }
+
+ @Test
+ public void rebuildResetsToNormal()
+ {
+ SSTableStreamRebuildState s = new SSTableStreamRebuildState();
+ assertTrue(s.tryBeginRebuild());
+ s.endRebuild();
+ assertEquals(State.NORMAL, s.state());
+ assertTrue(s.tryBeginStreaming());
+ }
+
+ @Test
+ public void endIsDefensiveAgainstOverRelease()
+ {
+ SSTableStreamRebuildState s = new SSTableStreamRebuildState();
+ s.endStreaming(); // no-op, must not go negative
+ s.endRebuild(); // no-op
+ assertEquals(State.NORMAL, s.state());
+ assertEquals(0, s.zcsStreamCount());
+ }
+
+ /**
+ * Under contention many threads race to begin a rebuild of the same
sstable. Because a rebuild is exclusive,
+ * exactly one attempt must win; every other attempt must observe the
in-flight rebuild and fail. Covers the
+ * "second concurrent rebuild attempt fails while one is in flight"
scenario under real contention rather than
+ * the sequential check in {@link #rebuildExcludesSecondRebuild()}.
+ */
+ @Test
+ public void concurrentRebuildAttemptsOnlyOneWins() throws Exception
+ {
+ for (int round = 0; round < 200; round++)
+ {
+ SSTableStreamRebuildState s = new SSTableStreamRebuildState();
+ int threads = 8;
+ CyclicBarrier start = new CyclicBarrier(threads);
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ try
+ {
+ AtomicInteger winners = new AtomicInteger();
+ Future<?>[] futures = new Future<?>[threads];
+ for (int i = 0; i < threads; i++)
+ {
+ futures[i] = pool.submit(() -> {
+ awaitBarrier(start);
+ if (s.tryBeginRebuild())
+ winners.incrementAndGet();
+ });
+ }
+ for (Future<?> f : futures)
+ f.get(30, TimeUnit.SECONDS);
+
+ assertEquals("exactly one rebuild attempt must win the race",
1, winners.get());
+ assertEquals(State.REBUILDING, s.state());
+ assertFalse("streaming must be blocked while the winning
rebuild is in flight", s.tryBeginStreaming());
+
+ s.endRebuild();
+ assertEquals(State.NORMAL, s.state());
+ }
+ finally
+ {
+ pool.shutdownNow();
+ }
+ }
+ }
+
+ /**
+ * Concurrent entire-sstable streams of the same sstable must all be
admitted and reference counted correctly,
+ * so that no update is lost under contention (the count must equal the
number of successful begins, and only
+ * the final release returns the sstable to {@code NORMAL}).
+ */
+ @Test
+ public void concurrentStreamsAreRefCountedCorrectly() throws Exception
+ {
+ for (int round = 0; round < 200; round++)
+ {
+ SSTableStreamRebuildState s = new SSTableStreamRebuildState();
+ int threads = 8;
+ CyclicBarrier start = new CyclicBarrier(threads);
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ try
+ {
+ AtomicInteger admitted = new AtomicInteger();
+ Future<?>[] futures = new Future<?>[threads];
+ for (int i = 0; i < threads; i++)
+ {
+ futures[i] = pool.submit(() -> {
+ awaitBarrier(start);
+ if (s.tryBeginStreaming())
+ admitted.incrementAndGet();
+ });
+ }
+ for (Future<?> f : futures)
+ f.get(30, TimeUnit.SECONDS);
+
+ assertEquals("all concurrent streams must be admitted
(streaming is not exclusive)", threads, admitted.get());
+ assertEquals(State.ZCS_STREAMING, s.state());
+ assertEquals(threads, s.zcsStreamCount());
+ assertFalse("a rebuild must be blocked while any stream is in
flight", s.tryBeginRebuild());
+
+ for (int i = 0; i < threads; i++)
+ s.endStreaming();
+
+ assertEquals(State.NORMAL, s.state());
+ assertEquals(0, s.zcsStreamCount());
+ }
+ finally
+ {
+ pool.shutdownNow();
+ }
+ }
+ }
+
+ /**
+ * Stress test hammering {@code tryBeginRebuild}/{@code tryBeginStreaming}
from many threads to assert the core
+ * mutual-exclusion invariant never breaks under contention: a rebuild and
any stream are never both active at
+ * once, and at most one rebuild is ever active. This complements the
sequential state-transition tests, which
+ * cannot detect races in the check-and-set logic.
+ */
+ @Test
+ public void mutualExclusionHoldsUnderStress() throws Exception
+ {
+ SSTableStreamRebuildState s = new SSTableStreamRebuildState();
+ int threads = 12;
+ int iterationsPerThread = 20_000;
+
+ AtomicInteger activeRebuilds = new AtomicInteger();
+ AtomicInteger activeStreams = new AtomicInteger();
+ AtomicReference<String> violation = new AtomicReference<>();
+
+ CyclicBarrier start = new CyclicBarrier(threads);
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ try
+ {
+ Future<?>[] futures = new Future<?>[threads];
+ for (int t = 0; t < threads; t++)
+ {
+ final int seed = t;
+ futures[t] = pool.submit(() -> {
+ awaitBarrier(start);
+ java.util.Random rnd = new java.util.Random(seed);
+ for (int i = 0; i < iterationsPerThread && violation.get()
== null; i++)
+ {
+ if (rnd.nextBoolean())
+ {
+ if (s.tryBeginRebuild())
+ {
+ int r = activeRebuilds.incrementAndGet();
+ if (r != 1)
+ violation.compareAndSet(null, "two
rebuilds active simultaneously: " + r);
+ if (activeStreams.get() != 0)
+ violation.compareAndSet(null, "stream
active during rebuild");
+ activeRebuilds.decrementAndGet();
+ s.endRebuild();
+ }
+ }
+ else
+ {
+ if (s.tryBeginStreaming())
+ {
+ activeStreams.incrementAndGet();
+ if (activeRebuilds.get() != 0)
+ violation.compareAndSet(null, "rebuild
active during stream");
+ activeStreams.decrementAndGet();
+ s.endStreaming();
+ }
+ }
+ }
+ });
+ }
+ for (Future<?> f : futures)
+ f.get(60, TimeUnit.SECONDS);
+
+ assertNull(violation.get(), violation.get());
+ assertEquals("all operations released, must return to NORMAL",
State.NORMAL, s.state());
+ assertEquals(0, s.zcsStreamCount());
+ }
+ finally
+ {
+ pool.shutdownNow();
+ }
+ }
+
+ private static void awaitBarrier(CyclicBarrier barrier)
+ {
+ try
+ {
+ barrier.await(30, TimeUnit.SECONDS);
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
