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

Fix storage_compatibility_mode for streaming 

- Rename and refactor a bit property for overriding storage compatibility mode
- Fix streaming for tools
- consider bulkloader as a client instead of tool

Patch by Jacek Lewandowski; reviewed by Berenguer Blasi, Branimir Lambov for CASSANDRA-19126
This commit is contained in:
Jacek Lewandowski 2023-12-08 10:41:50 +01:00
parent c76b32492f
commit d422eb1f35
10 changed files with 56 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.0-beta2
* Fix storage_compatibility_mode for streaming (CASSANDRA-19126)
* Add support of vector type to cqlsh COPY command (CASSANDRA-19118)
* Make CQLSSTableWriter to support building of SAI indexes (CASSANDRA-18714)
* Append additional JVM options when using JDK17+ (CASSANDRA-19001)

4
build.xml
View File

@ -1310,7 +1310,7 @@
<jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
<jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
<jvmarg value="-Dcassandra.config=file:///${trie_yaml}"/>
<jvmarg value="-Dcassandra.junit_storage_compatibility_mode=NONE"/>
<jvmarg value="-Dcassandra.test.storage_compatibility_mode=NONE"/>
<jvmarg value="-Dcassandra.skip_sync=true" />
</testmacrohelper>
</sequential>
@ -1331,7 +1331,7 @@
<jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
<jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
<jvmarg value="-Dcassandra.config=file:///${scm_none_yaml}"/>
<jvmarg value="-Dcassandra.junit_storage_compatibility_mode=NONE"/>
<jvmarg value="-Dcassandra.test.storage_compatibility_mode=NONE"/>
<jvmarg value="-Dcassandra.skip_sync=true" />
</testmacrohelper>
</sequential>

17
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
View File

@ -316,14 +316,6 @@ public enum CassandraRelevantProperties
/** Java Virtual Machine implementation name */
JAVA_VM_NAME("java.vm.name"),
JOIN_RING("cassandra.join_ring", "true"),
/**
* {@link StorageCompatibilityMode} mode sets how the node will behave, sstable or messaging versions to use etc according to a yaml setting.
* But many tests don't load the config hence we need to force it otherwise they would run always under the default. Config is null for junits
* that don't load the config. Get from env var that CI/build.xml sets.
*
* This is a dev/CI only property. Do not use otherwise.
*/
JUNIT_STORAGE_COMPATIBILITY_MODE("cassandra.junit_storage_compatibility_mode", StorageCompatibilityMode.CASSANDRA_4.toString()),
/** startup checks properties */
LIBJEMALLOC("cassandra.libjemalloc"),
/** Line separator ("\n" on UNIX). */
@ -562,6 +554,15 @@ public enum CassandraRelevantProperties
TEST_SIMULATOR_PRINT_ASM_TYPES("cassandra.test.simulator.print_asm_types", ""),
TEST_SKIP_CRYPTO_PROVIDER_INSTALLATION("cassandra.test.security.skip.provider.installation", "false"),
TEST_SSTABLE_FORMAT_DEVELOPMENT("cassandra.test.sstableformatdevelopment"),
/**
* {@link StorageCompatibilityMode} mode sets how the node will behave, sstable or messaging versions to use etc.
* according to a yaml setting. But many tests don't load the config hence we need to force it otherwise they would
* run always under the default. Config is null for junits that don't load the config. Get from env var that
* CI/build.xml sets.
*
* This is a dev/CI only property. Do not use otherwise.
*/
TEST_STORAGE_COMPATIBILITY_MODE("cassandra.test.storage_compatibility_mode", StorageCompatibilityMode.CASSANDRA_4.toString()),
TEST_STRICT_LCS_CHECKS("cassandra.test.strict_lcs_checks"),
/** Turns some warnings into exceptions for testing. */
TEST_STRICT_RUNTIME_CHECKS("cassandra.strict.runtime.checks"),

