This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch cassandra-6.0 in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 679313653e1884d199aaad29ae5040391b347407 Merge: 9f3df03014 d020df9ce3 Author: Marcus Eriksson <[email protected]> AuthorDate: Wed Oct 21 15:32:04 2026 +0200 Merge branch 'cassandra-5.0' into cassandra-6.0 CHANGES.txt | 1 + NEWS.txt | 6 + conf/cassandra.yaml | 8 ++ src/java/org/apache/cassandra/config/Config.java | 5 + .../cassandra/config/DatabaseDescriptor.java | 10 ++ .../CassandraEntireSSTableStreamReader.java | 39 ++++++ .../cassandra/io/util/DataIntegrityMetadata.java | 10 +- .../apache/cassandra/metrics/StreamingMetrics.java | 8 ++ .../apache/cassandra/streaming/StreamSession.java | 5 + .../test/hostreplacement/HostReplacementTest.java | 83 +++++++++++++ .../CassandraEntireSSTableStreamWriterTest.java | 135 +++++++++++++++++++++ .../io/util/DataIntegrityMetadataTest.java | 72 +++++++++++ 12 files changed, 381 insertions(+), 1 deletion(-) diff --cc CHANGES.txt index e3e95fc728,8d2d52ecc9..b35c3c4a60 --- a/CHANGES.txt +++ b/CHANGES.txt @@@ -73,10 -15,10 +73,11 @@@ Merged from 5.0 * Render SubnetGroups as JSON in system_views.settings (CASSANDRA-21579) * Avoid rebuilding per-SSTable SAI components unless missing or corrupted (CASSANDRA-21515) * Propagate trickle_fsync settings to compressed SSTable writers (CASSANDRA-21487) - * Allow DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(0) to disable read-ahead buffer (CASSANDRA-21522) - * Return CorruptSSTableException if chunk metadata and file size are out of sync (CASSANDRA-21519) + * Allow setCompressedReadAheadBufferSizeInKb(0) to disable read-ahead buffer (CASSANDRA-21522) + * Fix ThreadLocalReadAheadBuffer#fill() to throw a CorruptBlockException if chunk metadata and file size are out of sync (CASSANDRA-21519) + * Fix memtable on-heap accounting drift in BTree.update and BTreeRow.merge (CASSANDRA-21472) Merged from 4.0: + * Validate Digest.crc32 when zero-copy streaming SSTables (CASSANDRA-21694) * Avoid DNS lookup for hostname column in ClientsTable and GossipInfoTable (CASSANDRA-21539) * Separate memtable memory usage tracking from limit checking, and fixes a a flush writeBarrier deadlock (CASSANDRA-21019) * Fixes for fully-expired SSTable index notification in compaction (CASSANDRA-21577) diff --cc NEWS.txt index af526ab092,ca6eb77eca..187a66f6ec --- a/NEWS.txt +++ b/NEWS.txt @@@ -89,168 -93,79 +89,174 @@@ New feature and manage repairs. See https://cwiki.apache.org/confluence/display/CASSANDRA/CEP-37+Apache+Cassandra+Unified+Repair+Solution for more details on the motivation and design. - -Upgrading ---------- - - The auto-repair feature requires enabling the JVM property `cassandra.autorepair.enable=true` (add - `-Dcassandra.autorepair.enable=true` to JVM options) before starting the node. This property creates the required - schema elements for auto-repair, including the auto_repair column in system_schema.tables and system_schema.views, - as well as the auto_repair_history and auto_repair_priority tables in system_distributed. After enabling this - property, you still need to enable auto-repair scheduling either in cassandra.yaml under the `auto_repair` section - or at runtime via JMX. - - Users who do not intend to use auto-repair can leave this property disabled (the default) to maintain schema - compatibility with pre-5.0.8 nodes during rolling upgrades. This property must be set consistently across all - nodes before startup and cannot be changed at runtime. - - WARNING: This property is non-reversible. Once enabled, it cannot be disabled. Attempting to start a node - with `cassandra.autorepair.enable=false` after it was previously enabled will cause the node to fail during - initialization due to schema incompatibility (the persisted schema contains auto-repair columns that are not - recognized when the property is disabled). To disable auto-repair scheduling after the property has been - enabled, use cassandra.yaml or JMX instead of changing the JVM property. - - IMPORTANT: The `cassandra.autorepair.enable` property must be enabled consistently across all nodes in the - cluster before any schema changes are made. When some nodes have the property enabled and others do not, the - system_distributed keyspace schema generation will differ between nodes (generation 7 with auto-repair vs - generation 6 without), causing schema disagreement. This is similar to what happens during a major version - upgrade when new system tables are added. Any schema change (e.g. CREATE KEYSPACE) attempted while nodes - are in this inconsistent state will time out and schema versions will not converge until all nodes are - brought up with the same setting. Once all nodes have the property set consistently, schema will converge - automatically. - -5.0.7 -====== - -Upgrading ---------- - - This release addresses a security vulnerability in the LZ4 dependency, documented in CASSANDRA-21052. For users - who did not use LZ4 native libraries, this will now fallback to a safer but less performant pure Java - implementation. During startup, a warning will be logged if the LZ4 native library is not available. - -5.0.5 -===== - -New features ------------- - - Full support for Java 17, it is not experimental anymore. - -5.0.4 -===== - + - CEP-21 Transactional Cluster Metadata introduces a distributed log for linearizing modifications to cluster + metadata. In the first instance, this encompasses cluster membership, token ownership and schema metadata. See + https://cwiki.apache.org/confluence/display/CASSANDRA/CEP-21%3A+Transactional+Cluster+Metadata for more detail on + the motivation and design, and see "Upgrading" below for specific instructions on migrating clusters to this system. + - CEP-42 Constraints Framework provides flexibility to Cassandra users and operators by providing a set of + usable constraints at table level, that will ease validations at application level and protect + the Database from misconfigured clients. + More updates and documentation to follow. + - New Guardrails added: + - Whether bulk loading of SSTables is allowed. + - Allowed keyspace properties. + - nodetool tpstats can display core pool size, max pool size and max tasks queued if --verbose / -v flag is specified. + system_views.thread_pools adds core_pool_size, max_pool_size and max_tasks_queued columns. + - Authentication mode is exposed in system_views.clients table, nodetool clientstats and ClientMetrics + to help operators identify which authentication modes are being used. nodetool clientstats introduces --verbose flag + behind which this information is visible. + - CEP-24 - Password validation / generation. When built-in 'password_policy' guardrail is enabled, it will + generate a password of configured password strength policy upon role creation or alteration + when 'GENERATED PASSWORD' clause is used. Character sets supported are: English, Cyrillic, modern Cyrillic, + German, Polish and Czech. + - CEP-55 - Role name generation and validation. When built-in 'role_name_policy' guardrail is enabled, it will + generate a role name automatically, without operators intervention. It is possible to configure + this policy for both generation and as well as validation of role names. + - There is new MBean of name org.apache.cassandra.service.snapshot:type=SnapshotManager which exposes user-facing + snapshot operations. Snapshot-related methods on StorageServiceMBean are still present and functional + but marked as deprecated. + - The unified compaction strategy will now parallelize individual compactions by splitting operations into separate + tasks per output shard. This significantly shortens compaction durations but cannot take advantage of preemptive + SSTable opening and thus can be less efficient if the strategy is configured to create large sstables. + This functionality is controlled by the parallelize_output_shards compaction option (the default can also be set + using -Dunified_compaction.parallelize_output_shards=false/true). + - Compaction parallelism for the unified compaction strategy also applies to major compactions. Because the number + of shards for a major compaction is usually quite high, this can dramatically improve the duration of major + compactions. To avoid the possibility of starving background compactions from resources, the number of threads + used for major compactions is limited to half the available compaction threads by default, and can be controlled + by a new --jobs / -j option of nodetool compact. + - It is possible to configure JMX server in cassandra.yaml in jmx_server_options configuration section. + JMX SSL configuration can be configured via jmx_encryption_options in jmx_server_options. The old way of + configuring JMX is still present and default, but new way is preferable as it will e.g. not leak credentials for + JMX to JVM parameters. You have to opt-in to use the configuration via cassandra.yaml by uncommenting + the respective configuration sections and by commenting out `configure_jmx` function call in cassandra-env.sh. + Enabling both ways of configuring JMX will result in a node failing to start. + - CEP-43 - it is possible to create a table by "copying" as `CREATE TABLE ks.tb_copy LIKE ks.tb;`. + A newly created table will have no data. + - New functions `octet_length` and `length` were introduced. Previously, users would either have to read + the data and check the length themselves or enable UDF and register UDFs to check the length of columns. + CASSANDRA-20102 adds a subset of the SQL99 (binary) string functions: "octet_length" defined on all types + and "length" defined on UTF8 strings. See CASSANDRA-20102 for more information. + - New functions `format_bytes` and `format_time` were added. See CASSANDRA-19546. + - It is possible to use Async-profiler for various profiling scenarios. See CASSANDRA-20854. + - It is possible to provide custom startup check via Java SPI. See CASSANDRA-21093. + - Single partition data reads served by the coordinator's local replica no longer serialize and deserialize the + response when it is small enough. The rows are kept as an in-memory partition instead, which saves the CPU time + spent on serializing and deserializing them and avoids allocating the rows and cells twice. + A response that exceeds either per-request limit is serialized in full, as before. + The limits are set by the system properties cassandra.data_response_in_memory_max_rows and + cassandra.data_response_in_memory_max_size; reads at ONE/LOCAL_ONE use + cassandra.data_response_in_memory_max_rows_cl_one andcassandra.data_response_in_memory_max_size_cl_one. + A limit is disabled by setting it to 0, and setting both limits of a tier to 0 disables the in-memory path for that tier. + Note that the limits are not enforced in the middle of a range tombstone span, so a response may exceed them in that case. + - SAI Memtable-attached indexes now support sharding via WITH OPTIONS. See CQL documentation for details. Upgrading --------- - - When more than one index exists on a single column, legacy table-backed indexes will now be prioritized over - storage-attached indexes (SAI) to make migration between the two safer. This behavior can be switched off via - the flag `sai_options.prioritize_over_legacy_index` (which defaults to `false`) in `cassandra.yaml` or via - `setPrioritizeSAIOverLegacyIndex(boolean)` in the JMX MBean `org.apache.cassandra.db:type=StorageService`. - - Java's -XX:MaxDirectMemorySize is by default explicitly half of max heap size (-Xmx) instead of implicitly equal to it. - -5.0.1 -===== - -Upgrading ---------- - - This release introduces safeguards and observability into possible data loss scenarios when nodes have a - divergent view of the cluster. This happens around edge-cases on unsafe bootstrapping, decommissions, or when a - node has a corrupted topology. Two configuration options have been added: `log_out_of_token_range_requests` and - `reject_out_of_token_range_requests`, both enabled by default. The former will make nodes log requests they - receive that don't belong in their current or pending token ranges. The latter will reject those requests - altogether, which prevents the vast majority of eventual data loss that can occur, but may also incur small - windows of degraded availability during range movements in an unhealthy cluster. Both options apply to streaming, - mutations, hints, single-partition reads, read-repair, and Paxos operations. Logging these requests is recommended - in all situations, and operators are encouraged to be aware when such events occur and to investigate if writes - have only been received on incorrect nodes. Operators that are confident of topology parity across all nodes may - disable rejection to maximize availability. See CASSANDRA-13704 for further details. Transactional Cluster - Metadata in version 5.1 will remove all such edge-cases, providing appropriate guarantees. + - memtable_offheap_space: 0 is now rejected at startup when memtable_allocation_type is + offheap_buffers or offheap_objects. Set a positive memtable_offheap_space, or an allocation + type that does not allocate off-heap. See CASSANDRA-21019. ++ - entire_sstable_stream_digest_validation_enabled now defaults to true. Entire-SSTable (zero-copy) ++ streams are validated against the streamed Digest.crc32 before being added to the table, which ++ detects corruption introduced in transit or already present on the sending node's disk. This may ++ reject sstables that would previously have been silently accepted despite being corrupted. Set ++ entire_sstable_stream_digest_validation_enabled to false in cassandra.yaml to restore the old ++ behavior. (CASSANDRA-21694) + [The following is a placeholder, to be revised asap] + - trickle_fsync is by default set to "true" instead of "false" in cassandra.yaml (see CASSANDRA-21572) + - Upgrading to 5.1 is currently only supported from 4.0, 4.1 or 5.0. Clusters running earlier releases must first + upgrade to at least the latest 4.0 release. When a cluster is upgraded to 5.1 there is an additional step for + operators to perform. The upgrade is not considered complete until: + 1. All UP nodes are running 5.1 + 2. The Cluster Metadata Service (CMS) has been enabled + The first step is just a regular upgrade, there are no changes to external APIs, SSTable formats to consider. + Step 2 requires an operator to run a new nodetool subcommand, intializecms, on one node in the cluster. + - > nodetool cms initialize + Doing so creates the CMS with the local node as its only member. The initialize command cannot be executed until + step 1 is complete and all nodes are running 5.1 as migrating metadata management over to the new TCM system + requires all nodes to be in agreement over the current state of the cluster. Essentially this means agreement on + schema and topology. Once the upgrade has started, but before the initialize command is run, metadata-changing + operations are not permitted and if attempted from an upgraded node will be rejected. + Prohibited operations include: + - schema changes + - node replacement + - bootstrap + - decommission + - move + - assasinate + For the time being there is no mechanism in place to prevent these operations being executed by a node still running + a previous version, though this is planned before release. Any automation which can trigger these operations should + be disabled for the cluster prior to starting the upgrade. + Should any of the prohibited operations be executed (i.e. on a node that is still running a pre-5.1 version) before + the CMS intialization, nodes which are DOWN or which have been upgraded will not process the metadata changes. + However, nodes still UP and running the old version will. This will eventually cause the migration to fail, as the + cluster will not be in agreement. + - > nodetool cms initialize + Got mismatching cluster metadatas. Check logs on peers ([/x.x.x.x:7000]) for details of mismatches. + Aborting migration. + See 'nodetool help' or 'nodetool help <command>'. + If the cms initialize command fails, it will indicate which nodes’ current metadata does not agree with the node + where the command was executed. To mitigate this situation, bring any mismatching nodes DOWN and rerun the + cms initialize command with the additional —ignore flag. + - nodetool cms intialize -ignore x.x.x.x + Once the command has run successfully the ignored nodes can be restarted but any metadata changes that they + accepted/and or applied whilst the cluster was in mixed mode will be lost. We plan to improve this before beta, but + in the meantime operators should ensure no schema or membership/ownership changes are performed during the upgrade. + Although the restrictions on metadata-changing operations will be removed as soon as the initial CMS migration is + complete, at that point the CMS will only contain a single member, which is not suitable for real clusters. To + modify the membership of the CMS a second nodetool subcommand, cms reconfigure, should be run. This command allows + the number of required number of CMS members to be specified for each datacenter. Consensus for metadata operations + by the CMS is obtained via Paxos, operating at SERIAL/QUORUM consistency, so the minimum safe size for the CMS is 3 + nodes. In multi-DC deployments, the CMS members may be distributed across DCs to ensure resilience in the case of a + DC outage. It is not recommended to make every node a member of the CMS. Redundancy is the primary concern here, not + locality or latency, so between 3 and 7 nodes per DC depending on the size of the cluster should be the goal. + Deploying to a fresh cluster is more straightforward. As the cluster comes up, at least one node will automatically + nominate itself as the first CMS node. A simple election mechanism ensures that only one node will be chosen if + multiple peers attempt to nominate themselves. As soon as the election is complete, metadata modifications are + supported. Typically, this takes only a few seconds and is likely to complete before all nodes are fully UP. Nodes + which come up during or after an election will learn of the elected first CMS node and direct metadata updates to + it. It is important to remember that at the completion of the election, the CMS still only comprises a single + member. Just as in the upgrade case, operators should add further members as soon as possible. + - native_transport_port_ssl property was removed. Please transition to using one port only. Encrypted communication + may be optional by setting `optional` flag in `client_encryption_options` to `true` and it should be set only + while in unencrypted or transitional operation. Please consult `client_encryption_options` in cassandra.yaml + for more information. + - system_views.settings virtual table output changed to json for collections, set system property + `cassandra.virtual_table_complex_settings_format_json` to false to disable (defaults to true). + - Authorizing a batch or an Accord transaction now performs the table permission checks once per run of + consecutive statements on the same table rather than once per statement. The permissions required are + unchanged, and the datacenter and CIDR checks that ride along with them still run at least once per + request. Auth cache request/hit/miss counts and the system_views.cidr_filtering_metrics_* counts and + latencies will therefore report fewer events than before. (see CASSANDRA-21606) + - Guardrail configurations of zero are now treated as zero instead of unlimited. See CASSANDRA-21517. + - The hostname column in system_views.clients will no longer perform a reverse DNS lookup and will instead + return the same value as the address column. This avoids the risk of client connections overwhelming + DNS. (CASSANDRA-21539) + +Deprecation +----------- + - `use_deterministic_table_id` is no longer supported and should be removed from cassandra.yaml. Table IDs may still be supplied explicitly on CREATE. + - Updating dependencies for JDK21 necessitated updating Chronicle Queue, which has changed the enums used for log + rolling (cassandra.yaml -> full_query_logging_options:roll_cycle). Older legacy options will still work for the + foreseeable future but you will see warnings in logs and future dependency upgrades may break your log rolling param. + The default log rolling param has been changed from HOURLY to FAST_HOURLY, primarily different on how frequently + indexes are built (256 in FAST_HOURLY vs. 16 in HOURLY). + - IEndpointSnitch has been deprecated as ClusterMetadata is now the source of truth regarding topology information. + The responsibilities of the snitch have been broken out to a handful of new classes: + * o.a.c.locator.Locator provides datacenter and rack info for endpoints. This is not configurable as topology is + always sourced from ClusterMetadata. + * o.a.c.locator.InitialLocationProvider supplies the local datacenter and rack for a new node joining a cluster + for the first time. This will be used exactly once to register the Location of the node in ClusterMetadata and + is configurable using the `initial_location_provider` yaml setting. + * o.a.c.locator.NodeProximity handles sorting and ranking of replica lists. It is configurable via the + `node_proximity` yaml setting. + * o.a.c.locator.NodeAddressConfig exists mainly to support the functionality of ReconnectableSnitchHelper and + Ec2MultiRegionSnitch which dynamically configures the broadcast address based on cloud metadata. Optionally + configurable using the `address_config` yaml setting. + For migration and to allow us to deprecate snitches in a controlled way, use of endpoint_snitch in yaml remains + fully supported and all existing IEndpointSnitch implementations should continue to work seamlessly and no action + is required by operators at upgrade time. The in-tree snitches have been refactored to extract implementations of + the new interfaces so that their functionality can also be used via the new `initial_location_provider`, + `node_proximity` and `address_config` settings. 5.0 === diff --cc conf/cassandra.yaml index b3fe1452f1,895350de63..aa08a102d2 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@@ -1316,6 -1283,14 +1316,14 @@@ uuid_sstable_identifiers_enabled: fals # disabled if internode encryption is enabled. # stream_entire_sstables: true + # When an entire SSTable is received via zero-copy streaming, validate the Digest.crc32 that was streamed with it + # against the received Data.db before the SSTable is added to the table. This detects corruption introduced in + # transit as well as pre-existing corruption on the sending node's disk, neither of which is otherwise noticed by + # this path, since it writes component bytes to disk without parsing them. + # If there's no Digest.crc32, this check passes. -# When unset, the default is not validating Digest.crc32. -# entire_sstable_stream_digest_validation_enabled: false ++# When unset, the default is validating Digest.crc32. ++# entire_sstable_stream_digest_validation_enabled: true + # Throttles entire SSTable outbound streaming file transfers on # this node to the given total throughput in Mbps. # Setting this value to 0 it disables throttling. diff --cc src/java/org/apache/cassandra/config/Config.java index 2ec4d7f236,dca5f03a4e..86260395f1 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@@ -809,6 -698,11 +809,11 @@@ public class Confi public volatile int max_concurrent_automatic_sstable_upgrades = 1; public boolean stream_entire_sstables = true; + /** + * When receiving a zero-copy stream, validate Digest.crc32 before the sstable is made visible. + */ - public volatile boolean entire_sstable_stream_digest_validation_enabled = false; ++ public volatile boolean entire_sstable_stream_digest_validation_enabled = true; + public volatile boolean skip_stream_disk_space_check = false; public volatile AuditLogOptions audit_logging_options = new AuditLogOptions(); diff --cc src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java index f2f1784a9c,81a82c86cf..1209653fc2 --- a/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java +++ b/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java @@@ -19,8 -19,9 +19,10 @@@ package org.apache.cassandra.db.streaming; import java.io.IOException; +import java.nio.channels.ClosedChannelException; import java.util.Collection; + import java.util.List; + import java.util.concurrent.TimeUnit; import java.util.function.UnaryOperator; import org.slf4j.Logger; @@@ -37,10 -36,11 +39,12 @@@ import org.apache.cassandra.io.sstable. import org.apache.cassandra.io.sstable.Descriptor; import org.apache.cassandra.io.sstable.IOOptions; import org.apache.cassandra.io.sstable.SSTableMultiWriter; +import org.apache.cassandra.io.sstable.SSTableTxnSingleStreamWriter; import org.apache.cassandra.io.sstable.SSTableZeroCopyWriter; + import org.apache.cassandra.io.sstable.format.SSTableFormat.Components; import org.apache.cassandra.io.sstable.metadata.StatsMetadata; import org.apache.cassandra.io.util.DataInputPlus; + import org.apache.cassandra.io.util.DataIntegrityMetadata; import org.apache.cassandra.io.util.File; import org.apache.cassandra.io.util.SequentialWriterOption; import org.apache.cassandra.schema.TableId; @@@ -137,6 -138,8 +142,8 @@@ public class CassandraEntireSSTableStre prettyPrintMemory(totalSize)); } - validateDataDigest(writer.descriptor, manifest); ++ validateDataDigest(writer.descriptor(), manifest); + UnaryOperator<StatsMetadata> transform = stats -> stats.mutateLevel(header.sstableLevel) .mutateRepairedMetadata(messageHeader.repairedAt, messageHeader.pendingRepair, false); String description = String.format("level %s and repairedAt time %s and pendingRepair %s", diff --cc test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java index 700ee8b892,c36e83ce16..ce72ebbeda --- a/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java @@@ -19,14 -19,11 +19,16 @@@ package org.apache.cassandra.distributed.test.hostreplacement; import java.io.IOException; + import java.io.RandomAccessFile; +import java.net.InetAddress; import java.util.Arrays; +import java.util.LinkedHashSet; import java.util.List; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + import java.util.concurrent.TimeoutException; +import org.assertj.core.api.Assertions; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@@ -39,15 -38,12 +43,17 @@@ import org.apache.cassandra.distributed import org.apache.cassandra.distributed.api.IInvokableInstance; import org.apache.cassandra.distributed.api.SimpleQueryResult; import org.apache.cassandra.distributed.api.TokenSupplier; -import org.apache.cassandra.distributed.shared.AssertUtils; -import org.apache.cassandra.distributed.shared.ClusterUtils; +import org.apache.cassandra.distributed.shared.Uninterruptibles; import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.gms.ApplicationState; +import org.apache.cassandra.gms.Gossiper; + import org.apache.cassandra.io.sstable.format.SSTableFormat.Components; + import org.apache.cassandra.io.sstable.format.SSTableReader; -import org.assertj.core.api.Assertions; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.tcm.membership.NodeState; import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SKIP_SCHEMA_CHECK; import static org.apache.cassandra.config.CassandraRelevantProperties.GOSSIPER_QUARANTINE_DELAY; @@@ -231,6 -212,83 +237,83 @@@ public class HostReplacementTest extend } } + private static final int HEALTHY_PK = 10; + private static final int CORRUPT_PK = 11; + + /** + * With digest validation enabled, a single corrupted sstable must fail the entire stream session, so neither + * the corrupted sstable nor its healthy sibling should land on the replacing node. + */ + @Test + public void replaceHostAbortsStreamingWhenDigestValidationCatchesCorruption() throws IOException, TimeoutException + { + TokenSupplier even = TokenSupplier.evenlyDistributedTokens(2); + try (Cluster cluster = Cluster.build(2) + .withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK) + .set("entire_sstable_stream_digest_validation_enabled", true)) + .withTokenSupplier(node -> even.token(node == 3 ? 2 : node)) + .start()) + { + IInvokableInstance node1 = cluster.get(1); + IInvokableInstance nodeToRemove = cluster.get(2); + + setupCluster(cluster); + writeHealthyAndCorruptSSTables(cluster, node1); + + stopUnchecked(nodeToRemove); + + IInvokableInstance replacingNode = replaceHostAndStart(cluster, nodeToRemove); + + replacingNode.logs().watchFor("Digest mismatch"); + + // the whole stream session is aborted on a digest mismatch, so neither sstable should have landed - AssertUtils.assertRows(replacingNode.executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = " + HEALTHY_PK), EMPTY_ROWS); - AssertUtils.assertRows(replacingNode.executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = " + CORRUPT_PK), EMPTY_ROWS); ++ assertRows(replacingNode.executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = " + HEALTHY_PK), EMPTY_ROWS); ++ assertRows(replacingNode.executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = " + CORRUPT_PK), EMPTY_ROWS); + } + } + + /** + * Flushes two separate sstables on {@code node1} for {@code KEYSPACE.tbl}: one containing {@link #HEALTHY_PK} + * left untouched, and one containing {@link #CORRUPT_PK} that gets its Data.db corrupted. + */ + static void writeHealthyAndCorruptSSTables(Cluster cluster, IInvokableInstance node1) throws IOException + { + node1.nodetoolResult("disableautocompaction").asserts().success(); + + cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk) VALUES (?)", + ConsistencyLevel.ALL, + HEALTHY_PK); + node1.flush(KEYSPACE); + + cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk) VALUES (?)", + ConsistencyLevel.ALL, + CORRUPT_PK); + node1.flush(KEYSPACE); + + corruptSSTable(node1, "tbl"); + } + + /** Flips a byte in the middle of the Data.db file of the most recently flushed sstable for the given table. */ + static void corruptSSTable(IInvokableInstance instance, String table) throws IOException + { + String dataFilePath = instance.callOnInstance(() -> { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(table); + SSTableReader sstable = cfs.getLiveSSTables().stream() + .max(SSTableReader.maxTimestampAscending) + .orElseThrow(); + return sstable.descriptor.fileFor(Components.DATA).absolutePath(); + }); + + try (RandomAccessFile raf = new RandomAccessFile(dataFilePath, "rw")) + { + long corruptAt = raf.length() / 2; + raf.seek(corruptAt); + int b = raf.read(); + raf.seek(corruptAt); + raf.write(b ^ 0xFF); + } + } + static void setupCluster(Cluster cluster) { fixDistributedSchemas(cluster); diff --cc test/unit/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamWriterTest.java index 3cb982b89b,dcafae0058..a3324732a4 --- a/test/unit/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamWriterTest.java +++ b/test/unit/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamWriterTest.java @@@ -27,15 -30,24 +30,18 @@@ import org.junit.AfterClass import org.junit.BeforeClass; import org.junit.Test; -import io.netty.buffer.ByteBuf; -import io.netty.buffer.Unpooled; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelOutboundHandlerAdapter; -import io.netty.channel.ChannelPromise; -import io.netty.channel.DefaultFileRegion; -import io.netty.channel.embedded.EmbeddedChannel; 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.db.lifecycle.StreamingLifecycleTransaction; + import org.apache.cassandra.io.sstable.Component; import org.apache.cassandra.io.sstable.Descriptor; -import org.apache.cassandra.io.sstable.SSTableMultiWriter; +import org.apache.cassandra.io.sstable.SSTableTxnSingleStreamWriter; + import org.apache.cassandra.io.sstable.format.SSTableFormat.Components; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.DataInputBuffer; import org.apache.cassandra.locator.InetAddressAndPort; @@@ -56,17 -68,10 +62,19 @@@ import org.apache.cassandra.streaming.m import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; +import io.netty.channel.DefaultFileRegion; +import io.netty.channel.embedded.EmbeddedChannel; + +import static java.util.Collections.emptyList; import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID; + import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.junit.Assert.assertEquals; + import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; public class CassandraEntireSSTableStreamWriterTest @@@ -173,6 -186,124 +190,124 @@@ } } + @Test + public void testCorruptDataComponentFailsDigestValidation() throws Throwable + { + ByteBuf serializedFile = Unpooled.buffer(8192); + ComponentManifest manifest = writeEntireSSTable(serializedFile); + corruptDataComponent(serializedFile, manifest); + + DatabaseDescriptor.setEntireSSTableStreamDigestValidationEnabled(true); + assertThatThrownBy(() -> readEntireSSTable(serializedFile, manifest)) + .isInstanceOf(IOException.class) + .hasMessageContaining("Digest mismatch") + .hasCauseInstanceOf(IOException.class); + } + + @Test + public void testCorruptDataComponentAcceptedWhenValidationDisabled() throws Throwable + { + ByteBuf serializedFile = Unpooled.buffer(8192); + ComponentManifest manifest = writeEntireSSTable(serializedFile); + corruptDataComponent(serializedFile, manifest); + + DatabaseDescriptor.setEntireSSTableStreamDigestValidationEnabled(false); + + // shouldn't throw an error - SSTableMultiWriter writer = readEntireSSTable(serializedFile, manifest); ++ SSTableTxnSingleStreamWriter writer = readEntireSSTable(serializedFile, manifest); + assertNotNull(writer); + writer.abort(null); + } + + @Test + public void testMissingDigestIsAcceptedWithoutValidation() throws Throwable + { + DatabaseDescriptor.setEntireSSTableStreamDigestValidationEnabled(true); + + ComponentManifest full; + try (ComponentContext context = ComponentContext.create(sstable)) + { + full = context.manifest(); + } + + // remove digest component from sstable + LinkedHashMap<Component, Long> withoutDigest = new LinkedHashMap<>(); + for (Component component : full.components()) + { + if (!component.equals(Components.DIGEST)) + withoutDigest.put(component, full.sizeOf(component)); + } + ComponentManifest manifest = new ComponentManifest(withoutDigest); + + // stream from live files + ByteBuf serializedFile = Unpooled.buffer(8192); + for (Component component : manifest.components()) + { + try (InputStream in = descriptor.fileFor(component).newInputStream()) + { + serializedFile.writeBytes(in.readAllBytes()); + } + } + + // shouldn't throw an error - SSTableMultiWriter writer = readEntireSSTable(serializedFile, manifest); ++ SSTableTxnSingleStreamWriter writer = readEntireSSTable(serializedFile, manifest); + assertNotNull(writer); + writer.abort(null); + } + + private ComponentManifest writeEntireSSTable(ByteBuf serializedFile) throws Exception + { + StreamSession session = setupStreamingSessionForTest(); + EmbeddedChannel channel = createMockNettyChannel(serializedFile); + try (AsyncStreamingOutputPlus out = new AsyncStreamingOutputPlus(channel); + ComponentContext context = ComponentContext.create(sstable)) + { + new CassandraEntireSSTableStreamWriter(sstable, session, context).write(out); + return context.manifest(); + } + } + - private SSTableMultiWriter readEntireSSTable(ByteBuf serializedFile, ComponentManifest manifest) throws Throwable ++ private SSTableTxnSingleStreamWriter readEntireSSTable(ByteBuf serializedFile, ComponentManifest manifest) throws Throwable + { + StreamSession session = setupStreamingSessionForTest(); + InetAddressAndPort peer = FBUtilities.getBroadcastAddressAndPort(); - session.prepareReceiving(new StreamSummary(sstable.metadata().id, 1, 5104)); ++ session.prepareReceiving(new StreamSummary(sstable.metadata().id, emptyList(), 1, 5104)); + + CassandraStreamHeader header = + CassandraStreamHeader.builder() + .withSSTableVersion(sstable.descriptor.version) + .withSSTableLevel(0) + .withEstimatedKeys(sstable.estimatedKeys()) - .withSections(Collections.emptyList()) ++ .withSections(emptyList()) + .withSerializationHeader(sstable.header.toComponent()) + .withComponentManifest(manifest) + .isEntireSSTable(true) + .withFirstKey(sstable.getFirst()) + .withTableId(sstable.metadata().id) + .build(); + + CassandraEntireSSTableStreamReader reader = + new CassandraEntireSSTableStreamReader(new StreamMessageHeader(sstable.metadata().id, peer, session.planId(), false, 0, 0, 0, null), + header, session); + - return reader.read(new DataInputBuffer(serializedFile.nioBuffer(), false)); ++ return (SSTableTxnSingleStreamWriter) reader.read(new DataInputBuffer(serializedFile.nioBuffer(), false)); + } + + /** Flips every bit of one byte in the middle of the Data.db region of an already-serialized stream. */ + private static void corruptDataComponent(ByteBuf serializedFile, ComponentManifest manifest) + { + long dataOffset = 0; + for (Component component : manifest.components()) + { + if (component.equals(Components.DATA)) + break; + dataOffset += manifest.sizeOf(component); + } + + int corruptAt = (int) (dataOffset + manifest.sizeOf(Components.DATA) / 2); + serializedFile.setByte(corruptAt, serializedFile.getByte(corruptAt) ^ 0xff); + } + private EmbeddedChannel createMockNettyChannel(ByteBuf serializedFile) throws Exception { WritableByteChannel wbc = new WritableByteChannel() --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
