natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

CEP-15: (C*) Enhance in-memory FileSystem to work with mmap and support tests to add custom logic 

patch by David Capwell; reviewed by Caleb Rackliffe for CASSANDRA-18485
This commit is contained in:
David Capwell 2023-05-08 11:56:37 -07:00
parent af07ed334a
commit 4759a7353f
39 changed files with 1931 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/java/org/apache/cassandra/hints/HintsCatalog.java
View File

@ -168,6 +168,8 @@ final class HintsCatalog
}
else
{
if (SyncUtil.SKIP_SYNC)
return;
logger.error("Unable to open directory {}", hintsDirectory.absolutePath());
FileUtils.handleFSErrorAndPropagate(new FSWriteError(new IOException(String.format("Unable to open hint directory %s", hintsDirectory.absolutePath())), hintsDirectory.absolutePath()));
}

19
src/java/org/apache/cassandra/io/util/File.java
View File

@ -38,6 +38,7 @@ import java.util.function.Predicate;
import java.util.stream.Stream;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.RateLimiter;
import net.openhft.chronicle.core.util.ThrowingFunction;
@ -125,6 +126,18 @@ public class File implements Comparable<File>
if (!path.isAbsolute() || path.isOpaque()) throw new IllegalArgumentException();
}
/**
* Unsafe constructor that allows a File to use a differet {@link FileSystem} than {@link File#filesystem}.
*
* The main caller of such a method are cases such as JVM Dtest functions that need access to the logging framwork
* files, which exists on in {@link FileSystems#getDefault()}.
*/
@VisibleForTesting
public File(FileSystem fs, String first, String... more)
{
this.path = fs.getPath(first, more);
}
/**
* @param path the path to wrap
*/
@ -777,6 +790,12 @@ public class File implements Comparable<File>
return path;
}
@VisibleForTesting
public static FileSystem unsafeGetFilesystem()
{
return filesystem;
}
public static void unsafeSetFilesystem(FileSystem fs)
{
filesystem = fs;

12
src/java/org/apache/cassandra/utils/ByteBufferUtil.java
View File

@ -26,6 +26,7 @@ package org.apache.cassandra.utils;
import java.io.*;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
@ -866,4 +867,15 @@ public class ByteBufferUtil
return true;
}
public static void readFully(FileChannel channel, ByteBuffer dst, long position) throws IOException
{
while (dst.hasRemaining())
{
int read = channel.read(dst, position);
if (read == -1)
throw new EOFException();
position += read;
}
}
}

8
src/java/org/apache/cassandra/utils/SyncUtil.java
View File