2
src/java/org/apache/cassandra/config/Config.java
View File

@ -1269,5 +1269,5 @@ public class Config
public double severity_during_decommission = 0;
public StorageCompatibilityMode storage_compatibility_mode = StorageCompatibilityMode.CASSANDRA_4;
public StorageCompatibilityMode storage_compatibility_mode;
}

22
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -228,6 +228,7 @@ public class DatabaseDescriptor
private static ImmutableMap<String, SSTableFormat<?, ?>> sstableFormats;
private static volatile SSTableFormat<?, ?> selectedSSTableFormat;
private static StorageCompatibilityMode storageCompatibilityMode = CassandraRelevantProperties.TEST_STORAGE_COMPATIBILITY_MODE.getEnum(true, StorageCompatibilityMode.class);
private static Function<CommitLog, AbstractCommitLogSegmentManager> commitLogSegmentMgrProvider = c -> DatabaseDescriptor.isCDCEnabled()
? new CommitLogSegmentManagerCDC(c, DatabaseDescriptor.getCommitLogLocation())
@ -291,6 +292,8 @@ public class DatabaseDescriptor
setConfig(loadConfig());
applyCompatibilityMode();
applySSTableFormats();
applySimpleConfig();
@ -346,6 +349,7 @@ public class DatabaseDescriptor
setDefaultFailureDetector();
Config.setClientMode(true);
conf = configSupplier.get();
applyCompatibilityMode();
diskOptimizationStrategy = new SpinningDiskOptimizationStrategy();
applySSTableFormats();
}
@ -434,7 +438,8 @@ public class DatabaseDescriptor
private static void applyAll() throws ConfigurationException
{
//InetAddressAndPort cares that applySimpleConfig runs first
applyCompatibilityMode();
applySSTableFormats();
applyCryptoProvider();
@ -1521,6 +1526,15 @@ public class DatabaseDescriptor
return selectedFormat;
}
private static void applyCompatibilityMode()
{
if (isClientInitialized())
// tools or clients should not limit the sstable formats they support
storageCompatibilityMode = StorageCompatibilityMode.NONE;
else if (conf != null && conf.storage_compatibility_mode != null)
storageCompatibilityMode = conf.storage_compatibility_mode;
}
private static void applySSTableFormats()
{
ServiceLoader<SSTableFormat.Factory> loader = ServiceLoader.load(SSTableFormat.Factory.class, DatabaseDescriptor.class.getClassLoader());
@ -4968,11 +4982,7 @@ public class DatabaseDescriptor
public static StorageCompatibilityMode getStorageCompatibilityMode()
{
// Config is null for junits that don't load the config. Get from env var that CI/build.xml sets
if (conf == null)
return CassandraRelevantProperties.JUNIT_STORAGE_COMPATIBILITY_MODE.getEnum(StorageCompatibilityMode.CASSANDRA_4);
else
return conf.storage_compatibility_mode;
return storageCompatibilityMode;
}
public static ParameterizedClass getDefaultCompaction()

28
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -31,10 +31,10 @@ import java.util.stream.Collectors;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import io.netty.util.concurrent.Future; //checkstyle: permit this import
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.util.concurrent.Future; //checkstyle: permit this import
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.config.DatabaseDescriptor;
@ -51,7 +51,6 @@ import org.apache.cassandra.utils.concurrent.FutureCombiner;
import static java.util.Collections.synchronizedList;
import static java.util.concurrent.TimeUnit.MINUTES;
import static org.apache.cassandra.concurrent.Stage.MUTATION;
import static org.apache.cassandra.config.CassandraRelevantProperties.NON_GRACEFUL_SHUTDOWN;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
@ -222,8 +221,6 @@ public class MessagingService extends MessagingServiceMBeanImpl implements Messa
// c14227 TTL overflow, 'uint' timestamps
VERSION_50(13);
public static final Version CURRENT = DatabaseDescriptor.getStorageCompatibilityMode().isBefore(5) ? VERSION_40 : VERSION_50;
public final int value;
Version(int value)
@ -253,9 +250,26 @@ public class MessagingService extends MessagingServiceMBeanImpl implements Messa
public static final int VERSION_40 = 12;
public static final int VERSION_50 = 13; // c14227 TTL overflow, 'uint' timestamps
public static final int minimum_version = VERSION_40;
public static final int current_version = Version.CURRENT.value;
static AcceptVersions accept_messaging = new AcceptVersions(minimum_version, current_version);
static AcceptVersions accept_streaming = new AcceptVersions(current_version, current_version);
public static final int maximum_version = VERSION_50;
// we want to use a modified behavior for the tools and clients - that is, since they are not running a server, they
// should not need to run in a compatibility mode. They should be able to connect to the server regardless whether
// it uses messaving version 4 or 5
public static final int current_version = DatabaseDescriptor.getStorageCompatibilityMode().isBefore(5) ? VERSION_40 : VERSION_50;
static AcceptVersions accept_messaging;
static AcceptVersions accept_streaming;
static
{
if (DatabaseDescriptor.isClientInitialized())
{
accept_messaging = new AcceptVersions(minimum_version, maximum_version);
accept_streaming = new AcceptVersions(minimum_version, maximum_version);
}
else
{
accept_messaging = new AcceptVersions(minimum_version, current_version);
accept_streaming = new AcceptVersions(current_version, current_version);
}
}
static Map<Integer, Integer> versionOrdinalMap = Arrays.stream(Version.values()).collect(Collectors.toMap(v -> v.value, v -> v.ordinal()));
/**

2
src/java/org/apache/cassandra/net/OutboundConnectionSettings.java
View File

@ -420,7 +420,7 @@ public class OutboundConnectionSettings
if (tcpNoDelay != null)
return tcpNoDelay;
if (isInLocalDC(getEndpointSnitch(), getBroadcastAddressAndPort(), to))
if (DatabaseDescriptor.isClientOrToolInitialized() || isInLocalDC(getEndpointSnitch(), getBroadcastAddressAndPort(), to))
return INTRADC_TCP_NODELAY;
return DatabaseDescriptor.getInterDCTcpNoDelay();

4
src/java/org/apache/cassandra/repair/messages/PrepareMessage.java
View File

@ -89,7 +89,9 @@ public class PrepareMessage extends RepairMessage
}
private static final String MIXED_MODE_ERROR = "Some nodes involved in repair are on an incompatible major version. " +
"Repair is not supported in mixed major version clusters.";
"Repair is not supported in mixed major version clusters. Note that " +
"5.x nodes running in storage compatibility mode = 4 are considered " +
"4.x nodes.";
public static final IVersionedSerializer<PrepareMessage> serializer = new IVersionedSerializer<PrepareMessage>()
{

2
src/java/org/apache/cassandra/tools/BulkLoader.java
View File

@ -60,7 +60,7 @@ public class BulkLoader
public static void load(LoaderOptions options) throws BulkLoadException
{
DatabaseDescriptor.toolInitialization();
DatabaseDescriptor.clientInitialization();
OutputHandler handler = new OutputHandler.SystemOutput(options.verbose, options.debug);
SSTableLoader loader = new SSTableLoader(
options.directory.toAbsolute(),

2
test/unit/org/apache/cassandra/repair/FuzzTestBase.java
View File

@ -1204,7 +1204,7 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
{
try (DataOutputBuffer b = DataOutputBuffer.scratchBuffer.get())
{
int messagingVersion = MessagingService.Version.CURRENT.value;
int messagingVersion = MessagingService.current_version;
Message.serializer.serialize(msg, b, messagingVersion);
DataInputBuffer in = new DataInputBuffer(b.unsafeGetBufferAndFlip(), false);
return Message.serializer.deserialize(in, msg.from(), messagingVersion);