@ -31,6 +31,8 @@ import org.apache.cassandra.config.Config;
import com.google.common.base.Preconditions;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.utils.memory.MemoryUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -99,6 +101,12 @@ public class SyncUtil
public static MappedByteBuffer force(MappedByteBuffer buf)
{
Preconditions.checkNotNull(buf);
Object attachment = MemoryUtil.getAttachment(buf);
if (attachment instanceof Runnable)
{
((Runnable) attachment).run();
return buf;
}
if (SKIP_SYNC)
{
Object fd = null;

5
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -25,6 +25,7 @@ import java.net.BindException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.file.FileSystem;
import java.nio.file.FileSystems;
import java.security.Permission;
import java.util.ArrayList;
import java.util.Collections;
@ -221,10 +222,10 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
String suite = System.getProperty("suitename", "suitename_IS_UNDEFINED");
String clusterId = ClusterIDDefiner.getId();
String instanceId = InstanceIDDefiner.getInstanceId();
File f = new File(String.format("build/test/logs/%s/%s/%s/%s/system.log", tag, suite, clusterId, instanceId));
File f = new File(FileSystems.getDefault(), String.format("build/test/logs/%s/%s/%s/%s/system.log", tag, suite, clusterId, instanceId));
// when creating a cluster globally in a test class we get the logs without the suite, try finding those logs:
if (!f.exists())
f = new File(String.format("build/test/logs/%s/%s/%s/system.log", tag, clusterId, instanceId));
f = new File(FileSystems.getDefault(), String.format("build/test/logs/%s/%s/%s/system.log", tag, clusterId, instanceId));
if (!f.exists())
throw new AssertionError("Unable to locate system.log under " + new File("build/test/logs").absolutePath() + "; make sure ICluster.setup() is called or extend TestBaseImpl and do not define a static beforeClass function with @BeforeClass");
return new FileLogAction(f);

47
test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java
View File

@ -21,8 +21,8 @@ package org.apache.cassandra.simulator;
import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.nio.file.FileSystem;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.EnumMap;
import java.util.List;
import java.util.Map;
@ -35,8 +35,6 @@ import java.util.function.LongConsumer;
import java.util.function.Predicate;
import java.util.function.Supplier;
import com.google.common.jimfs.Configuration;
import com.google.common.jimfs.Jimfs;
import com.google.common.util.concurrent.AsyncFunction;
import com.google.common.util.concurrent.FutureCallback;
@ -55,6 +53,7 @@ import org.apache.cassandra.distributed.api.IIsolatedExecutor.SerializableRunnab
import org.apache.cassandra.distributed.impl.DirectStreamingConnectionFactory;
import org.apache.cassandra.distributed.impl.IsolatedExecutor;
import org.apache.cassandra.io.compress.LZ4Compressor;
import org.apache.cassandra.io.util.FileSystems;
import org.apache.cassandra.service.paxos.BallotGenerator;
import org.apache.cassandra.service.paxos.PaxosPrepare;
import org.apache.cassandra.simulator.RandomSource.Choices;
@ -62,6 +61,7 @@ import org.apache.cassandra.simulator.asm.InterceptAsClassTransformer;
import org.apache.cassandra.simulator.asm.NemesisFieldSelectors;
import org.apache.cassandra.simulator.cluster.ClusterActions;
import org.apache.cassandra.simulator.cluster.ClusterActions.TopologyChange;
import org.apache.cassandra.io.filesystem.ListenableFileSystem;
import org.apache.cassandra.simulator.systems.Failures;
import org.apache.cassandra.simulator.systems.InterceptedWait.CaptureSites.Capture;
import org.apache.cassandra.simulator.systems.InterceptibleThread;
@ -607,7 +607,7 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
public final SimulatedSystems simulated;
public final Cluster cluster;
public final S simulation;
private final FileSystem jimfs;
private final ListenableFileSystem fs;
protected final Map<Integer, List<Closeable>> onUnexpectedShutdown = new TreeMap<>();
protected final List<Callable<Void>> onShutdown = new CopyOnWriteArrayList<>();
protected final ThreadLocalRandomCheck threadLocalRandomCheck;
@ -618,9 +618,7 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
SimulationFactory<S> factory) throws IOException
{
this.random = random;
this.jimfs = Jimfs.newFileSystem(Long.toHexString(seed) + 'x' + uniqueNum, Configuration.unix().toBuilder()
.setMaxSize(4L << 30).setBlockSize(512)
.build());
this.fs = new ListenableFileSystem(FileSystems.jimfs(Long.toHexString(seed) + 'x' + uniqueNum));
final SimulatedMessageDelivery delivery;
final SimulatedExecution execution;
@ -670,23 +668,28 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
Failures failures = new Failures();
ThreadAllocator threadAllocator = new ThreadAllocator(random, builder.threadCount, numOfNodes);
List<String> allowedDiskAccessModes = Arrays.asList("mmap", "mmap_index_only", "standard");
String disk_access_mode = allowedDiskAccessModes.get(random.uniform(0, allowedDiskAccessModes.size() - 1));
boolean commitlogCompressed = random.decide(.5f);
cluster = snitch.setup(Cluster.build(numOfNodes)
.withRoot(jimfs.getPath("/cassandra"))
.withRoot(fs.getPath("/cassandra"))
.withSharedClasses(sharedClassPredicate)
.withConfig(config -> configUpdater.accept(threadAllocator.update(config
.with(Feature.BLANK_GOSSIP)
.set("read_request_timeout", String.format("%dms", NANOSECONDS.toMillis(builder.readTimeoutNanos)))
.set("write_request_timeout", String.format("%dms", NANOSECONDS.toMillis(builder.writeTimeoutNanos)))
.set("cas_contention_timeout", String.format("%dms", NANOSECONDS.toMillis(builder.contentionTimeoutNanos)))
.set("request_timeout", String.format("%dms", NANOSECONDS.toMillis(builder.requestTimeoutNanos)))
.set("memtable_heap_space", "1MiB")
.set("memtable_allocation_type", builder.memoryListener != null ? "unslabbed_heap_buffers_logged" : "heap_buffers")
.set("file_cache_size", "16MiB")
.set("use_deterministic_table_id", true)
.set("disk_access_mode", "standard")
.set("failure_detector", SimulatedFailureDetector.Instance.class.getName())
.set("commitlog_compression", new ParameterizedClass(LZ4Compressor.class.getName(), emptyMap()))
)))
.withConfig(config -> {
config.with(Feature.BLANK_GOSSIP)
.set("read_request_timeout", String.format("%dms", NANOSECONDS.toMillis(builder.readTimeoutNanos)))
.set("write_request_timeout", String.format("%dms", NANOSECONDS.toMillis(builder.writeTimeoutNanos)))
.set("cas_contention_timeout", String.format("%dms", NANOSECONDS.toMillis(builder.contentionTimeoutNanos)))
.set("request_timeout", String.format("%dms", NANOSECONDS.toMillis(builder.requestTimeoutNanos)))
.set("memtable_heap_space", "1MiB")
.set("memtable_allocation_type", builder.memoryListener != null ? "unslabbed_heap_buffers_logged" : "heap_buffers")
.set("file_cache_size", "16MiB")
.set("use_deterministic_table_id", true)
.set("disk_access_mode", disk_access_mode)
.set("failure_detector", SimulatedFailureDetector.Instance.class.getName());
if (commitlogCompressed)
config.set("commitlog_compression", new ParameterizedClass(LZ4Compressor.class.getName(), emptyMap()));
configUpdater.accept(threadAllocator.update(config));
})
.withInstanceInitializer(new IInstanceInitializer()
{
@Override

3
test/unit/org/apache/cassandra/auth/CreateAndAlterRoleTest.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.auth;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
@ -32,6 +33,8 @@ public class CreateAndAlterRoleTest extends CQLTester
@BeforeClass
public static void setUpClass()
{
ServerTestUtils.daemonInitialization();
DatabaseDescriptor.setPermissionsValidity(0);
DatabaseDescriptor.setRolesValidity(0);
DatabaseDescriptor.setCredentialsValidity(0);

2
test/unit/org/apache/cassandra/auth/GrantAndRevokeTest.java
View File

@ -22,6 +22,7 @@ import org.junit.BeforeClass;
import org.junit.Test;
import com.datastax.driver.core.ResultSet;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
@ -36,6 +37,7 @@ public class GrantAndRevokeTest extends CQLTester
@BeforeClass
public static void setUpClass()
{
ServerTestUtils.daemonInitialization();
DatabaseDescriptor.setPermissionsValidity(0);
CQLTester.setUpClass();
requireAuthentication();

2
test/unit/org/apache/cassandra/cql3/CDCStatementTest.java
View File

@ -22,6 +22,7 @@ import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
public class CDCStatementTest extends CQLTester
@ -29,6 +30,7 @@ public class CDCStatementTest extends CQLTester
@BeforeClass
public static void setUpClass()
{
ServerTestUtils.daemonInitialization();
DatabaseDescriptor.setCDCEnabled(true);
CQLTester.setUpClass();
}

52
test/unit/org/apache/cassandra/cql3/CQLTester.java
View File

@ -69,13 +69,17 @@ import org.apache.cassandra.auth.AuthTestUtils;
import org.apache.cassandra.auth.IRoleManager;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DataStorageSpec;
import org.apache.cassandra.config.EncryptionOptions;
import org.apache.cassandra.db.virtual.VirtualKeyspaceRegistry;
import org.apache.cassandra.db.virtual.VirtualSchemaKeyspace;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.index.SecondaryIndexManager;
import org.apache.cassandra.io.filesystem.ListenableFileSystem;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileSystems;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.metrics.CassandraMetricsRegistry;
@ -191,8 +195,6 @@ public abstract class CQLTester
nativeAddr = InetAddress.getLoopbackAddress();
nativePort = getAutomaticallyAllocatedPort(nativeAddr);
ServerTestUtils.daemonInitialization();
}
private List<String> keyspaces = new ArrayList<>();
@ -318,6 +320,8 @@ public abstract class CQLTester
@BeforeClass
public static void setUpClass()
{
ServerTestUtils.daemonInitialization();
if (ROW_CACHE_SIZE_IN_MIB > 0)
DatabaseDescriptor.setRowCacheSizeInMiB(ROW_CACHE_SIZE_IN_MIB);
StorageService.instance.setPartitionerUnsafe(Murmur3Partitioner.instance);
@ -1347,7 +1351,7 @@ public abstract class CQLTester
return QueryProcessor.instance.prepare(formatQuery(query), ClientState.forInternalCalls());
}
protected UntypedResultSet execute(String query, Object... values) throws Throwable
protected UntypedResultSet execute(String query, Object... values)
{
return executeFormattedQuery(formatQuery(query), values);
}
@ -1361,7 +1365,7 @@ public abstract class CQLTester
* Executes the provided query using the {@link ClientState#forInternalCalls()} as the expected ClientState. Note:
* this means permissions checking will not apply and queries will proceed regardless of role or guardrails.
*/
protected UntypedResultSet executeFormattedQuery(String query, Object... values) throws Throwable
protected UntypedResultSet executeFormattedQuery(String query, Object... values)
{
UntypedResultSet rs;
if (usePrepared)
@ -2442,4 +2446,44 @@ public abstract class CQLTester
&& Objects.equal(password, u.password);
}
}
public static abstract class InMemory extends CQLTester
{
protected static ListenableFileSystem fs = null;
/**
* Used by {@link #cleanupFileSystemListeners()} to know if file system listeners should be removed at the start
* of a test; can disable for cases where listeners are needed cross mutliple tests.
*/
protected boolean cleanupFileSystemListeners = true;
@BeforeClass
public static void setUpClass()
{
fs = FileSystems.newGlobalInMemoryFileSystem();
CassandraRelevantProperties.IGNORE_MISSING_NATIVE_FILE_HINTS.setBoolean(true);
FileSystems.maybeCreateTmp();
CQLTester.setUpClass();
}
@Before
public void cleanupFileSystemListeners()
{
if (!cleanupFileSystemListeners)
return;
fs.clearListeners();
}
protected ListenableFileSystem.PathFilter isCurrentTableIndexFile(String keyspace)
{
return path -> {
if (!path.getFileName().toString().endsWith("Index.db"))
return false;
Descriptor desc = Descriptor.fromFile(new File(path));
if (!desc.ksname.equals(keyspace) && desc.cfname.equals(currentTable()))
return false;
return true;
};
}
}
}

3
test/unit/org/apache/cassandra/cql3/CustomNowInSecondsTest.java
View File

@ -25,6 +25,7 @@ import com.google.common.collect.ImmutableList;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.cql3.statements.BatchStatement;
import org.apache.cassandra.cql3.statements.ModificationStatement;
import org.apache.cassandra.db.ConsistencyLevel;
@ -43,6 +44,8 @@ public class CustomNowInSecondsTest extends CQLTester
@BeforeClass
public static void setUpClass()
{
ServerTestUtils.daemonInitialization();
prepareServer();
requireNetwork();
}

2
test/unit/org/apache/cassandra/cql3/GcCompactionTest.java
View File

@ -55,7 +55,7 @@ public class GcCompactionTest extends CQLTester
}
@Override
protected UntypedResultSet execute(String query, Object... values) throws Throwable
protected UntypedResultSet execute(String query, Object... values)
{
return executeFormattedQuery(formatQuery(KEYSPACE_PER_TEST, query), values);
}

2
test/unit/org/apache/cassandra/cql3/KeyCacheCqlTest.java
View File

@ -120,7 +120,7 @@ public class KeyCacheCqlTest extends CQLTester
}
@Override
protected UntypedResultSet execute(String query, Object... values) throws Throwable
protected UntypedResultSet execute(String query, Object... values)
{
return executeFormattedQuery(formatQuery(KEYSPACE_PER_TEST, query), values);
}

3
test/unit/org/apache/cassandra/cql3/selection/SelectionColumnMappingTest.java
View File

@ -26,6 +26,7 @@ import java.util.List;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.schema.Schema;
@ -54,6 +55,8 @@ public class SelectionColumnMappingTest extends CQLTester
@BeforeClass
public static void setUpClass() // overrides CQLTester.setUpClass()
{
ServerTestUtils.daemonInitialization();
DatabaseDescriptor.setPartitionerUnsafe(ByteOrderedPartitioner.instance);
prepareServer();

2
test/unit/org/apache/cassandra/cql3/validation/entities/FrozenCollectionsTest.java
View File

@ -26,6 +26,7 @@ import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.restrictions.StatementRestrictions;
import org.apache.cassandra.db.marshal.*;
@ -43,6 +44,7 @@ public class FrozenCollectionsTest extends CQLTester
@BeforeClass
public static void setUpClass() // overrides CQLTester.setUpClass()
{
ServerTestUtils.daemonInitialization();
// Selecting partitioner for a table is not exposed on CREATE TABLE.
StorageService.instance.setPartitionerUnsafe(ByteOrderedPartitioner.instance);

2
test/unit/org/apache/cassandra/cql3/validation/entities/JsonTest.java
View File

@ -17,6 +17,7 @@
*/
package org.apache.cassandra.cql3.validation.entities;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.Duration;
@ -50,6 +51,7 @@ public class JsonTest extends CQLTester
@BeforeClass
public static void setUpClass()
{
ServerTestUtils.daemonInitialization();
if (ROW_CACHE_SIZE_IN_MIB > 0)
DatabaseDescriptor.setRowCacheSizeInMiB(ROW_CACHE_SIZE_IN_MIB);

3
test/unit/org/apache/cassandra/cql3/validation/entities/UserTypesTest.java
View File

@ -22,6 +22,7 @@ import java.util.UUID;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.dht.ByteOrderedPartitioner;
import org.apache.cassandra.exceptions.InvalidRequestException;
@ -32,6 +33,8 @@ public class UserTypesTest extends CQLTester
@BeforeClass
public static void setUpClass() // overrides CQLTester.setUpClass()
{
ServerTestUtils.daemonInitialization();
// Selecting partitioner for a table is not exposed on CREATE TABLE.
StorageService.instance.setPartitionerUnsafe(ByteOrderedPartitioner.instance);

3
test/unit/org/apache/cassandra/cql3/validation/entities/VirtualTableTest.java
View File

@ -39,6 +39,7 @@ import org.junit.BeforeClass;
import org.junit.Test;
import com.datastax.driver.core.exceptions.InvalidQueryException;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.marshal.Int32Type;
@ -208,6 +209,8 @@ public class VirtualTableTest extends CQLTester
@BeforeClass
public static void setUpClass()
{
ServerTestUtils.daemonInitialization();
TableMetadata vt1Metadata = TableMetadata.builder(KS_NAME, VT1_NAME)
.kind(TableMetadata.Kind.VIRTUAL)
.addPartitionKeyColumn("pk", UTF8Type.instance)

2
test/unit/org/apache/cassandra/cql3/validation/miscellaneous/SSTablesIteratedTest.java
View File

@ -79,7 +79,7 @@ public class SSTablesIteratedTest extends CQLTester
}
@Override
protected UntypedResultSet execute(String query, Object... values) throws Throwable
protected UntypedResultSet execute(String query, Object... values)
{
return executeFormattedQuery(formatQuery(KEYSPACE_PER_TEST, query), values);
}

3
test/unit/org/apache/cassandra/cql3/validation/operations/InsertUpdateIfConditionCollectionsTest.java
View File

@ -29,6 +29,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.SyntaxException;
@ -51,6 +52,8 @@ public class InsertUpdateIfConditionCollectionsTest extends CQLTester
@Parameterized.Parameters(name = "{index}: clusterMinVersion={0}")
public static Collection<Object[]> data()
{
ServerTestUtils.daemonInitialization();
return InsertUpdateIfConditionTest.data();
}

2
test/unit/org/apache/cassandra/cql3/validation/operations/InsertUpdateIfConditionStaticsTest.java
View File

@ -27,6 +27,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.cql3.CQLTester;
/* InsertUpdateIfConditionCollectionsTest class has been split into multiple ones because of timeout issues (CASSANDRA-16670)
@ -47,6 +48,7 @@ public class InsertUpdateIfConditionStaticsTest extends CQLTester
@Parameterized.Parameters(name = "{index}: clusterMinVersion={0}")
public static Collection<Object[]> data()
{
ServerTestUtils.daemonInitialization();
return InsertUpdateIfConditionTest.data();
}

2
test/unit/org/apache/cassandra/cql3/validation/operations/InsertUpdateIfConditionTest.java
View File

@ -28,6 +28,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.Util;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.Duration;
@ -61,6 +62,7 @@ public class InsertUpdateIfConditionTest extends CQLTester
@Parameterized.Parameters(name = "{index}: clusterMinVersion={0}")
public static Collection<Object[]> data()
{
ServerTestUtils.daemonInitialization();
return Arrays.asList(new Object[]{ "3.0", (Runnable) () -> {
Pair<Boolean, CassandraVersion> res = Gossiper.instance.isUpgradingFromVersionLowerThanC17653(new CassandraVersion("3.11"));
assertTrue(debugMsgCASSANDRA17653(res), res.left);

3
test/unit/org/apache/cassandra/cql3/validation/operations/SelectLimitTest.java
View File

@ -23,6 +23,7 @@ package org.apache.cassandra.cql3.validation.operations;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.dht.ByteOrderedPartitioner;
@ -34,6 +35,8 @@ public class SelectLimitTest extends CQLTester
@BeforeClass
public static void setUpClass()
{
ServerTestUtils.daemonInitialization();
StorageService.instance.setPartitionerUnsafe(ByteOrderedPartitioner.instance);
DatabaseDescriptor.setPartitionerUnsafe(ByteOrderedPartitioner.instance);

56
test/unit/org/apache/cassandra/db/CorruptPrimaryIndexTest.java Normal file
View File

@ -0,0 +1,56 @@
/*
* 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;
import org.junit.Test;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.UntypedResultSet;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
public class CorruptPrimaryIndexTest extends CQLTester.InMemory
{
@Test
public void bigPrimaryIndexDoesNotDetectDiskCorruption()
{
// Set listener early, before the file is opened; mmap access can not be listened to, so need to observe the open, which happens on flush
fs.onPostRead(isCurrentTableIndexFile(keyspace()), (path, channel, position, dst, read) -> {
// Reading the Primary index for the test!
// format
// 2 bytes: length of bytes for PK
// 4 bytes: pk as an int32
// variable bytes (see org.apache.cassandra.io.sstable.format.big.RowIndexEntry.IndexSerializer.deserialize(org.apache.cassandra.io.util.FileDataInput))
assertThat(position).describedAs("Unexpected access, should start read from start of file").isEqualTo(0);
// simulate bit rot by having 1 byte change... but make sure it's the pk!
dst.put(2, Byte.MAX_VALUE);
});
createTable("CREATE TABLE %s (id int PRIMARY KEY, value int)");
execute("INSERT INTO %s (id, value) VALUES (?, ?)", 0, 0);
flush();
UntypedResultSet rs = execute("SELECT * FROM %s WHERE id=?", 0);
// this assert check is here to get the test to be green... if the format is fixed and this data loss is not
// happening anymore, then this check should be updated
assertThatThrownBy(() -> assertRows(rs, row(0, 0))).hasMessage("Got less rows than expected. Expected 1 but got 0");
}
}

2
test/unit/org/apache/cassandra/db/KeyspaceTest.java
View File

@ -61,7 +61,7 @@ public class KeyspaceTest extends CQLTester
}
@Override
protected UntypedResultSet execute(String query, Object... values) throws Throwable
protected UntypedResultSet execute(String query, Object... values)
{
return executeFormattedQuery(formatQuery(KEYSPACE_PER_TEST, query), values);
}

3
test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDCTest.java
View File

@ -27,6 +27,7 @@ import java.util.*;
import java.util.concurrent.TimeUnit;
import com.google.monitoring.runtime.instrumentation.common.util.concurrent.Uninterruptibles;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileReader;
import org.junit.Assert;
@ -51,6 +52,8 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
@BeforeClass
public static void setUpClass()
{
ServerTestUtils.daemonInitialization();
DatabaseDescriptor.setCDCEnabled(true);
DatabaseDescriptor.setCDCTotalSpaceInMiB(1024);
CQLTester.setUpClass();

2
test/unit/org/apache/cassandra/db/memtable/MemtableSizeTestBase.java
View File

@ -30,6 +30,7 @@ import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
@ -76,6 +77,7 @@ public abstract class MemtableSizeTestBase extends CQLTester
public static void setup(Config.MemtableAllocationType allocationType)
{
ServerTestUtils.daemonInitialization();
try
{
Field confField = DatabaseDescriptor.class.getDeclaredField("conf");

3
test/unit/org/apache/cassandra/db/virtual/CredentialsCacheKeysTableTest.java
View File

@ -26,6 +26,7 @@ import org.junit.Test;
import com.datastax.driver.core.EndPoint;
import com.datastax.driver.core.PlainTextAuthProvider;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.auth.AuthTestUtils;
import org.apache.cassandra.auth.AuthenticatedUser;
import org.apache.cassandra.auth.IAuthenticator;
@ -48,6 +49,8 @@ public class CredentialsCacheKeysTableTest extends CQLTester
@BeforeClass
public static void setUpClass()
{
ServerTestUtils.daemonInitialization();
// high value is used for convenient debugging
DatabaseDescriptor.setCredentialsValidity(20_000);

3
test/unit/org/apache/cassandra/db/virtual/NetworkPermissionsCacheKeysTableTest.java
View File

@ -24,6 +24,7 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.auth.AuthTestUtils;
import org.apache.cassandra.auth.AuthenticatedUser;
import org.apache.cassandra.auth.DCPermissions;
@ -46,6 +47,8 @@ public class NetworkPermissionsCacheKeysTableTest extends CQLTester
@BeforeClass
public static void setUpClass()
{
ServerTestUtils.daemonInitialization();
// high value is used for convenient debugging
DatabaseDescriptor.setPermissionsValidity(20_000);

2
test/unit/org/apache/cassandra/db/virtual/PermissionsCacheKeysTableTest.java
View File

@ -28,6 +28,7 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.auth.AuthTestUtils;
import org.apache.cassandra.auth.AuthenticatedUser;
import org.apache.cassandra.auth.DataResource;
@ -52,6 +53,7 @@ public class PermissionsCacheKeysTableTest extends CQLTester
@BeforeClass
public static void setUpClass()
{
ServerTestUtils.daemonInitialization();
// high value is used for convenient debugging
DatabaseDescriptor.setPermissionsValidity(20_000);

2
test/unit/org/apache/cassandra/db/virtual/RolesCacheKeysTableTest.java
View File

@ -24,6 +24,7 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.auth.AuthTestUtils;
import org.apache.cassandra.auth.AuthenticatedUser;
import org.apache.cassandra.auth.IRoleManager;
@ -46,6 +47,7 @@ public class RolesCacheKeysTableTest extends CQLTester
@BeforeClass
public static void setUpClass()
{
ServerTestUtils.daemonInitialization();
// high value is used for convenient debugging
DatabaseDescriptor.setRolesValidity(20_000);

146
test/unit/org/apache/cassandra/io/filesystem/ForwardingFileChannel.java Normal file
View File

@ -0,0 +1,146 @@
/*
* 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.filesystem;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
public class ForwardingFileChannel extends FileChannel
{
protected final FileChannel delegate;
public ForwardingFileChannel(FileChannel delegate)
{
this.delegate = delegate;
}
protected FileChannel delegate()
{
return delegate;
}
@Override
public int read(ByteBuffer dst) throws IOException
{
return delegate().read(dst);
}
@Override
public long read(ByteBuffer[] dsts, int offset, int length) throws IOException
{
return delegate().read(dsts, offset, length);
}
@Override
public int write(ByteBuffer src) throws IOException
{
return delegate().write(src);
}
@Override
public long write(ByteBuffer[] srcs, int offset, int length) throws IOException
{
return delegate().write(srcs, offset, length);
}
@Override
public long position() throws IOException
{
return delegate().position();
}
@Override
public FileChannel position(long newPosition) throws IOException
{
return delegate().position(newPosition);
}
@Override
public long size() throws IOException
{
return delegate().size();
}
@Override
public FileChannel truncate(long size) throws IOException
{
return delegate().truncate(size);
}
@Override
public void force(boolean metaData) throws IOException
{
delegate().force(metaData);
}
@Override
public long transferTo(long position, long count, WritableByteChannel target) throws IOException
{
return delegate().transferTo(position, count, target);
}
@Override
public long transferFrom(ReadableByteChannel src, long position, long count) throws IOException
{
return delegate().transferFrom(src, position, count);
}
@Override
public int read(ByteBuffer dst, long position) throws IOException
{
return delegate().read(dst, position);
}
@Override
public int write(ByteBuffer src, long position) throws IOException
{
return delegate().write(src, position);
}
@Override
public MappedByteBuffer map(MapMode mode, long position, long size) throws IOException
{
return delegate().map(mode, position, size);
}
@Override
public FileLock lock(long position, long size, boolean shared) throws IOException
{
return delegate().lock(position, size, shared);
}
@Override
public FileLock tryLock(long position, long size, boolean shared) throws IOException
{
return delegate().tryLock(position, size, shared);
}
@Override
protected void implCloseChannel() throws IOException
{
// .close(), and .isOpen() are "final", so need to leverage implCloseChannel
// to close the underline channel
delegate().close();
}
}

129
test/unit/org/apache/cassandra/io/filesystem/ForwardingFileSystem.java Normal file
View File

@ -0,0 +1,129 @@
/*
* 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.filesystem;
import java.io.IOException;
import java.nio.file.FileStore;
import java.nio.file.FileSystem;
import java.nio.file.Path;
import java.nio.file.PathMatcher;
import java.nio.file.WatchService;
import java.nio.file.attribute.UserPrincipalLookupService;
import java.nio.file.spi.FileSystemProvider;
import java.util.Set;
import com.google.common.collect.Iterables;
public class ForwardingFileSystem extends FileSystem
{
protected final FileSystem delegate;
public ForwardingFileSystem(FileSystem delegate)
{
this.delegate = delegate;
}
protected FileSystem delegate()
{
return delegate;
}
protected Path wrap(Path p)
{
return p;
}
protected Path unwrap(Path p)
{
return p;
}
@Override
public FileSystemProvider provider()
{
return delegate().provider();
}
@Override
public void close() throws IOException
{
delegate().close();
}
@Override
public boolean isOpen()
{
return delegate().isOpen();
}
@Override
public boolean isReadOnly()
{
return delegate().isReadOnly();
}
@Override
public String getSeparator()
{
return delegate().getSeparator();
}
@Override
public Iterable<Path> getRootDirectories()
{
return Iterables.transform(delegate().getRootDirectories(), this::wrap);
}
@Override
public Iterable<FileStore> getFileStores()
{
return delegate().getFileStores();
}
@Override
public Set<String> supportedFileAttributeViews()
{
return delegate().supportedFileAttributeViews();
}
@Override
public Path getPath(String first, String... more)
{
return wrap(delegate().getPath(first, more));
}
@Override
public PathMatcher getPathMatcher(String syntaxAndPattern)
{
PathMatcher matcher = delegate().getPathMatcher(syntaxAndPattern);
return path -> matcher.matches(unwrap(path));
}
@Override
public UserPrincipalLookupService getUserPrincipalLookupService()
{
return delegate().getUserPrincipalLookupService();
}
@Override
public WatchService newWatchService() throws IOException
{
return delegate().newWatchService();
}
}

246
test/unit/org/apache/cassandra/io/filesystem/ForwardingFileSystemProvider.java Normal file
View File

@ -0,0 +1,246 @@
/*
* 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.filesystem;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.FileChannel;
import java.nio.channels.SeekableByteChannel;
import java.nio.file.AccessMode;
import java.nio.file.CopyOption;
import java.nio.file.DirectoryStream;
import java.nio.file.FileStore;
import java.nio.file.FileSystem;
import java.nio.file.LinkOption;
import java.nio.file.OpenOption;
import java.nio.file.Path;
import java.nio.file.attribute.BasicFileAttributes;
import java.nio.file.attribute.FileAttribute;
import java.nio.file.attribute.FileAttributeView;
import java.nio.file.spi.FileSystemProvider;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import com.google.common.collect.Iterators;
public class ForwardingFileSystemProvider extends FileSystemProvider
{
protected final FileSystemProvider delegate;
public ForwardingFileSystemProvider(FileSystemProvider delegate)
{
this.delegate = delegate;
}
protected FileSystemProvider delegate()
{
return delegate;
}
protected Path wrap(Path p)
{
return p;
}
protected Path unwrap(Path p)
{
return p;
}
@Override
public String getScheme()
{
return delegate().getScheme();
}
@Override
public FileSystem newFileSystem(URI uri, Map<String, ?> env) throws IOException
{
return delegate().newFileSystem(uri, env);
}
@Override
public FileSystem getFileSystem(URI uri)
{
return delegate().getFileSystem(uri);
}
@Override
public Path getPath(URI uri)
{
return wrap(delegate().getPath(uri));
}
@Override
public SeekableByteChannel newByteChannel(Path path, Set<? extends OpenOption> options, FileAttribute<?>... attrs) throws IOException
{
return delegate().newByteChannel(unwrap(path), options, attrs);
}
@Override
public DirectoryStream<Path> newDirectoryStream(Path dir, DirectoryStream.Filter<? super Path> filter) throws IOException
{
DirectoryStream<Path> stream = delegate().newDirectoryStream(unwrap(dir), filter);
return new DirectoryStream<Path>()
{
@Override
public Iterator<Path> iterator()
{
return Iterators.transform(stream.iterator(), ForwardingFileSystemProvider.this::wrap);
}
@Override
public void close() throws IOException
{
stream.close();
}
};
}
@Override
public void createDirectory(Path dir, FileAttribute<?>... attrs) throws IOException
{
delegate().createDirectory(unwrap(dir), attrs);
}
@Override
public void delete(Path path) throws IOException
{
delegate().delete(unwrap(path));
}
@Override
public void copy(Path source, Path target, CopyOption... options) throws IOException
{
delegate().copy(unwrap(source), unwrap(target), options);
}
@Override
public void move(Path source, Path target, CopyOption... options) throws IOException
{
delegate().move(unwrap(source), unwrap(target), options);
}
@Override
public boolean isSameFile(Path path, Path path2) throws IOException
{
return delegate().isSameFile(unwrap(path), unwrap(path2));
}
@Override
public boolean isHidden(Path path) throws IOException
{
return delegate().isHidden(unwrap(path));
}
@Override
public FileStore getFileStore(Path path) throws IOException
{
return delegate().getFileStore(unwrap(path));
}
@Override
public void checkAccess(Path path, AccessMode... modes) throws IOException
{
delegate().checkAccess(unwrap(path), modes);
}
@Override
public <V extends FileAttributeView> V getFileAttributeView(Path path, Class<V> type, LinkOption... options)
{
return delegate().getFileAttributeView(unwrap(path), type, options);
}
@Override
public <A extends BasicFileAttributes> A readAttributes(Path path, Class<A> type, LinkOption... options) throws IOException
{
return delegate().readAttributes(unwrap(path), type, options);
}
@Override
public Map<String, Object> readAttributes(Path path, String attributes, LinkOption... options) throws IOException
{
return delegate().readAttributes(unwrap(path), attributes, options);
}
@Override
public void setAttribute(Path path, String attribute, Object value, LinkOption... options) throws IOException
{
delegate().setAttribute(unwrap(path), attribute, value, options);
}
@Override
public FileSystem newFileSystem(Path path, Map<String, ?> env) throws IOException
{
return delegate().newFileSystem(unwrap(path), env);
}
@Override
public InputStream newInputStream(Path path, OpenOption... options) throws IOException
{
return delegate().newInputStream(unwrap(path), options);
}
@Override
public OutputStream newOutputStream(Path path, OpenOption... options) throws IOException
{
return delegate().newOutputStream(unwrap(path), options);
}
@Override
public FileChannel newFileChannel(Path path, Set<? extends OpenOption> options, FileAttribute<?>... attrs) throws IOException
{
return delegate().newFileChannel(unwrap(path), options, attrs);
}
@Override
public AsynchronousFileChannel newAsynchronousFileChannel(Path path, Set<? extends OpenOption> options, ExecutorService executor, FileAttribute<?>... attrs) throws IOException
{
return delegate().newAsynchronousFileChannel(unwrap(path), options, executor, attrs);
}
@Override
public void createSymbolicLink(Path link, Path target, FileAttribute<?>... attrs) throws IOException
{
delegate().createSymbolicLink(unwrap(link), target, attrs);
}
@Override
public void createLink(Path link, Path existing) throws IOException
{
delegate().createLink(unwrap(link), unwrap(existing));
}
@Override
public boolean deleteIfExists(Path path) throws IOException
{
return delegate().deleteIfExists(unwrap(path));
}
@Override
public Path readSymbolicLink(Path link) throws IOException
{
return wrap(delegate().readSymbolicLink(unwrap(link)));
}
}

234
test/unit/org/apache/cassandra/io/filesystem/ForwardingPath.java Normal file
View File

@ -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.io.filesystem;
import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.nio.file.FileSystem;
import java.nio.file.LinkOption;
import java.nio.file.Path;
import java.nio.file.WatchEvent;
import java.nio.file.WatchKey;
import java.nio.file.WatchService;
import java.util.Iterator;
import com.google.common.collect.Iterators;
public class ForwardingPath implements Path
{
protected final Path delegate;
public ForwardingPath(Path delegate)
{
this.delegate = delegate;
}
protected Path delegate()
{
return delegate;
}
protected Path wrap(Path a)
{
return a;
}
protected Path unwrap(Path p)
{
return p;
}
@Override
public FileSystem getFileSystem()
{
return delegate().getFileSystem();
}
@Override
public boolean isAbsolute()
{
return delegate().isAbsolute();
}
@Override
public Path getRoot()
{
return wrap(delegate().getRoot());
}
@Override
public Path getFileName()
{
return wrap(delegate().getFileName());
}
@Override
public Path getParent()
{
Path parent = delegate().getParent();
if (parent == null)
return null;
return wrap(parent);
}
@Override
public int getNameCount()
{
return delegate().getNameCount();
}
@Override
public Path getName(int index)
{
return wrap(delegate().getName(index));
}
@Override
public Path subpath(int beginIndex, int endIndex)
{
return wrap(delegate().subpath(beginIndex, endIndex));
}
@Override
public boolean startsWith(Path other)
{
return delegate().startsWith(unwrap(other));
}
@Override
public boolean startsWith(String other)
{
return delegate().startsWith(other);
}
@Override
public boolean endsWith(Path other)
{
return delegate().endsWith(unwrap(other));
}
@Override
public boolean endsWith(String other)
{
return delegate().endsWith(other);
}
@Override
public Path normalize()
{
return wrap(delegate().normalize());
}
@Override
public Path resolve(Path other)
{
return wrap(delegate().resolve(unwrap(other)));
}
@Override
public Path resolve(String other)
{
return wrap(delegate().resolve(other));
}
@Override
public Path resolveSibling(Path other)
{
return wrap(delegate().resolveSibling(unwrap(other)));
}
@Override
public Path resolveSibling(String other)
{
return wrap(delegate().resolveSibling(other));
}
@Override
public Path relativize(Path other)
{
return wrap(delegate().relativize(unwrap(other)));
}
@Override
public URI toUri()
{
return delegate().toUri();
}
@Override
public Path toAbsolutePath()
{
return wrap(delegate().toAbsolutePath());
}
@Override
public Path toRealPath(LinkOption... options) throws IOException
{
return wrap(delegate().toRealPath(options));
}
@Override
public File toFile()
{
return delegate().toFile();
}
@Override
public WatchKey register(WatchService watcher, WatchEvent.Kind<?>[] events, WatchEvent.Modifier... modifiers) throws IOException
{
return delegate().register(watcher, events, modifiers);
}
@Override
public WatchKey register(WatchService watcher, WatchEvent.Kind<?>... events) throws IOException
{
return delegate().register(watcher, events);
}
@Override
public Iterator<Path> iterator()
{
return Iterators.transform(delegate().iterator(), this::wrap);
}
@Override
public int compareTo(Path other)
{
return delegate().compareTo(unwrap(other));
}
@Override
public boolean equals(Object obj)
{
return delegate().equals(obj instanceof Path ? unwrap((Path) obj) : obj);
}
@Override
public int hashCode()
{
return delegate().hashCode();
}
@Override
public String toString()
{
return delegate().toString();
}
}

856
test/unit/org/apache/cassandra/io/filesystem/ListenableFileSystem.java Normal file
View File

@ -0,0 +1,856 @@
/*
* 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.filesystem;
import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.net.URI;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.Channels;
import java.nio.channels.FileChannel;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.SeekableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.nio.file.FileSystem;
import java.nio.file.FileSystems;
import java.nio.file.OpenOption;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.nio.file.attribute.FileAttribute;
import java.nio.file.spi.FileSystemProvider;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.memory.MemoryUtil;
public class ListenableFileSystem extends ForwardingFileSystem
{
@FunctionalInterface
public interface PathFilter
{
boolean accept(Path entry) throws IOException;
}
public interface Listener
{
}
public interface OnPreOpen extends Listener
{
void preOpen(Path path, Set<? extends OpenOption> options, FileAttribute<?>[] attrs) throws IOException;
}
public interface OnPostOpen extends Listener
{
void postOpen(Path path, Set<? extends OpenOption> options, FileAttribute<?>[] attrs, FileChannel channel) throws IOException;
}
public interface OnPreRead extends Listener
{
void preRead(Path path, FileChannel channel, long position, ByteBuffer dst) throws IOException;
}
public interface OnPostRead extends Listener
{
void postRead(Path path, FileChannel channel, long position, ByteBuffer dst, int read) throws IOException;
}
public interface OnPreTransferTo extends Listener
{
void preTransferTo(Path path, FileChannel channel, long position, long count, WritableByteChannel target) throws IOException;
}
public interface OnPostTransferTo extends Listener
{
void postTransferTo(Path path, FileChannel channel, long position, long count, WritableByteChannel target, long transfered) throws IOException;
}
public interface OnPreTransferFrom extends Listener
{
void preTransferFrom(Path path, FileChannel channel, ReadableByteChannel src, long position, long count) throws IOException;
}
public interface OnPostTransferFrom extends Listener
{
void postTransferFrom(Path path, FileChannel channel, ReadableByteChannel src, long position, long count, long transfered) throws IOException;
}
public interface OnPreWrite extends Listener
{
void preWrite(Path path, FileChannel channel, long position, ByteBuffer src) throws IOException;
}
public interface OnPostWrite extends Listener
{
void postWrite(Path path, FileChannel channel, long position, ByteBuffer src, int wrote) throws IOException;
}
public interface OnPrePosition extends Listener
{
void prePosition(Path path, FileChannel channel, long position, long newPosition) throws IOException;
}
public interface OnPostPosition extends Listener
{
void postPosition(Path path, FileChannel channel, long position, long newPosition) throws IOException;
}
public interface OnPreTruncate extends Listener
{
void preTruncate(Path path, FileChannel channel, long size, long targetSize) throws IOException;
}
public interface OnPostTruncate extends Listener
{
void postTruncate(Path path, FileChannel channel, long size, long targetSize, long newSize) throws IOException;
}
public interface OnPreForce extends Listener
{
void preForce(Path path, FileChannel channel, boolean metaData) throws IOException;
}
public interface OnPostForce extends Listener
{
void postForce(Path path, FileChannel channel, boolean metaData) throws IOException;
}
public interface Unsubscribable extends AutoCloseable
{
@Override
void close();
}
private final List<OnPreOpen> onPreOpen = new CopyOnWriteArrayList<>();
private final List<OnPostOpen> onPostOpen = new CopyOnWriteArrayList<>();
private final List<OnPreTransferTo> onPreTransferTo = new CopyOnWriteArrayList<>();
private final List<OnPostTransferTo> onPostTransferTo = new CopyOnWriteArrayList<>();
private final List<OnPreRead> onPreRead = new CopyOnWriteArrayList<>();
private final List<OnPostRead> onPostRead = new CopyOnWriteArrayList<>();
private final List<OnPreWrite> onPreWrite = new CopyOnWriteArrayList<>();
private final List<OnPostWrite> onPostWrite = new CopyOnWriteArrayList<>();
private final List<OnPreTransferFrom> onPreTransferFrom = new CopyOnWriteArrayList<>();
private final List<OnPostTransferFrom> onPostTransferFrom = new CopyOnWriteArrayList<>();
private final List<OnPreForce> onPreForce = new CopyOnWriteArrayList<>();
private final List<OnPostForce> onPostForce = new CopyOnWriteArrayList<>();
private final List<OnPreTruncate> onPreTruncate = new CopyOnWriteArrayList<>();
private final List<OnPostTruncate> onPostTruncate = new CopyOnWriteArrayList<>();
private final List<OnPrePosition> onPrePosition = new CopyOnWriteArrayList<>();
private final List<OnPostPosition> onPostPosition = new CopyOnWriteArrayList<>();
private final List<List<? extends Listener>> lists = Arrays.asList(onPreOpen, onPostOpen,
onPreRead, onPostRead,
onPreTransferTo, onPostTransferTo,
onPreWrite, onPostWrite,
onPreTransferFrom, onPostTransferFrom,
onPreForce, onPostForce,
onPreTruncate, onPostTruncate,
onPrePosition, onPostPosition);
private final ListenableFileSystemProvider provider;
public ListenableFileSystem(FileSystem delegate)
{
super(delegate);
this.provider = new ListenableFileSystemProvider(super.provider());
}
public Unsubscribable listen(Listener listener)
{
List<List<? extends Listener>> matches = new ArrayList<>(1);
if (listener instanceof OnPreOpen)
{
onPreOpen.add((OnPreOpen) listener);
matches.add(onPreOpen);
}
if (listener instanceof OnPostOpen)
{
onPostOpen.add((OnPostOpen) listener);
matches.add(onPostOpen);
}
if (listener instanceof OnPreRead)
{
onPreRead.add((OnPreRead) listener);
matches.add(onPreRead);
}
if (listener instanceof OnPostRead)
{
onPostRead.add((OnPostRead) listener);
matches.add(onPostRead);
}
if (listener instanceof OnPreTransferTo)
{
onPreTransferTo.add((OnPreTransferTo) listener);
matches.add(onPreTransferTo);
}
if (listener instanceof OnPostTransferTo)
{
onPostTransferTo.add((OnPostTransferTo) listener);
matches.add(onPostTransferTo);
}
if (listener instanceof OnPreWrite)
{
onPreWrite.add((OnPreWrite) listener);
matches.add(onPreWrite);
}
if (listener instanceof OnPostWrite)
{
onPostWrite.add((OnPostWrite) listener);
matches.add(onPostWrite);
}
if (listener instanceof OnPreTransferFrom)
{
onPreTransferFrom.add((OnPreTransferFrom) listener);
matches.add(onPreTransferFrom);
}
if (listener instanceof OnPostTransferFrom)
{
onPostTransferFrom.add((OnPostTransferFrom) listener);
matches.add(onPostTransferFrom);
}
if (listener instanceof OnPreForce)
{
onPreForce.add((OnPreForce) listener);
matches.add(onPreForce);
}
if (listener instanceof OnPostForce)
{
onPostForce.add((OnPostForce) listener);
matches.add(onPostForce);
}
if (listener instanceof OnPreTruncate)
{
onPreTruncate.add((OnPreTruncate) listener);
matches.add(onPreTruncate);
}
if (listener instanceof OnPostTruncate)
{
onPostTruncate.add((OnPostTruncate) listener);
matches.add(onPostTruncate);
}
if (listener instanceof OnPrePosition)
{
onPrePosition.add((OnPrePosition) listener);
matches.add(onPrePosition);
}
if (listener instanceof OnPostPosition)
{
onPostPosition.add((OnPostPosition) listener);
matches.add(onPostPosition);
}
if (matches.isEmpty())
throw new IllegalArgumentException("Unable to find a listenable type for " + listener.getClass());
return () -> remove(matches, listener);
}
public Unsubscribable onPreOpen(OnPreOpen callback)
{
return listen(callback);
}
public Unsubscribable onPreOpen(PathFilter filter, OnPreOpen callback)
{
return onPreOpen((path, options, attrs) -> {
if (filter.accept(path))
callback.preOpen(path, options, attrs);
});
}
public Unsubscribable onPostOpen(OnPostOpen callback)
{
return listen(callback);
}
public Unsubscribable onPostOpen(PathFilter filter, OnPostOpen callback)
{
return onPostOpen((path, options, attrs, channel) -> {
if (filter.accept(path))
callback.postOpen(path, options, attrs, channel);
});
}
public Unsubscribable onPreRead(OnPreRead callback)
{
return listen(callback);
}
public Unsubscribable onPreRead(PathFilter filter, OnPreRead callback)
{
return onPreRead((path, channel, position, dst) -> {
if (filter.accept(path))
callback.preRead(path, channel, position, dst);
});
}
public Unsubscribable onPostRead(OnPostRead callback)
{
return listen(callback);
}
public Unsubscribable onPostRead(PathFilter filter, OnPostRead callback)
{
return onPostRead((path, channel, position, dst, read) -> {
if (filter.accept(path))
callback.postRead(path, channel, position, dst, read);
});
}
public Unsubscribable onPreTransferTo(OnPreTransferTo callback)
{
return listen(callback);
}
public Unsubscribable onPreTransferTo(PathFilter filter, OnPreTransferTo callback)
{
return onPreTransferTo((path, channel, position, count, target) -> {
if (filter.accept(path))
callback.preTransferTo(path, channel, position, count, target);
});
}
public Unsubscribable onPostTransferTo(OnPostTransferTo callback)
{
return listen(callback);
}
public Unsubscribable onPostTransferTo(PathFilter filter, OnPostTransferTo callback)
{
return onPostTransferTo((path, channel, position, count, target, transfered) -> {
if (filter.accept(path))
callback.postTransferTo(path, channel, position, count, target, transfered);
});
}
public Unsubscribable onPreTransferFrom(OnPreTransferFrom callback)
{
return listen(callback);
}
public Unsubscribable onPreTransferFrom(PathFilter filter, OnPreTransferFrom callback)
{
return onPreTransferFrom((path, channel, src, position, count) -> {
if (filter.accept(path))
callback.preTransferFrom(path, channel, src, position, count);
});
}
public Unsubscribable onPostTransferFrom(OnPostTransferFrom callback)
{
return listen(callback);
}
public Unsubscribable onPostTransferFrom(PathFilter filter, OnPostTransferFrom callback)
{
return onPostTransferFrom((path, channel, src, position, count, transfered) -> {
if (filter.accept(path))
callback.postTransferFrom(path, channel, src, position, count, transfered);
});
}
public Unsubscribable onPreWrite(OnPreWrite callback)
{
return listen(callback);
}
public Unsubscribable onPreWrite(PathFilter filter, OnPreWrite callback)
{
return onPreWrite((path, channel, position, src) -> {
if (filter.accept(path))
callback.preWrite(path, channel, position, src);
});
}
public Unsubscribable onPostWrite(OnPostWrite callback)
{
return listen(callback);
}
public Unsubscribable onPostWrite(PathFilter filter, OnPostWrite callback)
{
return onPostWrite((path, channel, position, src, wrote) -> {
if (filter.accept(path))
callback.postWrite(path, channel, position, src, wrote);
});
}
public Unsubscribable onPrePosition(OnPrePosition callbackk)
{
return listen(callbackk);
}
public Unsubscribable onPrePosition(PathFilter filter, OnPrePosition callbackk)
{
return onPrePosition((path, channel, position, newPosition) -> {
if (filter.accept(path))
callbackk.prePosition(path, channel, position, newPosition);
});
}
public Unsubscribable onPostPosition(OnPostPosition callbackk)
{
return listen(callbackk);
}
public Unsubscribable onPostPosition(PathFilter filter, OnPostPosition callbackk)
{
return onPostPosition((path, channel, position, newPosition) -> {
if (filter.accept(path))
callbackk.postPosition(path, channel, position, newPosition);
});
}
public Unsubscribable onPreTruncate(OnPreTruncate callbackk)
{
return listen(callbackk);
}
public Unsubscribable onPreTruncate(PathFilter filter, OnPreTruncate callbackk)
{
return onPreTruncate((path, channel, size, targetSize) -> {
if (filter.accept(path))
callbackk.preTruncate(path, channel, size, targetSize);
});
}
public Unsubscribable onPostTruncate(OnPostTruncate callbackk)
{
return listen(callbackk);
}
public Unsubscribable onPostTruncate(PathFilter filter, OnPostTruncate callbackk)
{
return onPostTruncate((path, channel, size, targetSize, newSize) -> {
if (filter.accept(path))
callbackk.postTruncate(path, channel, size, targetSize, newSize);
});
}
public Unsubscribable onPreForce(OnPreForce callbackk)
{
return listen(callbackk);
}
public Unsubscribable onPreForce(PathFilter filter, OnPreForce callback)
{
return onPreForce((path, channel, metadata) -> {
if (filter.accept(path))
callback.preForce(path, channel, metadata);
});
}
public Unsubscribable onPostForce(OnPostForce callbackk)
{
return listen(callbackk);
}
public Unsubscribable onPostForce(PathFilter filter, OnPostForce callback)
{
return onPostForce((path, channel, metadata) -> {
if (filter.accept(path))
callback.postForce(path, channel, metadata);
});
}
public void remove(Listener listener)
{
remove(lists, listener);
}
private static void remove(List<List<? extends Listener>> lists, Listener listener)
{
lists.forEach(l -> l.remove(listener));
}
public void clearListeners()
{
lists.forEach(List::clear);
}
private interface ListenerAction<T>
{
void accept(T value) throws IOException;
}
private <T> void notifyListeners(List<T> listeners, ListenerAction<T> fn) throws IOException
{
for (T listener : listeners)
fn.accept(listener);
}
@Override
protected Path wrap(Path p)
{
return p instanceof ListenablePath ? p : new ListenablePath(p);
}
@Override
protected Path unwrap(Path p)
{
return p instanceof ListenablePath ? ((ListenablePath) p).delegate : p;
}
@Override
public ListenableFileSystemProvider provider()
{
return provider;
}
private class ListenableFileSystemProvider extends ForwardingFileSystemProvider
{
ListenableFileSystemProvider(FileSystemProvider delegate)
{
super(delegate);
}
@Override
protected Path wrap(Path a)
{
return ListenableFileSystem.this.wrap(a);
}
@Override
protected Path unwrap(Path p)
{
return ListenableFileSystem.this.unwrap(p);
}
@Override
public OutputStream newOutputStream(Path path, OpenOption... options) throws IOException
{
int len = options.length;
Set<OpenOption> opts = new HashSet<>(len + 3);
if (len == 0)
{
opts.add(StandardOpenOption.CREATE);
opts.add(StandardOpenOption.TRUNCATE_EXISTING);
}
else
{
for (OpenOption opt : options)
{
if (opt == StandardOpenOption.READ)
throw new IllegalArgumentException("READ not allowed");
opts.add(opt);
}
}
opts.add(StandardOpenOption.WRITE);
return Channels.newOutputStream(newFileChannel(path, opts));
}
@Override
public InputStream newInputStream(Path path, OpenOption... options) throws IOException
{
for (OpenOption opt : options)
{
// All OpenOption values except for APPEND and WRITE are allowed
if (opt == StandardOpenOption.APPEND ||
opt == StandardOpenOption.WRITE)
throw new UnsupportedOperationException("'" + opt + "' not allowed");
}
Set<OpenOption> opts = new HashSet<>(Arrays.asList(options));
return Channels.newInputStream(newFileChannel(path, opts));
}
@Override
public SeekableByteChannel newByteChannel(Path path, Set<? extends OpenOption> options, FileAttribute<?>... attrs) throws IOException
{
return newFileChannel(path, options, attrs);
}
@Override
public FileChannel newFileChannel(Path path, Set<? extends OpenOption> options, FileAttribute<?>... attrs) throws IOException
{
notifyListeners(onPreOpen, l -> l.preOpen(path, options, attrs));
ListenableFileChannel channel = new ListenableFileChannel(path, delegate().newFileChannel(unwrap(path), options, attrs));
notifyListeners(onPostOpen, l -> l.postOpen(path, options, attrs, channel));
return channel;
}
@Override
public AsynchronousFileChannel newAsynchronousFileChannel(Path path, Set<? extends OpenOption> options, ExecutorService executor, FileAttribute<?>... attrs) throws IOException
{
throw new UnsupportedOperationException("TODO");
}
// block the APIs that try to switch FileSystem based off schema
@Override
public FileSystem newFileSystem(URI uri, Map<String, ?> env) throws IOException
{
throw new UnsupportedOperationException();
}
@Override
public FileSystem newFileSystem(Path path, Map<String, ?> env) throws IOException
{
throw new UnsupportedOperationException();
}
@Override
public String getScheme()
{
throw new UnsupportedOperationException();
}
@Override
public FileSystem getFileSystem(URI uri)
{
throw new UnsupportedOperationException();
}
}
private class ListenablePath extends ForwardingPath
{
public ListenablePath(Path delegate)
{
super(delegate);
}
@Override
protected Path wrap(Path a)
{
return ListenableFileSystem.this.wrap(a);
}
@Override
protected Path unwrap(Path p)
{
return ListenableFileSystem.this.unwrap(p);
}
@Override
public FileSystem getFileSystem()
{
return ListenableFileSystem.this;
}
@Override
public File toFile()
{
if (delegate().getFileSystem() == FileSystems.getDefault())
return delegate().toFile();
throw new UnsupportedOperationException();
}
}
private class ListenableFileChannel extends ForwardingFileChannel
{
private final AtomicReference<Mapped> mutable = new AtomicReference<>();
private final Path path;
ListenableFileChannel(Path path, FileChannel delegate)
{
super(delegate);
this.path = path;
}
@Override
public int read(ByteBuffer dst) throws IOException
{
long position = position();
notifyListeners(onPreRead, l -> l.preRead(path, this, position, dst));
int read = super.read(dst);
notifyListeners(onPostRead, l -> l.postRead(path, this, position, dst, read));
return read;
}
@Override
public int read(ByteBuffer dst, long position) throws IOException
{
notifyListeners(onPreRead, l -> l.preRead(path, this, position, dst));
int read = super.read(dst, position);
notifyListeners(onPostRead, l -> l.postRead(path, this, position, dst, read));
return read;
}
@Override
public int write(ByteBuffer src) throws IOException
{
long position = position();
notifyListeners(onPreWrite, l -> l.preWrite(path, this, position, src));
int write = super.write(src);
notifyListeners(onPostWrite, l -> l.postWrite(path, this, position, src, write));
return write;
}
@Override
public int write(ByteBuffer src, long position) throws IOException
{
notifyListeners(onPreWrite, l -> l.preWrite(path, this, position, src));
int write = super.write(src, position);
notifyListeners(onPostWrite, l -> l.postWrite(path, this, position, src, write));
return write;
}
@Override
public FileChannel position(long newPosition) throws IOException
{
long position = position();
notifyListeners(onPrePosition, l -> l.prePosition(path, this, position, newPosition));
super.position(newPosition);
notifyListeners(onPostPosition, l -> l.postPosition(path, this, position, newPosition));
return this;
}
@Override
public FileChannel truncate(long size) throws IOException
{
long currentSize = this.size();
notifyListeners(onPreTruncate, l -> l.preTruncate(path, this, currentSize, size));
super.truncate(size);
long latestSize = this.size();
notifyListeners(onPostTruncate, l -> l.postTruncate(path, this, currentSize, size, latestSize));
return this;
}
@Override
public void force(boolean metaData) throws IOException
{
notifyListeners(onPreForce, l -> l.preForce(path, this, metaData));
super.force(metaData);
notifyListeners(onPostForce, l -> l.postForce(path, this, metaData));
}
@Override
public long transferTo(long position, long count, WritableByteChannel target) throws IOException
{
notifyListeners(onPreTransferTo, l -> l.preTransferTo(path, this, position, count, target));
long transfered = super.transferTo(position, count, target);
notifyListeners(onPostTransferTo, l -> l.postTransferTo(path, this, position, count, target, transfered));
return transfered;
}
@Override
public long transferFrom(ReadableByteChannel src, long position, long count) throws IOException
{
notifyListeners(onPreTransferFrom, l -> l.preTransferFrom(path, this, src, position, count));
long transfered = super.transferFrom(src, position, count);
notifyListeners(onPostTransferFrom, l -> l.postTransferFrom(path, this, src, position, count, transfered));
return transfered;
}
@Override
public MappedByteBuffer map(MapMode mode, long position, long size) throws IOException
{
// this behavior isn't 100% correct... if you mix access with FileChanel and ByteBuffer you will get different
// results than with a real mmap solution... This limitation is due to ByteBuffer being private, so not able
// to create custom BBs to mimc this access...
if (mode == MapMode.READ_WRITE && mutable.get() != null)
throw new UnsupportedOperationException("map called twice with mode READ_WRITE; first was " + mutable.get() + ", now " + new Mapped(mode, null, position, Math.toIntExact(size)));
int isize = Math.toIntExact(size);
MappedByteBuffer bb = (MappedByteBuffer) ByteBuffer.allocateDirect(isize);
Mapped mapped = new Mapped(mode, bb, position, isize);
if (mode == MapMode.READ_ONLY)
{
ByteBufferUtil.readFully(this, mapped.bb, position);
mapped.bb.flip();
Runnable forcer = () -> {
};
MemoryUtil.setAttachment(bb, forcer);
}
else if (mode == MapMode.READ_WRITE)
{
if (delegate().size() - position > 0)
{
ByteBufferUtil.readFully(this, mapped.bb, position);
mapped.bb.flip();
}
// with real files the FD gets copied so the close of the channel does not block the BB mutation
// from flushing... it's possible to support this use case, but kept things simplier for now by
// failing if the backing channel was closed.
Runnable forcer = () -> {
ByteBuffer local = bb.duplicate();
local.position(0);
long pos = position;
try
{
while (local.hasRemaining())
{
int wrote = write(local, pos);
if (wrote == -1)
throw new EOFException();
pos += wrote;
}
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
MemoryUtil.setAttachment(bb, forcer);
if (!mutable.compareAndSet(null, mapped))
throw new UnsupportedOperationException("map called twice");
}
else
{
throw new UnsupportedOperationException("Unsupported mode: " + mode);
}
return mapped.bb;
}
@Override
protected void implCloseChannel() throws IOException
{
super.implCloseChannel();
mutable.set(null);
}
}
private static class Mapped
{
final FileChannel.MapMode mode;
final MappedByteBuffer bb;
final long position;
final int size;
private Mapped(FileChannel.MapMode mode, MappedByteBuffer bb, long position, int size)
{
this.mode = mode;
this.bb = bb;
this.position = position;
this.size = size;
}
@Override
public String toString()
{
return "Mapped{" +
"mode=" + mode +
", position=" + position +
", size=" + size +
'}';
}
}
}

89
test/unit/org/apache/cassandra/io/util/FileSystems.java Normal file
View File

@ -0,0 +1,89 @@
/*
* 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.util;
import java.io.IOException;
import java.nio.file.FileSystem;
import com.google.common.base.StandardSystemProperty;
import com.google.common.jimfs.Configuration;
import com.google.common.jimfs.Jimfs;
import org.apache.cassandra.io.filesystem.ForwardingFileSystem;
import org.apache.cassandra.io.filesystem.ListenableFileSystem;
public class FileSystems
{
public static ListenableFileSystem newGlobalInMemoryFileSystem()
{
return global(jimfs());
}
public static ListenableFileSystem global()
{
return global(File.unsafeGetFilesystem());
}
public static ListenableFileSystem global(FileSystem real)
{
FileSystem current = File.unsafeGetFilesystem();
ListenableFileSystem fs = new ListenableFileSystem(new ForwardingFileSystem(real)
{
@Override
public void close() throws IOException
{
try
{
super.close();
}
finally
{
File.unsafeSetFilesystem(current);
}
}
});
File.unsafeSetFilesystem(fs);
return fs;
}
public static FileSystem jimfs()
{
return Jimfs.newFileSystem(jimfsConfig());
}
public static FileSystem jimfs(String name)
{
return Jimfs.newFileSystem(name, jimfsConfig());
}
private static Configuration jimfsConfig()
{
return Configuration.unix().toBuilder()
.setMaxSize(4L << 30).setBlockSize(512)
.build();
}
public static File maybeCreateTmp()
{
File dir = new File(StandardSystemProperty.JAVA_IO_TMPDIR.value());
if (!dir.exists())
dir.tryCreateDirectories();
return dir;
}
}

3
test/unit/org/apache/cassandra/schema/RemoveWithoutDroppingTest.java
View File

@ -28,6 +28,7 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.io.util.File;
@ -46,6 +47,8 @@ public class RemoveWithoutDroppingTest
@BeforeClass
public static void beforeClass()
{
ServerTestUtils.daemonInitialization();
System.setProperty(SchemaUpdateHandlerFactoryProvider.SUH_FACTORY_CLASS_PROPERTY, TestSchemaUpdateHandlerFactory.class.getName());
CQLTester.prepareServer();
Schema.instance.registerListener(listener);

3
test/unit/org/apache/cassandra/service/PartitionDenylistTest.java
View File

@ -23,6 +23,7 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.UntypedResultSet;
@ -44,6 +45,8 @@ public class PartitionDenylistTest
@BeforeClass
public static void init()
{
ServerTestUtils.daemonInitialization();
CQLTester.prepareServer();
KeyspaceMetadata schema = KeyspaceMetadata.create(ks_cql,