You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by bd...@apache.org on 2026/04/14 21:47:20 UTC
(cassandra) 31/36: CEP-45: Add feature flag for Mutation Tracking
This is an automated email from the ASF dual-hosted git repository.
bdeggleston pushed a commit to branch cep-45-mutation-tracking
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 1c10de236fdc76a4769926155db01b9374d973ce
Author: Blake Eggleston <bl...@ultrablake.com>
AuthorDate: Thu Sep 25 12:06:18 2025 -0700
CEP-45: Add feature flag for Mutation Tracking
patch by Caleb Rackliffe; reviewed by Blake Eggleston and Sam Tunnicliffe for CASSANDRA-20926
Co-authored-by: Blake Eggleston <bl...@ultrablake.com>
Co-authored-by: Caleb Rackliffe <ca...@gmail.com>
---
.../config/CassandraRelevantProperties.java | 1 +
src/java/org/apache/cassandra/config/Config.java | 2 +-
.../cassandra/config/DatabaseDescriptor.java | 22 ++-
.../cassandra/config/MutationTrackingSpec.java | 25 +++
.../cql3/statements/ModificationStatement.java | 2 +-
.../statements/schema/AlterKeyspaceStatement.java | 27 ++-
.../statements/schema/CreateKeyspaceStatement.java | 40 ++++-
.../org/apache/cassandra/db/ColumnFamilyStore.java | 10 +-
src/java/org/apache/cassandra/db/Keyspace.java | 5 +-
.../cassandra/db/PartitionRangeReadCommand.java | 2 +-
.../org/apache/cassandra/db/SSTableImporter.java | 2 +-
.../cassandra/db/SinglePartitionReadCommand.java | 2 +-
.../org/apache/cassandra/db/SystemKeyspace.java | 60 +++----
.../cassandra/db/compaction/CompactionTask.java | 2 +-
.../db/streaming/CassandraStreamReceiver.java | 2 +-
.../db/tracked/TrackedKeyspaceWriteHandler.java | 2 +-
.../cassandra/db/virtual/MutationJournalTable.java | 80 ---------
.../db/virtual/MutationTrackingShardsTable.java | 110 ------------
.../db/virtual/MutationTrackingTables.java | 186 +++++++++++++++++++++
.../cassandra/db/virtual/SystemViewsKeyspace.java | 3 +-
.../cassandra/io/sstable/format/SSTableWriter.java | 2 +-
.../metrics/CassandraMetricsRegistry.java | 106 ++++++------
.../cassandra/metrics/MutationTrackingMetrics.java | 16 +-
.../org/apache/cassandra/repair/SyncTasks.java | 2 +-
.../cassandra/replication/ActivationRequest.java | 2 +-
.../cassandra/replication/ActiveLogReconciler.java | 12 +-
.../cassandra/replication/BroadcastLogOffsets.java | 3 +-
.../cassandra/replication/CoordinatedTransfer.java | 2 +-
.../cassandra/replication/CoordinatorLog.java | 4 +-
.../cassandra/replication/ForwardedWrite.java | 7 +-
.../cassandra/replication/MutationJournal.java | 33 +++-
.../replication/MutationTrackingService.java | 76 +++++++--
.../replication/PullMutationsRequest.java | 3 +-
.../cassandra/replication/PushMutationRequest.java | 5 +-
.../cassandra/replication/TrackedWriteRequest.java | 4 +-
.../replication/UnreconciledMutations.java | 2 +-
.../apache/cassandra/service/CassandraDaemon.java | 6 +-
.../org/apache/cassandra/service/StorageProxy.java | 4 +-
.../apache/cassandra/service/StorageService.java | 4 +-
.../service/TrackedWriteResponseHandler.java | 4 +-
.../cassandra/service/paxos/PaxosCommit.java | 12 +-
.../service/paxos/PaxosPrepareRefresh.java | 2 +-
.../apache/cassandra/service/paxos/PaxosState.java | 4 +-
.../paxos/PrepareRefreshForwardHandler.java | 2 +-
.../service/reads/tracked/PartialTrackedRead.java | 2 +-
.../service/reads/tracked/ReadReconcileAck.java | 2 +
.../service/reads/tracked/ReadReconciliations.java | 8 +-
.../service/reads/tracked/TrackedRead.java | 7 +-
.../reads/tracked/TrackedSummaryResponse.java | 2 +
.../apache/cassandra/streaming/StreamSession.java | 6 +-
.../messages/IncomingMutationLogStreamMessage.java | 2 +-
src/java/org/apache/cassandra/tcm/Startup.java | 24 ++-
test/conf/cassandra-jmx-disabled-sslconfig.yaml | 6 +-
test/conf/cassandra-jmx-pem-sslconfig.yaml | 6 +-
.../cassandra-jmx-sslconfig-with-passwordfile.yaml | 6 +-
test/conf/cassandra-jmx-sslconfig.yaml | 6 +-
test/conf/cassandra-mtls.yaml | 2 +
test/conf/cassandra-murmur.yaml | 2 +
test/conf/cassandra-old.yaml | 4 +-
test/conf/cassandra-pem-jks-sslcontextfactory.yaml | 2 +
...pem-sslcontextfactory-invalidconfiguration.yaml | 2 +
test/conf/cassandra-pem-sslcontextfactory.yaml | 2 +
test/conf/cassandra-seeds.yaml | 2 +
...dra-sslcontextfactory-invalidconfiguration.yaml | 2 +
test/conf/cassandra-sslcontextfactory.yaml | 2 +
test/conf/cassandra.yaml | 4 +
.../cassandra/distributed/impl/Instance.java | 10 +-
.../cassandra/distributed/impl/InstanceConfig.java | 9 +
.../test/MutationTrackingMigrationTest.java | 2 +-
.../distributed/test/TrackedBatchTest.java | 6 +-
.../test/TransientRangeMovement2Test.java | 3 -
.../test/TransientRangeMovementTest.java | 3 -
.../test/WitnessAlwaysReadsFullReplicaTest.java | 1 -
...MultiNodeTableWalkWithMutationTrackingTest.java | 1 -
.../TrackedHostReplacementTest.java | 4 +-
.../test/log/MetadataChangeSimulationTest.java | 1 -
.../distributed/test/log/NTSSimulationTest.java | 1 -
.../test/log/OperationalEquivalenceTest.java | 1 -
.../test/log/SimpleStrategySimulationTest.java | 1 -
.../replication/MutationTrackingServiceTest.java | 2 +-
.../test/tracking/MutationTrackingBounceTest.java | 2 +-
.../MutationTrackingBounce_ValidateRunnable.java | 4 +-
.../MutationTrackingCasForwardingTest.java | 9 +-
.../MutationTrackingCommitAndPrepareTest.java | 8 +-
.../tracking/MutationTrackingFeatureFlagTest.java | 116 +++++++++++++
.../tracking/MutationTrackingLogPersisterTest.java | 13 +-
.../test/tracking/MutationTrackingMetricsTest.java | 50 +++---
.../MutationTrackingPartitionReadTest.java | 1 -
.../tracking/MutationTrackingPendingReadTest.java | 12 +-
.../tracking/MutationTrackingRangeReadTest.java | 4 +-
.../MutationTrackingReadReconciliationTest.java | 20 +--
.../test/tracking/MutationTrackingTest.java | 52 +++---
.../test/tracking/MutationTrackingUtils.java | 6 +-
.../MutationTrackingWriteForwardingTest.java | 9 +-
.../test/tracking/OffsetBroadcastTest.java | 8 +-
.../test/tracking/TrackedTransferTestBase.java | 7 +-
.../db/commitlog/CommitLogStressTest.java | 2 +-
.../cassandra/simulator/ClusterSimulation.java | 2 +-
.../unit/org/apache/cassandra/ServerTestUtils.java | 2 +-
.../config/DatabaseDescriptorRefTest.java | 1 +
.../apache/cassandra/db/CleanupTransientTest.java | 5 +-
.../db/CoordinatorLogOffsetsLifecycleTest.java | 6 +-
...onTrackingMutationVerbHandlerMigrationTest.java | 3 +-
.../db/MutationVerbHandlerOutOfRangeTest.java | 3 +-
.../db/ReadCommandVerbHandlerOutOfRangeTest.java | 2 +-
.../cassandra/db/ReadCommandVerbHandlerTest.java | 2 +-
.../db/virtual/MutationJournalTableTest.java | 4 +-
.../virtual/MutationTrackingShardsTableTest.java | 8 +-
.../org/apache/cassandra/dht/BootStrapperTest.java | 2 +-
.../cassandra/locator/SimpleStrategyTest.java | 1 -
.../RepairMessageVerbHandlerOutOfRangeTest.java | 2 +-
.../replication/CoordinatorLogOffsetsTest.java | 58 +++----
.../cassandra/replication/CoordinatorLogTest.java | 6 +-
.../cassandra/replication/MutationJournalTest.java | 2 +-
.../apache/cassandra/replication/ShardTest.java | 2 +-
.../replication/UnreconciledMutationsTest.java | 6 +-
.../cassandra/service/BootstrapTransientTest.java | 1 -
.../cassandra/service/JoinTokenRingTest.java | 2 +-
.../apache/cassandra/service/PaxosStateTest.java | 2 +-
.../cassandra/service/StorageServiceTest.java | 2 +-
.../service/WriteResponseHandlerTransientTest.java | 3 +-
.../cassandra/service/paxos/PaxosStateTest.java | 2 +-
.../paxos/PaxosVerbHandlerOutOfRangeTest.java | 2 +-
.../reads/repair/RepairedDataVerifierTest.java | 2 +-
.../cassandra/streaming/StreamReaderTest.java | 2 +-
.../streaming/StreamSessionOwnedRangesTest.java | 2 +-
.../AlterSchemaMutationTrackingTest.java | 2 +-
127 files changed, 912 insertions(+), 598 deletions(-)
diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
index 72b8f0e1fb..b39a2f1cd1 100644
--- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
+++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
@@ -239,6 +239,7 @@ public enum CassandraRelevantProperties
DTEST_IS_IN_JVM_DTEST("org.apache.cassandra.dtest.is_in_jvm_dtest"),
/** In_JVM dtest property indicating that the test should use "latest" configuration */
DTEST_JVM_DTESTS_USE_LATEST("jvm_dtests.latest"),
+ DTEST_MUTATION_TRACKING_ENABLED("jvm_dtest.mutation_tracking.enabled", "true"),
ENABLE_DC_LOCAL_COMMIT("cassandra.enable_dc_local_commit", "true"),
/**
* Whether {@link org.apache.cassandra.db.ConsistencyLevel#NODE_LOCAL} should be allowed.
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index bc24ece29b..694737701a 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -695,7 +695,7 @@ public class Config
public boolean dynamic_data_masking_enabled = false;
- public boolean mutation_tracking_enabled = false;
+ public MutationTrackingSpec mutation_tracking = new MutationTrackingSpec();
/**
* Time in milliseconds after a warning will be emitted to the log and to the client that a UDF runs too long.
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 78b61f7ac0..1e7bca447d 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -773,6 +773,11 @@ public class DatabaseDescriptor
if (commitLogWriteDiskAccessMode != conf.commitlog_disk_access_mode)
logger.info("commitlog_disk_access_mode resolved to: {}", commitLogWriteDiskAccessMode);
+ if (conf.mutation_tracking.enabled && conf.mutation_tracking.journal_directory == null)
+ {
+ conf.mutation_tracking.journal_directory = storagedirFor("mutation_journal");
+ }
+
if (conf.accord.journal_directory == null)
{
conf.accord.journal_directory = storagedirFor("accord_journal");
@@ -2315,6 +2320,13 @@ public class DatabaseDescriptor
throw new ConfigurationException("commitlog_directory must be specified", false);
FileUtils.createDirectory(conf.commitlog_directory);
+ if (conf.mutation_tracking.enabled)
+ {
+ if (conf.mutation_tracking.journal_directory == null)
+ throw new ConfigurationException("mutation_tracking.journal_directory must be specified", false);
+ FileUtils.createDirectory(conf.mutation_tracking.journal_directory);
+ }
+
if (conf.accord.journal_directory == null)
throw new ConfigurationException("accord.journal_directory must be specified", false);
FileUtils.createDirectory(conf.accord.journal_directory);
@@ -6039,16 +6051,12 @@ public class DatabaseDescriptor
public static boolean getMutationTrackingEnabled()
{
- return conf.mutation_tracking_enabled;
+ return conf.mutation_tracking.enabled;
}
- public static void setMutationTrackingEnabled(boolean enabled)
+ public static String getMutationTrackingJournalDirectory()
{
- if (enabled != conf.mutation_tracking_enabled)
- {
- logger.info("Setting mutation_tracking_enabled to {}", enabled);
- conf.mutation_tracking_enabled = enabled;
- }
+ return conf.mutation_tracking.journal_directory;
}
public static OptionalDouble getSeverityDuringDecommission()
diff --git a/src/java/org/apache/cassandra/config/MutationTrackingSpec.java b/src/java/org/apache/cassandra/config/MutationTrackingSpec.java
new file mode 100644
index 0000000000..d9b04a0f40
--- /dev/null
+++ b/src/java/org/apache/cassandra/config/MutationTrackingSpec.java
@@ -0,0 +1,25 @@
+/*
+ * 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.config;
+
+public class MutationTrackingSpec
+{
+ public boolean enabled = false;
+ public String journal_directory;
+}
diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
index ba152c9be2..a19d2cb090 100644
--- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
@@ -860,7 +860,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
{
String keyspaceName = mutation.getKeyspaceName();
Token token = mutation.key().getToken();
- MutationId id = MutationTrackingService.instance.nextMutationId(keyspaceName, token);
+ MutationId id = MutationTrackingService.instance().nextMutationId(keyspaceName, token);
mutation = mutation.withMutationId(id);
mutation.apply();
}
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java
index 5ef2b68a9c..30faee655e 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java
@@ -40,6 +40,7 @@ import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.KeyspaceMetadata.KeyspaceDiff;
import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
+import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientState;
@@ -52,6 +53,7 @@ import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.config.CassandraRelevantProperties.ALLOW_ALTER_RF_DURING_RANGE_MOVEMENT;
import static org.apache.cassandra.config.CassandraRelevantProperties.ALLOW_UNSAFE_TRANSIENT_CHANGES;
+import static org.apache.cassandra.replication.MutationTrackingService.DISABLED_MESSAGE;
public final class AlterKeyspaceStatement extends AlterSchemaStatement
{
@@ -91,9 +93,6 @@ public final class AlterKeyspaceStatement extends AlterSchemaStatement
KeyspaceMetadata newKeyspace = keyspace.withSwapped(attrs.asAlteredKeyspaceParams(keyspace.params));
- if (attrs.getReplicationStrategyClass() != null && attrs.getReplicationStrategyClass().equals(SimpleStrategy.class.getSimpleName()))
- Guardrails.simpleStrategyEnabled.ensureEnabled(state);
-
if (keyspace.params.replication.isMeta() && !keyspace.name.equals(SchemaConstants.METADATA_KEYSPACE_NAME))
throw ire("Can not alter a keyspace to use MetaReplicationStrategy");
@@ -135,6 +134,28 @@ public final class AlterKeyspaceStatement extends AlterSchemaStatement
super.validate(state);
Guardrails.keyspaceProperties.guard(attrs.updatedProperties(), attrs::removeProperty, state);
+
+ KeyspaceMetadata keyspace = Schema.instance.getKeyspaceMetadata(keyspaceName);
+ if (null == keyspace)
+ {
+ if (!ifExists)
+ throw ire("Keyspace '%s' doesn't exist", keyspaceName);
+ return;
+ }
+
+ KeyspaceMetadata newKeyspace = keyspace.withSwapped(attrs.asAlteredKeyspaceParams(keyspace.params));
+
+ if (attrs.getReplicationStrategyClass() != null && attrs.getReplicationStrategyClass().equals(SimpleStrategy.class.getSimpleName()))
+ Guardrails.simpleStrategyEnabled.ensureEnabled(state);
+
+ if (newKeyspace.params.replicationType.isTracked() && !keyspace.params.replicationType.isTracked())
+ {
+ if (SchemaConstants.isSystemKeyspace(keyspaceName))
+ throw ire("Mutation tracking is not supported on system keyspaces");
+
+ if (!DatabaseDescriptor.getMutationTrackingEnabled())
+ throw ire(DISABLED_MESSAGE);
+ }
}
@Override
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java
index a622b86192..9b02c04bb8 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java
@@ -27,6 +27,7 @@ import org.apache.cassandra.auth.DataResource;
import org.apache.cassandra.auth.FunctionResource;
import org.apache.cassandra.auth.IResource;
import org.apache.cassandra.auth.Permission;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLStatement;
import org.apache.cassandra.db.guardrails.Guardrails;
import org.apache.cassandra.exceptions.AlreadyExistsException;
@@ -39,12 +40,15 @@ import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.transport.Event.SchemaChange;
import org.apache.cassandra.transport.Event.SchemaChange.Change;
+import static org.apache.cassandra.replication.MutationTrackingService.DISABLED_MESSAGE;
+
public final class CreateKeyspaceStatement extends AlterSchemaStatement
{
private final KeyspaceAttributes attrs;
@@ -75,14 +79,6 @@ public final class CreateKeyspaceStatement extends AlterSchemaStatement
@Override
public Keyspaces apply(ClusterMetadata metadata)
{
- attrs.validate();
-
- if (!attrs.hasOption(Option.REPLICATION))
- throw ire("Missing mandatory option '%s'", Option.REPLICATION);
-
- if (attrs.getReplicationStrategyClass() != null && attrs.getReplicationStrategyClass().equals(SimpleStrategy.class.getSimpleName()))
- Guardrails.simpleStrategyEnabled.ensureEnabled("SimpleStrategy", state);
-
Keyspaces schema = metadata.schema.getKeyspaces();
if (schema.containsKeyspace(keyspaceName))
{
@@ -146,10 +142,38 @@ public final class CreateKeyspaceStatement extends AlterSchemaStatement
{
super.validate(state);
+ attrs.validate();
+
// Guardrail on number of keyspaces
Guardrails.keyspaces.guard(Schema.instance.getUserKeyspaces().size() + 1, keyspaceName, false, state);
Guardrails.keyspaceProperties.guard(attrs.updatedProperties(), attrs::removeProperty, state);
+
+ if (!attrs.hasOption(Option.REPLICATION))
+ throw ire("Missing mandatory option '%s'", Option.REPLICATION);
+
+ if (attrs.getReplicationStrategyClass() != null && attrs.getReplicationStrategyClass().equals(SimpleStrategy.class.getSimpleName()))
+ Guardrails.simpleStrategyEnabled.ensureEnabled("SimpleStrategy", state);
+
+ KeyspaceMetadata keyspace = Schema.instance.getKeyspaceMetadata(keyspaceName);
+ if (keyspace != null)
+ {
+ if (ifNotExists)
+ return;
+
+ throw new AlreadyExistsException(keyspaceName);
+ }
+
+ KeyspaceMetadata keyspaceMetadata = KeyspaceMetadata.create(keyspaceName, attrs.asNewKeyspaceParams());
+
+ if (keyspaceMetadata.params.replicationType.isTracked())
+ {
+ if (SchemaConstants.isSystemKeyspace(keyspaceName))
+ throw ire("Mutation tracking is not supported on system keyspaces");
+
+ if (!DatabaseDescriptor.getMutationTrackingEnabled())
+ throw ire(DISABLED_MESSAGE);
+ }
}
public static final class Raw extends CQLStatement.Raw
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index fa4f493f76..1626867245 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -524,7 +524,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
{
CommitLogPosition commitLogPosition;
if (metadata().replicationType().isTracked())
- commitLogPosition = MutationJournal.instance.getCurrentPosition();
+ commitLogPosition = MutationJournal.instance().getCurrentPosition();
else
commitLogPosition = CommitLog.instance.getCurrentPosition();
initialMemtable = createMemtable(new AtomicReference<>(commitLogPosition));
@@ -1196,7 +1196,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
commitLogUpperBound = mainMemtable.getFinalCommitLogUpperBound();
TableMetadata metadata = metadata();
- MutationJournal.instance.notifyFlushed(metadata.id, commitLogLowerBound, commitLogUpperBound);
+ if (metadata().replicationType().isTracked())
+ MutationJournal.instance().notifyFlushed(metadata.id, commitLogLowerBound, commitLogUpperBound);
+
CommitLog.instance.discardCompletedSegments(metadata.id, commitLogLowerBound, commitLogUpperBound);
}
@@ -1475,7 +1477,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
{
CommitLogPosition commitLogPosition;
if (useMutationJournal)
- commitLogPosition = MutationJournal.instance.getCurrentPosition();
+ commitLogPosition = MutationJournal.instance().getCurrentPosition();
else
commitLogPosition = CommitLog.instance.getCurrentPosition();
@@ -3397,7 +3399,7 @@ public <T> T withAllSSTables(final OperationType operationType, Function<Lifecyc
// TODO (required): test mutation tracking + table dropping
if (metadata().replicationType().isTracked())
- MutationJournal.instance.notifyFlushed(metadata.id, new CommitLogPosition(0, 0), MutationJournal.instance.getCurrentPosition());
+ MutationJournal.instance().notifyFlushed(metadata.id, new CommitLogPosition(0, 0), MutationJournal.instance().getCurrentPosition());
else
CommitLog.instance.forceRecycleAllSegments(Collections.singleton(metadata.id));
diff --git a/src/java/org/apache/cassandra/db/Keyspace.java b/src/java/org/apache/cassandra/db/Keyspace.java
index 79fee4ce31..78b344947b 100644
--- a/src/java/org/apache/cassandra/db/Keyspace.java
+++ b/src/java/org/apache/cassandra/db/Keyspace.java
@@ -616,6 +616,7 @@ public class Keyspace
*/
private Future<?> applyInternalTracked(Mutation mutation, Promise<?> future)
{
+ MutationTrackingService.ensureEnabled();
Preconditions.checkState(MigrationRouter.isFullyTracked(mutation) && !mutation.id().isNone());
ClusterMetadata cm = ClusterMetadata.current();
@@ -625,7 +626,7 @@ public class Keyspace
boolean started;
try (WriteContext ctx = trackedWriteHandler.beginWrite(mutation, true))
{
- started = MutationTrackingService.instance.startWriting(mutation);
+ started = MutationTrackingService.instance().startWriting(mutation);
if (started)
{
@@ -675,7 +676,7 @@ public class Keyspace
}
if (started)
- MutationTrackingService.instance.finishWriting(mutation);
+ MutationTrackingService.instance().finishWriting(mutation);
if (future != null)
diff --git a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
index f3d6b01eb5..0293bc238c 100644
--- a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
+++ b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
@@ -460,7 +460,7 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
@Override
protected MutationSummary createMutationSummaryInternal(boolean includePending)
{
- return MutationTrackingService.instance.createSummaryForRange(dataRange.keyRange, metadata().id, includePending);
+ return MutationTrackingService.instance().createSummaryForRange(dataRange.keyRange, metadata().id, includePending);
}
@Override
diff --git a/src/java/org/apache/cassandra/db/SSTableImporter.java b/src/java/org/apache/cassandra/db/SSTableImporter.java
index fd6d796dc5..b8e264ad11 100644
--- a/src/java/org/apache/cassandra/db/SSTableImporter.java
+++ b/src/java/org/apache/cassandra/db/SSTableImporter.java
@@ -236,7 +236,7 @@ public class SSTableImporter
cfs.indexManager.buildSSTableAttachedIndexesBlocking(newSSTables);
if (isTracked)
- MutationTrackingService.instance.executeTransfers(cfs.keyspace.getName(), newSSTables, ConsistencyLevel.ALL);
+ MutationTrackingService.instance().executeTransfers(cfs.keyspace.getName(), newSSTables, ConsistencyLevel.ALL);
else
cfs.getTracker().addSSTables(newSSTables);
diff --git a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
index 8867197dfc..378f834ee4 100644
--- a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
+++ b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
@@ -941,7 +941,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
protected MutationSummary createMutationSummaryInternal(boolean includePending)
{
- return MutationTrackingService.instance.createSummaryForKey(partitionKey, metadata().id, includePending);
+ return MutationTrackingService.instance().createSummaryForKey(partitionKey, metadata().id, includePending);
}
@Override
diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java
index 3cea8e3079..e3716955c4 100644
--- a/src/java/org/apache/cassandra/db/SystemKeyspace.java
+++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java
@@ -650,35 +650,37 @@ public final class SystemKeyspace
private static Tables tables()
{
- return Tables.of(BuiltIndexes,
- Batches,
- Paxos,
- PaxosRepairHistoryTable,
- Local,
- PeersV2,
- LegacyPeers,
- PeerEventsV2,
- LegacyPeerEvents,
- CompactionHistory,
- LegacySSTableActivity,
- SSTableActivity,
- LegacySizeEstimates,
- TableEstimates,
- AvailableRangesV2,
- LegacyAvailableRanges,
- TransferredRangesV2,
- LegacyTransferredRanges,
- ViewBuildsInProgress,
- BuiltViews,
- PreparedStatements,
- Repairs,
- TopPartitions,
- LocalMetadataLog,
- Snapshots,
- ConsensusMigrationState,
- HostLogId,
- Shards,
- CoordinatorLogs);
+ Tables.Builder builder = Tables.builder().add(BuiltIndexes,
+ Batches,
+ Paxos,
+ PaxosRepairHistoryTable,
+ Local,
+ PeersV2,
+ LegacyPeers,
+ PeerEventsV2,
+ LegacyPeerEvents,
+ CompactionHistory,
+ LegacySSTableActivity,
+ SSTableActivity,
+ LegacySizeEstimates,
+ TableEstimates,
+ AvailableRangesV2,
+ LegacyAvailableRanges,
+ TransferredRangesV2,
+ LegacyTransferredRanges,
+ ViewBuildsInProgress,
+ BuiltViews,
+ PreparedStatements,
+ Repairs,
+ TopPartitions,
+ LocalMetadataLog,
+ Snapshots,
+ ConsensusMigrationState);
+
+ if (DatabaseDescriptor.getMutationTrackingEnabled())
+ builder.add(HostLogId, Shards, CoordinatorLogs);
+
+ return builder.build();
}
private static volatile Map<TableId, Pair<CommitLogPosition, Long>> truncationRecords;
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
index d052c7d3d8..7e41d980de 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
@@ -448,7 +448,7 @@ public class CompactionTask extends AbstractCompactionTask
ImmutableCoordinatorLogOffsets.Builder builder = new ImmutableCoordinatorLogOffsets.Builder();
for (SSTableReader sstable : sstables)
builder.addAll(sstable.getCoordinatorLogOffsets());
- builder.purgeTransfers(MutationTrackingService.instance::isDurablyReconciled);
+ builder.purgeTransfers(id -> MutationTrackingService.instance().isDurablyReconciled(id));
return builder.build();
}
diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java
index edf154989e..e78ddd77ff 100644
--- a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java
+++ b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java
@@ -139,7 +139,7 @@ public class CassandraStreamReceiver implements StreamReceiver
if (cfs.metadata().replicationType().isTracked() && session.streamOperation().isTrackable())
{
PendingLocalTransfer transfer = new PendingLocalTransfer(cfs.metadata().id, session.planId(), sstables);
- MutationTrackingService.instance.received(transfer);
+ MutationTrackingService.instance().received(transfer);
}
}
diff --git a/src/java/org/apache/cassandra/db/tracked/TrackedKeyspaceWriteHandler.java b/src/java/org/apache/cassandra/db/tracked/TrackedKeyspaceWriteHandler.java
index 50470f4aff..6d8ed90149 100644
--- a/src/java/org/apache/cassandra/db/tracked/TrackedKeyspaceWriteHandler.java
+++ b/src/java/org/apache/cassandra/db/tracked/TrackedKeyspaceWriteHandler.java
@@ -42,7 +42,7 @@ public class TrackedKeyspaceWriteHandler implements KeyspaceWriteHandler
MigrationRouter.validateTrackedMutation(mutation);
Tracing.trace("Appending to mutation journal");
- CommitLogPosition pointer = MutationJournal.instance.write(mutation.id(), mutation);
+ CommitLogPosition pointer = MutationJournal.instance().write(mutation.id(), mutation);
return new CassandraWriteContext(group, pointer);
}
diff --git a/src/java/org/apache/cassandra/db/virtual/MutationJournalTable.java b/src/java/org/apache/cassandra/db/virtual/MutationJournalTable.java
deleted file mode 100644
index 603e6d521b..0000000000
--- a/src/java/org/apache/cassandra/db/virtual/MutationJournalTable.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * 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.virtual;
-
-import org.apache.cassandra.db.Mutation;
-import org.apache.cassandra.db.marshal.BooleanType;
-import org.apache.cassandra.db.marshal.Int32Type;
-import org.apache.cassandra.db.marshal.LongType;
-import org.apache.cassandra.db.marshal.UTF8Type;
-import org.apache.cassandra.dht.LocalPartitioner;
-import org.apache.cassandra.journal.ActiveSegment;
-import org.apache.cassandra.journal.Segment;
-import org.apache.cassandra.replication.MutationJournal;
-import org.apache.cassandra.replication.ShortMutationId;
-import org.apache.cassandra.schema.TableMetadata;
-
-final class MutationJournalTable extends AbstractVirtualTable
-{
- private static final String SEGMENT_ID = "segment_id";
- private static final String IS_ACTIVE = "is_active";
- private static final String BYTES_ON_DISK = "bytes_on_disk";
- private static final String RECORDS_COUNT = "records_count";
- private static final String WRITTEN_TO = "written_to";
- private static final String FSYNCED_TO = "fsynced_to";
- private static final String NEEDS_REPLAY = "needs_replay";
- private static final String FILE_PATH = "file_path";
-
- MutationJournalTable(String keyspace)
- {
- super(TableMetadata.builder(keyspace, "mutation_journal")
- .comment("mutation journal segments and their contents")
- .kind(TableMetadata.Kind.VIRTUAL)
- .partitioner(new LocalPartitioner(LongType.instance))
- .addPartitionKeyColumn(SEGMENT_ID, LongType.instance)
- .addRegularColumn(IS_ACTIVE, BooleanType.instance)
- .addRegularColumn(BYTES_ON_DISK, LongType.instance)
- .addRegularColumn(RECORDS_COUNT, Int32Type.instance)
- .addRegularColumn(WRITTEN_TO, Int32Type.instance)
- .addRegularColumn(FSYNCED_TO, Int32Type.instance)
- .addRegularColumn(NEEDS_REPLAY, BooleanType.instance)
- .addRegularColumn(FILE_PATH, UTF8Type.instance)
- .build());
- }
-
- @Override
- public DataSet data()
- {
- SimpleDataSet result = new SimpleDataSet(metadata());
-
- for (Segment<ShortMutationId, Mutation> segment : MutationJournal.instance.getAllSegments())
- {
- result.row(segment.id())
- .column(IS_ACTIVE, segment instanceof ActiveSegment)
- .column(BYTES_ON_DISK, segment.segmentSizeOnDisk())
- .column(RECORDS_COUNT, segment.metadata().totalCount())
- .column(WRITTEN_TO, segment.writtenTo())
- .column(FSYNCED_TO, segment.fsyncedTo())
- .column(NEEDS_REPLAY, segment.metadata().needsReplay())
- .column(FILE_PATH, segment.filePath());
- }
-
- return result;
- }
-}
diff --git a/src/java/org/apache/cassandra/db/virtual/MutationTrackingShardsTable.java b/src/java/org/apache/cassandra/db/virtual/MutationTrackingShardsTable.java
deleted file mode 100644
index 724b6a596d..0000000000
--- a/src/java/org/apache/cassandra/db/virtual/MutationTrackingShardsTable.java
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
- * 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.virtual;
-
-import org.apache.cassandra.db.DecoratedKey;
-import org.apache.cassandra.db.marshal.Int32Type;
-import org.apache.cassandra.db.marshal.UTF8Type;
-import org.apache.cassandra.dht.LocalPartitioner;
-import org.apache.cassandra.replication.CoordinatorLog;
-import org.apache.cassandra.replication.CoordinatorLogId;
-import org.apache.cassandra.replication.MutationTrackingService;
-import org.apache.cassandra.replication.Shard;
-import org.apache.cassandra.schema.TableMetadata;
-
-import java.util.Map;
-
-public class MutationTrackingShardsTable extends AbstractVirtualTable
-{
- private static final String KEYSPACE = "keyspace";
- private static final String LOG_ID = "log_id";
- private static final String RANGE_START = "range_start";
- private static final String RANGE_END = "range_end";
- private static final String LOCAL_NODE_ID = "local_node_id";
- private static final String PARTICIPANTS = "participants";
- private static final String WITNESSED_OFFSETS = "witnessed_offsets";
- private static final String RECONCILED_OFFSETS = "reconciled_offsets";
- private static final String PERSISTED_OFFSETS = "persisted_offsets";
-
- MutationTrackingShardsTable(String keyspace) {
- super(TableMetadata.builder(keyspace, "mutation_tracking_shards")
- .comment("mutation tracking shards and their offset information")
- .kind(TableMetadata.Kind.VIRTUAL).partitioner(new LocalPartitioner(UTF8Type.instance))
- .addPartitionKeyColumn(KEYSPACE, UTF8Type.instance)
- .addClusteringColumn(LOG_ID, UTF8Type.instance)
- .addClusteringColumn(RANGE_START, UTF8Type.instance)
- .addClusteringColumn(RANGE_END, UTF8Type.instance)
- .addRegularColumn(LOCAL_NODE_ID, Int32Type.instance)
- .addRegularColumn(PARTICIPANTS, UTF8Type.instance)
- .addRegularColumn(WITNESSED_OFFSETS, UTF8Type.instance)
- .addRegularColumn(RECONCILED_OFFSETS, UTF8Type.instance)
- .addRegularColumn(PERSISTED_OFFSETS, UTF8Type.instance)
- .build());
- }
-
- private void addShardRows(Shard shard, SimpleDataSet result)
- {
- Shard.DebugInfo shardDebugInfo = shard.getDebugInfo();
- for (Map.Entry<CoordinatorLogId, CoordinatorLog.DebugInfo> entry : shardDebugInfo.logs.entrySet())
- {
- CoordinatorLogId logId = entry.getKey();
- CoordinatorLog.DebugInfo logDebugInfo = entry.getValue();
- result.row(shardDebugInfo.keyspace,
- logId.toString(),
- shardDebugInfo.range.left.toString(),
- shardDebugInfo.range.right.toString())
- .column(LOCAL_NODE_ID, shardDebugInfo.localNodeId)
- .column(PARTICIPANTS, shardDebugInfo.participants.toString())
- .column(WITNESSED_OFFSETS, logDebugInfo.witnessedOffsets)
- .column(RECONCILED_OFFSETS, logDebugInfo.reconciledOffsets)
- .column(PERSISTED_OFFSETS, logDebugInfo.persistedOffsets);
- }
- }
-
- @Override
- public DataSet data()
- {
- SimpleDataSet result = new SimpleDataSet(metadata());
-
- for (Shard shard : MutationTrackingService.instance.getShards())
- {
- addShardRows(shard, result);
- }
-
- return result;
- }
-
- @Override
- public DataSet data(DecoratedKey key)
- {
- String keyspaceName = UTF8Type.instance.compose(key.getKey());
- SimpleDataSet result = new SimpleDataSet(metadata());
-
- for (Shard shard : MutationTrackingService.instance.getShards())
- {
- Shard.DebugInfo debugInfo = shard.getDebugInfo();
- if (!debugInfo.keyspace.equals(keyspaceName))
- continue;
-
- addShardRows(shard, result);
- }
-
- return result;
- }
-}
diff --git a/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java b/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java
new file mode 100644
index 0000000000..406693d23e
--- /dev/null
+++ b/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java
@@ -0,0 +1,186 @@
+/*
+ * 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.virtual;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.marshal.BooleanType;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.db.marshal.LongType;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.dht.LocalPartitioner;
+import org.apache.cassandra.journal.ActiveSegment;
+import org.apache.cassandra.journal.Segment;
+import org.apache.cassandra.replication.CoordinatorLog;
+import org.apache.cassandra.replication.CoordinatorLogId;
+import org.apache.cassandra.replication.MutationJournal;
+import org.apache.cassandra.replication.MutationTrackingService;
+import org.apache.cassandra.replication.Shard;
+import org.apache.cassandra.replication.ShortMutationId;
+import org.apache.cassandra.schema.TableMetadata;
+
+public class MutationTrackingTables
+{
+ public static final String MUTATION_JOURNAL = "mutation_journal";
+ public static final String MUTATION_TRACKING_SHARDS = "mutation_tracking_shards";
+
+ private MutationTrackingTables() {}
+
+ public static Collection<VirtualTable> getAll(String keyspace)
+ {
+ if (!DatabaseDescriptor.getMutationTrackingEnabled())
+ return Collections.emptyList();
+
+ return List.of(new MutationJournalTable(keyspace), new MutationTrackingShardsTable(keyspace));
+ }
+
+ public static final class MutationJournalTable extends AbstractVirtualTable
+ {
+ private static final String SEGMENT_ID = "segment_id";
+ private static final String IS_ACTIVE = "is_active";
+ private static final String BYTES_ON_DISK = "bytes_on_disk";
+ private static final String RECORDS_COUNT = "records_count";
+ private static final String WRITTEN_TO = "written_to";
+ private static final String FSYNCED_TO = "fsynced_to";
+ private static final String NEEDS_REPLAY = "needs_replay";
+ private static final String FILE_PATH = "file_path";
+
+ MutationJournalTable(String keyspace)
+ {
+ super(TableMetadata.builder(keyspace, MUTATION_JOURNAL)
+ .comment("mutation journal segments and their contents")
+ .kind(TableMetadata.Kind.VIRTUAL)
+ .partitioner(new LocalPartitioner(LongType.instance))
+ .addPartitionKeyColumn(SEGMENT_ID, LongType.instance)
+ .addRegularColumn(IS_ACTIVE, BooleanType.instance)
+ .addRegularColumn(BYTES_ON_DISK, LongType.instance)
+ .addRegularColumn(RECORDS_COUNT, Int32Type.instance)
+ .addRegularColumn(WRITTEN_TO, Int32Type.instance)
+ .addRegularColumn(FSYNCED_TO, Int32Type.instance)
+ .addRegularColumn(NEEDS_REPLAY, BooleanType.instance)
+ .addRegularColumn(FILE_PATH, UTF8Type.instance)
+ .build());
+ }
+
+ @Override
+ public DataSet data()
+ {
+ SimpleDataSet result = new SimpleDataSet(metadata());
+
+ for (Segment<ShortMutationId, Mutation> segment : MutationJournal.instance().getAllSegments())
+ {
+ result.row(segment.id())
+ .column(IS_ACTIVE, segment instanceof ActiveSegment)
+ .column(BYTES_ON_DISK, segment.segmentSizeOnDisk())
+ .column(RECORDS_COUNT, segment.metadata().totalCount())
+ .column(WRITTEN_TO, segment.writtenTo())
+ .column(FSYNCED_TO, segment.fsyncedTo())
+ .column(NEEDS_REPLAY, segment.metadata().needsReplay())
+ .column(FILE_PATH, segment.filePath());
+ }
+
+ return result;
+ }
+ }
+
+ public static class MutationTrackingShardsTable extends AbstractVirtualTable
+ {
+ private static final String KEYSPACE = "keyspace";
+ private static final String LOG_ID = "log_id";
+ private static final String RANGE_START = "range_start";
+ private static final String RANGE_END = "range_end";
+ private static final String LOCAL_NODE_ID = "local_node_id";
+ private static final String PARTICIPANTS = "participants";
+ private static final String WITNESSED_OFFSETS = "witnessed_offsets";
+ private static final String RECONCILED_OFFSETS = "reconciled_offsets";
+ private static final String PERSISTED_OFFSETS = "persisted_offsets";
+
+ MutationTrackingShardsTable(String keyspace) {
+ super(TableMetadata.builder(keyspace, MUTATION_TRACKING_SHARDS)
+ .comment("mutation tracking shards and their offset information")
+ .kind(TableMetadata.Kind.VIRTUAL).partitioner(new LocalPartitioner(UTF8Type.instance))
+ .addPartitionKeyColumn(KEYSPACE, UTF8Type.instance)
+ .addClusteringColumn(LOG_ID, UTF8Type.instance)
+ .addClusteringColumn(RANGE_START, UTF8Type.instance)
+ .addClusteringColumn(RANGE_END, UTF8Type.instance)
+ .addRegularColumn(LOCAL_NODE_ID, Int32Type.instance)
+ .addRegularColumn(PARTICIPANTS, UTF8Type.instance)
+ .addRegularColumn(WITNESSED_OFFSETS, UTF8Type.instance)
+ .addRegularColumn(RECONCILED_OFFSETS, UTF8Type.instance)
+ .addRegularColumn(PERSISTED_OFFSETS, UTF8Type.instance)
+ .build());
+ }
+
+ private void addShardRows(Shard shard, SimpleDataSet result)
+ {
+ Shard.DebugInfo shardDebugInfo = shard.getDebugInfo();
+ for (Map.Entry<CoordinatorLogId, CoordinatorLog.DebugInfo> entry : shardDebugInfo.logs.entrySet())
+ {
+ CoordinatorLogId logId = entry.getKey();
+ CoordinatorLog.DebugInfo logDebugInfo = entry.getValue();
+ result.row(shardDebugInfo.keyspace,
+ logId.toString(),
+ shardDebugInfo.range.left.toString(),
+ shardDebugInfo.range.right.toString())
+ .column(LOCAL_NODE_ID, shardDebugInfo.localNodeId)
+ .column(PARTICIPANTS, shardDebugInfo.participants.toString())
+ .column(WITNESSED_OFFSETS, logDebugInfo.witnessedOffsets)
+ .column(RECONCILED_OFFSETS, logDebugInfo.reconciledOffsets)
+ .column(PERSISTED_OFFSETS, logDebugInfo.persistedOffsets);
+ }
+ }
+
+ @Override
+ public DataSet data()
+ {
+ SimpleDataSet result = new SimpleDataSet(metadata());
+
+ for (Shard shard : MutationTrackingService.instance().getShards())
+ {
+ addShardRows(shard, result);
+ }
+
+ return result;
+ }
+
+ @Override
+ public DataSet data(DecoratedKey key)
+ {
+ String keyspaceName = UTF8Type.instance.compose(key.getKey());
+ SimpleDataSet result = new SimpleDataSet(metadata());
+
+ for (Shard shard : MutationTrackingService.instance().getShards())
+ {
+ Shard.DebugInfo debugInfo = shard.getDebugInfo();
+ if (!debugInfo.keyspace.equals(keyspaceName))
+ continue;
+
+ addShardRows(shard, result);
+ }
+
+ return result;
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java b/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java
index 7308167364..7b82cd18a8 100644
--- a/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java
+++ b/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java
@@ -69,8 +69,7 @@ public final class SystemViewsKeyspace extends VirtualKeyspace
.add(new LocalTable(VIRTUAL_VIEWS))
.add(new ClusterMetadataLogTable(VIRTUAL_VIEWS))
.add(new ClusterMetadataDirectoryTable(VIRTUAL_VIEWS))
- .add(new MutationJournalTable(VIRTUAL_VIEWS))
- .add(new MutationTrackingShardsTable(VIRTUAL_VIEWS))
+ .addAll(MutationTrackingTables.getAll(VIRTUAL_VIEWS))
.addAll(LocalRepairTables.getAll(VIRTUAL_VIEWS))
.addAll(CIDRFilteringMetricsTable.getAll(VIRTUAL_VIEWS))
.addAll(StorageAttachedIndexTables.getAll(VIRTUAL_VIEWS))
diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java
index 79ea0608cf..9252ca6837 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java
@@ -357,7 +357,7 @@ public abstract class SSTableWriter extends SSTable implements Transactional
if (metadata().replicationType().isTracked() && repairedAt == ActiveRepairService.UNREPAIRED_SSTABLE && reconcile)
{
Preconditions.checkState(Objects.equals(pendingRepair, ActiveRepairService.NO_PENDING_REPAIR));
- if (MutationTrackingService.instance.isDurablyReconciled(coordinatorLogOffsets))
+ if (MutationTrackingService.instance().isDurablyReconciled(coordinatorLogOffsets))
{
repairedAt = Clock.Global.currentTimeMillis();
logger.debug("Marking SSTable {} as reconciled with repairedAt {}", descriptor, repairedAt);
diff --git a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java
index 48dbedc68d..de874c8068 100644
--- a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java
+++ b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java
@@ -49,6 +49,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.virtual.CollectionVirtualTableAdapter;
import org.apache.cassandra.db.virtual.VirtualTable;
import org.apache.cassandra.db.virtual.model.CounterMetricRow;
@@ -113,59 +114,62 @@ public class CassandraMetricsRegistry extends MetricRegistry
// For example, if a new metric group is added, the appropriate tests will fail, because of the missing
// group in this collection, which in turn guarantees that all metric groups are registered and exposed
// for virtual tables.
- metricGroups = ImmutableSet.<String>builder()
- .add(AbstractMetrics.TYPE)
- .add(AccordCoordinatorMetrics.ACCORD_COORDINATOR)
- .add(AccordCacheMetrics.ACCORD_CACHE)
- .add(AccordReplicaMetrics.ACCORD_REPLICA)
- .add(AccordExecutorMetrics.ACCORD_EXECUTOR)
- .add(AccordSystemMetrics.ACCORD_SYSTEM)
- .add(BatchMetrics.TYPE_NAME)
- .add(BufferPoolMetrics.TYPE_NAME)
- .add(CIDRAuthorizerMetrics.TYPE_NAME)
- .add(CQLMetrics.TYPE_NAME)
- .add(CacheMetrics.TYPE_NAME)
- .add(ChunkCacheMetrics.TYPE_NAME)
- .add(ClientMessageSizeMetrics.TYPE)
- .add(ClientMetrics.TYPE_NAME)
- .add(ClientRequestMetrics.TYPE_NAME)
- .add(ClientRequestSizeMetrics.TYPE)
- .add(CommitLogMetrics.TYPE_NAME)
- .add(CompactionMetrics.TYPE_NAME)
- .add(DenylistMetrics.TYPE_NAME)
- .add(DroppedMessageMetrics.TYPE)
- .add(HintedHandoffMetrics.TYPE_NAME)
- .add(HintsServiceMetrics.TYPE_NAME)
- .add(org.apache.cassandra.index.accord.IndexMetrics.TYPE)
- .add(InternodeInboundMetrics.TYPE_NAME)
- .add(InternodeOutboundMetrics.TYPE_NAME)
- .add(org.apache.cassandra.journal.Metrics.TYPE_NAME)
- .add(KeyspaceMetrics.TYPE_NAME)
- .add(MemtablePool.TYPE_NAME)
- .add(MessagingMetrics.TYPE_NAME)
- .add(MutationTrackingMetrics.TYPE_NAME)
- .add(MutualTlsMetrics.TYPE_NAME)
- .add(PaxosMetrics.TYPE_NAME)
- .add(ReadRepairMetrics.TYPE_NAME)
- .add(RepairMetrics.TYPE_NAME)
- .add(RowIndexEntry.TYPE_NAME)
- .add(StorageMetrics.TYPE_NAME)
- .add(StreamingMetrics.TYPE_NAME)
- .add(TCMMetrics.TYPE_NAME)
- .add(TableMetrics.ALIAS_TYPE_NAME)
- .add(TableMetrics.TYPE_NAME)
- .add(TableMetrics.INDEX_TYPE_NAME)
- .add(TableMetrics.INDEX_ALIAS_TYPE_NAME)
- .add(ThreadPoolMetrics.TYPE_NAME)
- .add(TrieMemtableMetricsView.TYPE_NAME)
- .add(UnweightedCacheMetrics.TYPE_NAME)
- .add(AutoRepairMetrics.TYPE_NAME)
- .build();
+ ImmutableSet.Builder<String> builder = ImmutableSet.builder();
+ builder = ImmutableSet.<String>builder()
+ .add(AbstractMetrics.TYPE)
+ .add(AccordCoordinatorMetrics.ACCORD_COORDINATOR)
+ .add(AccordCacheMetrics.ACCORD_CACHE)
+ .add(AccordReplicaMetrics.ACCORD_REPLICA)
+ .add(AccordExecutorMetrics.ACCORD_EXECUTOR)
+ .add(AccordSystemMetrics.ACCORD_SYSTEM)
+ .add(BatchMetrics.TYPE_NAME)
+ .add(BufferPoolMetrics.TYPE_NAME)
+ .add(CIDRAuthorizerMetrics.TYPE_NAME)
+ .add(CQLMetrics.TYPE_NAME)
+ .add(CacheMetrics.TYPE_NAME)
+ .add(ChunkCacheMetrics.TYPE_NAME)
+ .add(ClientMessageSizeMetrics.TYPE)
+ .add(ClientMetrics.TYPE_NAME)
+ .add(ClientRequestMetrics.TYPE_NAME)
+ .add(ClientRequestSizeMetrics.TYPE)
+ .add(CommitLogMetrics.TYPE_NAME)
+ .add(CompactionMetrics.TYPE_NAME)
+ .add(DenylistMetrics.TYPE_NAME)
+ .add(DroppedMessageMetrics.TYPE)
+ .add(HintedHandoffMetrics.TYPE_NAME)
+ .add(HintsServiceMetrics.TYPE_NAME)
+ .add(org.apache.cassandra.index.accord.IndexMetrics.TYPE)
+ .add(InternodeInboundMetrics.TYPE_NAME)
+ .add(InternodeOutboundMetrics.TYPE_NAME)
+ .add(org.apache.cassandra.journal.Metrics.TYPE_NAME)
+ .add(KeyspaceMetrics.TYPE_NAME)
+ .add(MemtablePool.TYPE_NAME)
+ .add(MessagingMetrics.TYPE_NAME)
+ .add(MutationTrackingMetrics.TYPE_NAME)
+ .add(MutualTlsMetrics.TYPE_NAME)
+ .add(PaxosMetrics.TYPE_NAME)
+ .add(ReadRepairMetrics.TYPE_NAME)
+ .add(RepairMetrics.TYPE_NAME)
+ .add(RowIndexEntry.TYPE_NAME)
+ .add(StorageMetrics.TYPE_NAME)
+ .add(StreamingMetrics.TYPE_NAME)
+ .add(TCMMetrics.TYPE_NAME)
+ .add(TableMetrics.ALIAS_TYPE_NAME)
+ .add(TableMetrics.TYPE_NAME)
+ .add(TableMetrics.INDEX_TYPE_NAME)
+ .add(TableMetrics.INDEX_ALIAS_TYPE_NAME)
+ .add(ThreadPoolMetrics.TYPE_NAME)
+ .add(TrieMemtableMetricsView.TYPE_NAME)
+ .add(UnweightedCacheMetrics.TYPE_NAME)
+ .add(AutoRepairMetrics.TYPE_NAME);
+
+ if (DatabaseDescriptor.isDaemonInitialized() && DatabaseDescriptor.getMutationTrackingEnabled())
+ builder.add(MutationTrackingMetrics.TYPE_NAME);
+
+ metricGroups = builder.build();
}
- private CassandraMetricsRegistry()
- {
- }
+ private CassandraMetricsRegistry() {}
@SuppressWarnings("rawtypes")
public static String getValueAsString(Metric metric)
diff --git a/src/java/org/apache/cassandra/metrics/MutationTrackingMetrics.java b/src/java/org/apache/cassandra/metrics/MutationTrackingMetrics.java
index 822d7c5ac1..2b7650e36b 100644
--- a/src/java/org/apache/cassandra/metrics/MutationTrackingMetrics.java
+++ b/src/java/org/apache/cassandra/metrics/MutationTrackingMetrics.java
@@ -21,17 +21,27 @@ package org.apache.cassandra.metrics;
import com.codahale.metrics.Counter;
import com.codahale.metrics.Gauge;
import com.codahale.metrics.Histogram;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.replication.MutationTrackingService;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
+import static org.apache.cassandra.replication.MutationTrackingService.DISABLED_MESSAGE;
public class MutationTrackingMetrics
{
public static final String TYPE_NAME = "MutationTracking";
private static final MetricNameFactory factory = new DefaultNameFactory(TYPE_NAME);
- public static final MutationTrackingMetrics instance = new MutationTrackingMetrics();
+ private static final MutationTrackingMetrics instance = DatabaseDescriptor.getMutationTrackingEnabled() ? new MutationTrackingMetrics() : null;
+
+ public static MutationTrackingMetrics instance()
+ {
+ if (instance == null)
+ throw new IllegalStateException(DISABLED_MESSAGE);
+ return instance;
+ }
public final Counter broadcastOffsetsDiscovered; // Newly-witnessed offsets discovered via broadcast
public final Counter writeTimeOffsetsDiscovered; // Newly-witnessed offsets discovered at write time
@@ -47,11 +57,11 @@ public class MutationTrackingMetrics
readSummarySize = Metrics.histogram(factory.createMetricName("ReadSummarySize"), true);
unreconciledMutationCount = Metrics.register(
factory.createMetricName("UnreconciledMutationCount"),
- () -> MutationTrackingService.instance.getUnreconciledMutationCount()
+ () -> MutationTrackingService.instance().getUnreconciledMutationCount()
);
journalDiskSpaceUsed = Metrics.register(
factory.createMetricName("JournalDiskSpaceUsed"),
- () -> MutationJournal.instance.getDiskSpaceUsed()
+ () -> MutationJournal.instance().getDiskSpaceUsed()
);
}
}
\ No newline at end of file
diff --git a/src/java/org/apache/cassandra/repair/SyncTasks.java b/src/java/org/apache/cassandra/repair/SyncTasks.java
index d5d60a81d1..83aa7e25a5 100644
--- a/src/java/org/apache/cassandra/repair/SyncTasks.java
+++ b/src/java/org/apache/cassandra/repair/SyncTasks.java
@@ -68,7 +68,7 @@ public class SyncTasks extends AbstractCollection<SyncTask>
*/
static SyncTasks tracked(Keyspace keyspace, List<SyncTask> tasks)
{
- return MutationTrackingService.instance.alignToShardBoundaries(keyspace, tasks);
+ return MutationTrackingService.instance().alignToShardBoundaries(keyspace, tasks);
}
public void add(Shard shard, SyncTask task)
diff --git a/src/java/org/apache/cassandra/replication/ActivationRequest.java b/src/java/org/apache/cassandra/replication/ActivationRequest.java
index 1b9f260739..930fde8b8b 100644
--- a/src/java/org/apache/cassandra/replication/ActivationRequest.java
+++ b/src/java/org/apache/cassandra/replication/ActivationRequest.java
@@ -135,7 +135,7 @@ public class ActivationRequest
public void apply()
{
- MutationTrackingService.instance.activateLocal(this);
+ MutationTrackingService.instance().activateLocal(this);
}
public static final Serializer serializer = new Serializer();
diff --git a/src/java/org/apache/cassandra/replication/ActiveLogReconciler.java b/src/java/org/apache/cassandra/replication/ActiveLogReconciler.java
index ccc597d546..3f52f6eedb 100644
--- a/src/java/org/apache/cassandra/replication/ActiveLogReconciler.java
+++ b/src/java/org/apache/cassandra/replication/ActiveLogReconciler.java
@@ -161,21 +161,21 @@ public final class ActiveLogReconciler implements Shutdownable
@Override
public void onResponse(Message<NoPayload> msg)
{
- MutationTrackingService.instance.receivedWriteResponse(mutationId, toHost);
+ MutationTrackingService.instance().receivedWriteResponse(mutationId, toHost);
}
@Override
public void onFailure(InetAddressAndPort from, RequestFailure failureReason)
{
- MutationTrackingService.instance.retryFailedWrite(mutationId, toHost, failureReason);
+ MutationTrackingService.instance().retryFailedWrite(mutationId, toHost, failureReason);
}
void send()
{
- RecordPointer pointer = MutationJournal.instance.lookUp(mutationId);
+ RecordPointer pointer = MutationJournal.instance().lookUp(mutationId);
Preconditions.checkNotNull(pointer, "Mutation %s not found in the journal", mutationId);
- MutationJournal.instance.read(pointer, (segment, position, key, buffer, version) -> {
+ MutationJournal.instance().read(pointer, (segment, position, key, buffer, version) -> {
// don't send mutations to nodes that have migrated to, or are in the process of migrating to untracked replication
try (DataInputBuffer in = new DataInputBuffer(buffer, true))
@@ -227,7 +227,7 @@ public final class ActiveLogReconciler implements Shutdownable
public void onResponse(Message<NoPayload> msg)
{
logger.debug("Received activation ack for TransferTask from {}", toHost);
- MutationTrackingService.instance.receivedActivationResponse(transfer, toHost);
+ MutationTrackingService.instance().receivedActivationResponse(transfer, toHost);
}
@Override
@@ -239,7 +239,7 @@ public final class ActiveLogReconciler implements Shutdownable
public void onFailure(Throwable cause)
{
logger.debug("Received activation failure for TransferTask from {} due to", toHost, cause);
- MutationTrackingService.instance.retryFailedTransfer(transfer, toHost, cause);
+ MutationTrackingService.instance().retryFailedTransfer(transfer, toHost, cause);
}
void send()
diff --git a/src/java/org/apache/cassandra/replication/BroadcastLogOffsets.java b/src/java/org/apache/cassandra/replication/BroadcastLogOffsets.java
index abd23dc3f2..a9832f90c6 100644
--- a/src/java/org/apache/cassandra/replication/BroadcastLogOffsets.java
+++ b/src/java/org/apache/cassandra/replication/BroadcastLogOffsets.java
@@ -67,8 +67,9 @@ public class BroadcastLogOffsets
}
public static final IVerbHandler<BroadcastLogOffsets> verbHandler = message -> {
+ MutationTrackingService.ensureEnabled();
BroadcastLogOffsets replicatedOffsets = message.payload;
- MutationTrackingService.instance.updateReplicatedOffsets(replicatedOffsets.keyspace,
+ MutationTrackingService.instance().updateReplicatedOffsets(replicatedOffsets.keyspace,
replicatedOffsets.range,
replicatedOffsets.replicatedOffsets,
replicatedOffsets.durable,
diff --git a/src/java/org/apache/cassandra/replication/CoordinatedTransfer.java b/src/java/org/apache/cassandra/replication/CoordinatedTransfer.java
index 65456705dd..0c6015657f 100644
--- a/src/java/org/apache/cassandra/replication/CoordinatedTransfer.java
+++ b/src/java/org/apache/cassandra/replication/CoordinatedTransfer.java
@@ -187,7 +187,7 @@ public abstract class CoordinatedTransfer
logger.debug("{} Activation successfully applied on {}", logPrefix(), msg.from());
streamResults.computeIfPresent(msg.payload.syncPair, (peer, result) -> result.committed());
- MutationTrackingService.instance.receivedActivationResponse(CoordinatedTransfer.this, msg.from());
+ MutationTrackingService.instance().receivedActivationResponse(CoordinatedTransfer.this, msg.from());
if (responses.decrementAndGet() == 0)
{
diff --git a/src/java/org/apache/cassandra/replication/CoordinatorLog.java b/src/java/org/apache/cassandra/replication/CoordinatorLog.java
index 70479c12be..7db4059296 100644
--- a/src/java/org/apache/cassandra/replication/CoordinatorLog.java
+++ b/src/java/org/apache/cassandra/replication/CoordinatorLog.java
@@ -229,7 +229,7 @@ public abstract class CoordinatorLog
});
// Record metric for newly witnessed offsets only
- MutationTrackingMetrics.instance.broadcastOffsetsDiscovered.inc(newlyWitnessedCount[0]);
+ MutationTrackingMetrics.instance().broadcastOffsetsDiscovered.inc(newlyWitnessedCount[0]);
}
private void updatePersistedReplicatedOffsets(Offsets offsets, int onNodeId)
@@ -327,7 +327,7 @@ public abstract class CoordinatorLog
return;
// Track write-time discovery of newly-witnessed offset
- MutationTrackingMetrics.instance.writeTimeOffsetsDiscovered.inc();
+ MutationTrackingMetrics.instance().writeTimeOffsetsDiscovered.inc();
unreconciledMutations.finishWriting(mutation);
maybeMoveOffset(offset);
diff --git a/src/java/org/apache/cassandra/replication/ForwardedWrite.java b/src/java/org/apache/cassandra/replication/ForwardedWrite.java
index ba4aecb002..d40d3cfbf6 100644
--- a/src/java/org/apache/cassandra/replication/ForwardedWrite.java
+++ b/src/java/org/apache/cassandra/replication/ForwardedWrite.java
@@ -194,7 +194,7 @@ public class ForwardedWrite
String keyspaceName = mutation.getKeyspaceName();
Token token = mutation.key().getToken();
- MutationId id = MutationTrackingService.instance.nextMutationId(keyspaceName, token);
+ MutationId id = MutationTrackingService.instance().nextMutationId(keyspaceName, token);
// Do not wait for handler completion, since the coordinator is already waiting and we don't want to block the stage
LeaderCallback handler = new LeaderCallback(id, ackTo);
applyLocallyAndForwardToReplicas(mutation.withMutationId(id), recipients, handler, ackTo);
@@ -427,7 +427,7 @@ public class ForwardedWrite
Keyspace ks = Keyspace.open(keyspaceName);
ReplicaPlan.ForWrite plan = ReplicaPlans.forWrite(ks, counterMutation.consistency(), token, ReplicaPlans.writeAll);
- MutationId id = MutationTrackingService.instance.nextMutationId(keyspaceName, token);
+ MutationId id = MutationTrackingService.instance().nextMutationId(keyspaceName, token);
logger.trace("Forwarded counter mutation {}: applying locally with ID and forwarding to other replicas", id);
@@ -483,6 +483,7 @@ public class ForwardedWrite
public static final IVerbHandler<MutationRequest> verbHandler = incoming ->
{
+ MutationTrackingService.ensureEnabled();
if (approxTime.now() > incoming.expiresAtNanos())
{
Tracing.trace("Discarding mutation from {} (timed out)", incoming.from());
@@ -524,7 +525,7 @@ public class ForwardedWrite
{
// Local mutations are witnessed from Keyspace.applyInternalTracked
if (msg != null)
- MutationTrackingService.instance.receivedWriteResponse(id, msg.from());
+ MutationTrackingService.instance().receivedWriteResponse(id, msg.from());
// Local write needs to be ack'd to coordinator
if (msg == null && ackTo != null)
diff --git a/src/java/org/apache/cassandra/replication/MutationJournal.java b/src/java/org/apache/cassandra/replication/MutationJournal.java
index 9fae5371ef..7d8c2425ed 100644
--- a/src/java/org/apache/cassandra/replication/MutationJournal.java
+++ b/src/java/org/apache/cassandra/replication/MutationJournal.java
@@ -50,12 +50,13 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;
import java.util.zip.CRC32;
+import static org.apache.cassandra.replication.MutationTrackingService.DISABLED_MESSAGE;
import static org.apache.cassandra.utils.FBUtilities.getAvailableProcessors;
// TODO (required): handle table truncations
public class MutationJournal
{
- public static final MutationJournal instance = new MutationJournal();
+ private static final MutationJournal instance = DatabaseDescriptor.getMutationTrackingEnabled() ? new MutationJournal() : null;
private final Journal<ShortMutationId, Mutation> journal;
private final Map<Long, SegmentStateTracker> segmentStateTrackers;
@@ -93,7 +94,31 @@ public class MutationJournal
private MutationJournal()
{
- this(new File(DatabaseDescriptor.getCommitLogLocation()), new JournalParams());
+ this(new File(DatabaseDescriptor.getMutationTrackingJournalDirectory()), new JournalParams());
+ }
+
+ public static void start()
+ {
+ if (instance == null)
+ return;
+
+ instance.startInternal();
+ }
+
+ public static void shutdown()
+ {
+ if (instance == null)
+ return;
+
+ instance.shutdownBlocking();
+ }
+
+ public static MutationJournal instance()
+ {
+ if (instance == null)
+ throw new IllegalStateException(DISABLED_MESSAGE);
+
+ return instance;
}
@VisibleForTesting
@@ -142,12 +167,12 @@ public class MutationJournal
}
}
- public void start()
+ void startInternal()
{
journal.start();
}
- public void shutdownBlocking()
+ void shutdownBlocking()
{
journal.shutdown();
}
diff --git a/src/java/org/apache/cassandra/replication/MutationTrackingService.java b/src/java/org/apache/cassandra/replication/MutationTrackingService.java
index 7dc59c7988..e890b9c034 100644
--- a/src/java/org/apache/cassandra/replication/MutationTrackingService.java
+++ b/src/java/org/apache/cassandra/replication/MutationTrackingService.java
@@ -45,8 +45,9 @@ import com.google.common.collect.Iterables;
import org.agrona.collections.IntArrayList;
import org.agrona.collections.IntHashSet;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
-import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Mutation;
@@ -97,7 +98,59 @@ import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
// TODO (expected): handle topology changes
public class MutationTrackingService
{
- private static final ScheduledExecutorPlus executor = executorFactory().scheduled("Mutation-Tracking-Service", NORMAL);
+ public static final String DISABLED_MESSAGE = "Mutation tracking is not enabled. (See mutation_tracking.enabled in cassandra.yaml)";
+
+ private static final MutationTrackingService instance;
+ private static final ScheduledExecutorPlus executor;
+
+ static
+ {
+ if (DatabaseDescriptor.getMutationTrackingEnabled())
+ {
+ instance = new MutationTrackingService();
+ executor = executorFactory().scheduled("Mutation-Tracking-Service", NORMAL);
+ }
+ else
+ {
+ instance = null;
+ executor = null;
+ }
+ }
+
+ /**
+ * Callers of this method should have validated that mutation tracking is enabled
+ */
+ public static MutationTrackingService instance()
+ {
+ if (instance == null)
+ throw new IllegalStateException(DISABLED_MESSAGE);
+ return instance;
+ }
+
+ public static boolean isEnabled()
+ {
+ return DatabaseDescriptor.getMutationTrackingEnabled();
+ }
+
+ public static void ensureEnabled()
+ {
+ if (!DatabaseDescriptor.getMutationTrackingEnabled())
+ throw new IllegalStateException(DISABLED_MESSAGE);
+ }
+
+ public static void start(ClusterMetadata metadata)
+ {
+ if (!isEnabled())
+ return;
+ instance().startInternal(metadata);
+ }
+
+ public static void shutdown() throws InterruptedException
+ {
+ if (!isEnabled())
+ return;
+ instance().shutdownBlocking();
+ }
/**
* Split ranges into this many shards.
@@ -110,7 +163,6 @@ public class MutationTrackingService
private static final int SHARD_MULTIPLIER = 1;
private static final Logger logger = LoggerFactory.getLogger(MutationTrackingService.class);
- public static final MutationTrackingService instance = new MutationTrackingService();
private final TrackedLocalReads localReads = new TrackedLocalReads();
private ConcurrentHashMap<String, KeyspaceShards> keyspaceShards = new ConcurrentHashMap<>();
@@ -154,7 +206,7 @@ public class MutationTrackingService
};
}
- public synchronized void start(ClusterMetadata metadata)
+ private synchronized void startInternal(ClusterMetadata metadata)
{
if (started)
return;
@@ -204,7 +256,7 @@ public class MutationTrackingService
return builder.build();
}
- public void registerTCMListener()
+ public void registerMetadataListener()
{
ClusterMetadataService.instance().log().addListener(tcmListener);
}
@@ -214,7 +266,7 @@ public class MutationTrackingService
return started;
}
- public void shutdownBlocking() throws InterruptedException
+ private void shutdownBlocking() throws InterruptedException
{
ClusterMetadataService.instance().log().removeListener(tcmListener);
activeReconciler.shutdownBlocking();
@@ -488,7 +540,7 @@ public class MutationTrackingService
try
{
MutationSummary summary = getOrCreateShards(tableId).createSummaryForKey(key, tableId, includePending);
- MutationTrackingMetrics.instance.readSummarySize.update(summary.size());
+ MutationTrackingMetrics.instance().readSummarySize.update(summary.size());
return summary;
}
finally
@@ -503,7 +555,7 @@ public class MutationTrackingService
try
{
MutationSummary summary = getOrCreateShards(tableId).createSummaryForRange(range, tableId, includePending);
- MutationTrackingMetrics.instance.readSummarySize.update(summary.size());
+ MutationTrackingMetrics.instance().readSummarySize.update(summary.size());
return summary;
}
finally
@@ -834,7 +886,7 @@ public class MutationTrackingService
{
Log2OffsetsMap.Mutable reconciledOffsets = new Log2OffsetsMap.Mutable();
collectDurablyReconciledOffsets(reconciledOffsets);
- MutationJournal.instance.dropReconciledSegments(reconciledOffsets);
+ MutationJournal.instance().dropReconciledSegments(reconciledOffsets);
}
/**
@@ -1258,7 +1310,7 @@ public class MutationTrackingService
public void run(boolean durable)
{
- MutationTrackingService.instance.forEachKeyspace(ks -> run(ks, durable));
+ MutationTrackingService.instance().forEachKeyspace(ks -> run(ks, durable));
}
private void run(KeyspaceShards shards, boolean durable)
@@ -1301,9 +1353,9 @@ public class MutationTrackingService
private void run(boolean dropSegments)
{
- MutationTrackingService.instance.forEachKeyspace(this::run);
+ MutationTrackingService.instance().forEachKeyspace(this::run);
if (dropSegments)
- MutationTrackingService.instance.truncateMutationJournal();
+ MutationTrackingService.instance().truncateMutationJournal();
}
private void run(KeyspaceShards shards)
diff --git a/src/java/org/apache/cassandra/replication/PullMutationsRequest.java b/src/java/org/apache/cassandra/replication/PullMutationsRequest.java
index d8e6f0febb..fa6d00285f 100644
--- a/src/java/org/apache/cassandra/replication/PullMutationsRequest.java
+++ b/src/java/org/apache/cassandra/replication/PullMutationsRequest.java
@@ -66,10 +66,11 @@ public final class PullMutationsRequest
@Override
public void doVerb(Message<PullMutationsRequest> message)
{
+ MutationTrackingService.ensureEnabled();
InetAddressAndPort forHost = message.from();
Offsets offsets = message.payload.offsets;
logger.trace("Received pull mutations request from {} for {}", forHost, offsets);
- MutationTrackingService.instance.requestMissingMutations(offsets, forHost);
+ MutationTrackingService.instance().requestMissingMutations(offsets, forHost);
}
};
diff --git a/src/java/org/apache/cassandra/replication/PushMutationRequest.java b/src/java/org/apache/cassandra/replication/PushMutationRequest.java
index 84f50ed97e..c104b7b306 100644
--- a/src/java/org/apache/cassandra/replication/PushMutationRequest.java
+++ b/src/java/org/apache/cassandra/replication/PushMutationRequest.java
@@ -64,7 +64,7 @@ public interface PushMutationRequest
public long serializedSize(int version)
{
// TODO (expected): handle mismatched (messaging) versions
- int size = MutationJournal.instance.sizeOfRecord(pointer);
+ int size = MutationJournal.instance().sizeOfRecord(pointer);
Preconditions.checkState(size > 0, "Couldn't read mutation %s size from the mutation journal", id);
return size;
}
@@ -72,7 +72,7 @@ public interface PushMutationRequest
@Override
public void serialize(DataOutputPlus out, int version) throws IOException
{
- boolean read = MutationJournal.instance.read(pointer, (segment, position, key, buffer, userVersion) ->
+ boolean read = MutationJournal.instance().read(pointer, (segment, position, key, buffer, userVersion) ->
{
try
{
@@ -168,6 +168,7 @@ public interface PushMutationRequest
@Override
public void doVerb(Message<Materialized> message)
{
+ MutationTrackingService.ensureEnabled();
if (approxTime.now() > message.expiresAtNanos())
{
Tracing.trace("Discarding mutation from {} (timed out)", message.from());
diff --git a/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java b/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java
index ba35f231d9..0eb64a2b41 100644
--- a/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java
+++ b/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java
@@ -186,7 +186,7 @@ public class TrackedWriteRequest
logger.trace("Local tracked request {} {}", mutation, plan);
writeMetrics.localRequests.mark();
- MutationId id = MutationTrackingService.instance.nextMutationId(keyspaceName, token);
+ MutationId id = MutationTrackingService.instance().nextMutationId(keyspaceName, token);
mutation = mutation.withMutationId(id);
if (logger.isTraceEnabled())
@@ -338,7 +338,7 @@ public class TrackedWriteRequest
}
if (remoteReplicas != null)
- MutationTrackingService.instance.sentWriteRequest(mutation, remoteReplicas);
+ MutationTrackingService.instance().sentWriteRequest(mutation, remoteReplicas);
}
/*
diff --git a/src/java/org/apache/cassandra/replication/UnreconciledMutations.java b/src/java/org/apache/cassandra/replication/UnreconciledMutations.java
index 61b5e1746a..83c21bef08 100644
--- a/src/java/org/apache/cassandra/replication/UnreconciledMutations.java
+++ b/src/java/org/apache/cassandra/replication/UnreconciledMutations.java
@@ -268,7 +268,7 @@ public class UnreconciledMutations
for (int offset = iter.start(), end = iter.end(); offset <= end; offset++)
{
ShortMutationId id = new ShortMutationId(witnessed.logId, offset);
- Mutation mutation = MutationJournal.instance.read(id);
+ Mutation mutation = MutationJournal.instance().read(id);
if (mutation != null)
{
result.addDirectly(mutation);
diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java
index fb19b997d2..dbbf460a84 100644
--- a/src/java/org/apache/cassandra/service/CassandraDaemon.java
+++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java
@@ -272,7 +272,7 @@ public class CassandraDaemon
Keyspace.setInitialized();
CommitLog.instance.start();
- MutationJournal.instance.start();
+ MutationJournal.start();
SnapshotManager.instance.start(false);
SnapshotManager.instance.clearExpiredSnapshots();
@@ -361,7 +361,9 @@ public class CassandraDaemon
NodeId self = ClusterMetadata.current().myNodeId();
if (self != null)
AccordService.localStartup(self);
- MutationJournal.instance.replayStaticSegments();
+
+ if (DatabaseDescriptor.getMutationTrackingEnabled())
+ MutationJournal.instance().replayStaticSegments();
}
catch (IOException e)
{
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 29fcb17c93..3d58496fe6 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -1078,7 +1078,7 @@ public class StorageProxy implements StorageProxyMBean
MutationId mutationId;
if (proposal.mutation.id().isNone())
{
- mutationId = MutationTrackingService.instance.nextMutationId(keyspaceName, tk);
+ mutationId = MutationTrackingService.instance().nextMutationId(keyspaceName, tk);
proposal = proposal.withMutationId(mutationId);
}
else
@@ -1145,7 +1145,7 @@ public class StorageProxy implements StorageProxyMBean
// Register write request with tracking service
if (!remoteReplicas.isEmpty())
- MutationTrackingService.instance.sentWriteRequest(proposal.makeMutation(), remoteReplicas);
+ MutationTrackingService.instance().sentWriteRequest(proposal.makeMutation(), remoteReplicas);
if (shouldBlock)
responseHandler.get();
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 418ef70cc0..c4fa7f97f0 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -3963,7 +3963,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
SnapshotManager.instance.shutdownAndWait(1L, MINUTES);
HintsService.instance.shutdownBlocking();
- MutationTrackingService.instance.shutdownBlocking();
+ MutationTrackingService.shutdown();
// Interrupt ongoing compactions and shutdown CM to prevent further compactions.
CompactionManager.instance.forceShutdown();
@@ -3973,7 +3973,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
CommitLog.instance.forceRecycleAllSegments();
CommitLog.instance.shutdownBlocking();
- MutationJournal.instance.shutdownBlocking();
+ MutationJournal.shutdown();
AutoRepair.instance.shutdownBlocking();
diff --git a/src/java/org/apache/cassandra/service/TrackedWriteResponseHandler.java b/src/java/org/apache/cassandra/service/TrackedWriteResponseHandler.java
index f961fceb01..e98212772b 100644
--- a/src/java/org/apache/cassandra/service/TrackedWriteResponseHandler.java
+++ b/src/java/org/apache/cassandra/service/TrackedWriteResponseHandler.java
@@ -56,7 +56,7 @@ public class TrackedWriteResponseHandler<T> extends AbstractWriteResponseHandler
{
if (logger.isTraceEnabled())
logger.trace("Received write response for mutation {} from {}", mutationId, msg.from());
- MutationTrackingService.instance.receivedWriteResponse(mutationId, msg.from());
+ MutationTrackingService.instance().receivedWriteResponse(mutationId, msg.from());
}
wrapped.onResponse(msg);
}
@@ -66,7 +66,7 @@ public class TrackedWriteResponseHandler<T> extends AbstractWriteResponseHandler
{
if (logger.isTraceEnabled())
logger.trace("Write failed for mutation {} from {}: {}", mutationId, from, failure);
- MutationTrackingService.instance.retryFailedWrite(mutationId, from, failure);
+ MutationTrackingService.instance().retryFailedWrite(mutationId, from, failure);
wrapped.onFailure(from, failure);
}
diff --git a/src/java/org/apache/cassandra/service/paxos/PaxosCommit.java b/src/java/org/apache/cassandra/service/paxos/PaxosCommit.java
index 41e85cf105..ae25c5cafd 100644
--- a/src/java/org/apache/cassandra/service/paxos/PaxosCommit.java
+++ b/src/java/org/apache/cassandra/service/paxos/PaxosCommit.java
@@ -147,7 +147,7 @@ public class PaxosCommit<OnDone extends Consumer<? super PaxosCommit.Status>> ex
if (commit.mutation.id().isNone())
{
Token token = commit.partitionKey().getToken();
- MutationId mutationId = MutationTrackingService.instance.nextMutationId(commit.metadata().keyspace, token);
+ MutationId mutationId = MutationTrackingService.instance().nextMutationId(commit.metadata().keyspace, token);
Mutation mutationWithId = commit.makeMutation(mutationId);
commitToUse = new Commit.Agreed(commit.ballot, mutationWithId);
}
@@ -354,7 +354,7 @@ public class PaxosCommit<OnDone extends Consumer<? super PaxosCommit.Status>> ex
if (remoteReplicas != null)
{
checkState(!remoteReplicas.isEmpty());
- MutationTrackingService.instance.sentWriteRequest(commit.makeMutation(), remoteReplicas);
+ MutationTrackingService.instance().sentWriteRequest(commit.makeMutation(), remoteReplicas);
}
if (executeOnSelf)
@@ -404,7 +404,7 @@ public class PaxosCommit<OnDone extends Consumer<? super PaxosCommit.Status>> ex
// Track failed response for tracked keyspaces
if (isTracked())
- MutationTrackingService.instance.retryFailedWrite(commit.mutation.id(), from, reason);
+ MutationTrackingService.instance().retryFailedWrite(commit.mutation.id(), from, reason);
response(false, from);
Replica replica = replicas.lookup(from);
@@ -423,7 +423,7 @@ public class PaxosCommit<OnDone extends Consumer<? super PaxosCommit.Status>> ex
// Track successful response for tracked keyspaces
// (Local mutations are witnessed from Keyspace.applyInternalTracked)
if (isTracked())
- MutationTrackingService.instance.receivedWriteResponse(commit.mutation.id(), response.from());
+ MutationTrackingService.instance().receivedWriteResponse(commit.mutation.id(), response.from());
response(true, response.from());
}
@@ -462,9 +462,9 @@ public class PaxosCommit<OnDone extends Consumer<? super PaxosCommit.Status>> ex
if (isTracked())
{
if (response != null)
- MutationTrackingService.instance.receivedWriteResponse(commit.mutation.id(), from);
+ MutationTrackingService.instance().receivedWriteResponse(commit.mutation.id(), from);
else
- MutationTrackingService.instance.retryFailedWrite(commit.mutation.id(), from, RequestFailure.UNKNOWN);
+ MutationTrackingService.instance().retryFailedWrite(commit.mutation.id(), from, RequestFailure.UNKNOWN);
}
response(response != null, from);
diff --git a/src/java/org/apache/cassandra/service/paxos/PaxosPrepareRefresh.java b/src/java/org/apache/cassandra/service/paxos/PaxosPrepareRefresh.java
index d14350728b..08a2d84e26 100644
--- a/src/java/org/apache/cassandra/service/paxos/PaxosPrepareRefresh.java
+++ b/src/java/org/apache/cassandra/service/paxos/PaxosPrepareRefresh.java
@@ -109,7 +109,7 @@ public class PaxosPrepareRefresh implements RequestCallbackWithFailure<PaxosPrep
{
// We ARE a replica - generate mutation ID and update the commit
String keyspaceName = commit.metadata().keyspace;
- MutationId mutationId = MutationTrackingService.instance.nextMutationId(keyspaceName, commit.partitionKey().getToken());
+ MutationId mutationId = MutationTrackingService.instance().nextMutationId(keyspaceName, commit.partitionKey().getToken());
Mutation mutationWithId = commit.makeMutation(mutationId);
Committed commitWithId = new Commit.Committed(commit.ballot, mutationWithId);
diff --git a/src/java/org/apache/cassandra/service/paxos/PaxosState.java b/src/java/org/apache/cassandra/service/paxos/PaxosState.java
index ed2f40fcac..6076df0022 100644
--- a/src/java/org/apache/cassandra/service/paxos/PaxosState.java
+++ b/src/java/org/apache/cassandra/service/paxos/PaxosState.java
@@ -739,8 +739,8 @@ public class PaxosState implements PaxosOperationLock
{
KeyspaceMetadata ksm = Schema.instance.getKeyspaceMetadata(mutation.getKeyspaceName());
if (ksm != null && ksm.params.replicationType.isTracked()
- && MutationTrackingService.instance.startWriting(mutation))
- MutationTrackingService.instance.finishWriting(mutation);
+ && MutationTrackingService.instance().startWriting(mutation))
+ MutationTrackingService.instance().finishWriting(mutation);
}
}
diff --git a/src/java/org/apache/cassandra/service/paxos/PrepareRefreshForwardHandler.java b/src/java/org/apache/cassandra/service/paxos/PrepareRefreshForwardHandler.java
index 578ec53d9c..9a1f4a56e9 100644
--- a/src/java/org/apache/cassandra/service/paxos/PrepareRefreshForwardHandler.java
+++ b/src/java/org/apache/cassandra/service/paxos/PrepareRefreshForwardHandler.java
@@ -90,7 +90,7 @@ public class PrepareRefreshForwardHandler implements IVerbHandler<PrepareRefresh
}
Token token = request.commit.partitionKey().getToken();
- MutationId mutationId = MutationTrackingService.instance.nextMutationId(ksName, token);
+ MutationId mutationId = MutationTrackingService.instance().nextMutationId(ksName, token);
Mutation mutationWithId = request.commit.makeMutation(mutationId);
Committed commitWithId = new Commit.Committed(request.commit.ballot, mutationWithId);
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java b/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
index cebac78909..5a8607b3c5 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
@@ -254,7 +254,7 @@ public abstract class PartialTrackedRead
*/
void augment(ShortMutationId mutationId)
{
- Mutation mutation = MutationJournal.instance.read(mutationId);
+ Mutation mutation = MutationJournal.instance().read(mutationId);
if (mutation == null)
{
logger.error("Could not augment read with mutation not present in journal {}", mutationId);
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/ReadReconcileAck.java b/src/java/org/apache/cassandra/service/reads/tracked/ReadReconcileAck.java
index bfa716a9e0..6d4688cad3 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/ReadReconcileAck.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/ReadReconcileAck.java
@@ -26,6 +26,7 @@ import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.net.IVerbHandler;
+import org.apache.cassandra.replication.MutationTrackingService;
/**
* Notifies the reconcile coordinator (data node) that this node has received
@@ -50,6 +51,7 @@ public class ReadReconcileAck
public static final IVerbHandler<ReadReconcileAck> verbHandler = message ->
{
+ MutationTrackingService.ensureEnabled();
ReadReconcileAck notify = message.payload;
logger.trace("Received reconcile ack from {}, for {}", message.from(), notify.readId);
ReadReconciliations.instance.acceptSyncAck(message.from(), notify.readId);
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/ReadReconciliations.java b/src/java/org/apache/cassandra/service/reads/tracked/ReadReconciliations.java
index 33d3c68e2f..48722ea512 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/ReadReconciliations.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/ReadReconciliations.java
@@ -222,7 +222,7 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
if (logId.hostId() != LOCAL_NODE)
continue;
- MutationTrackingService.instance.collectRemotelyMissingMutations(ourOffsets, remoteNodes, missingOffsets);
+ MutationTrackingService.instance().collectRemotelyMissingMutations(ourOffsets, remoteNodes, missingOffsets);
// we don't listen to or block on delivery of these mutations, intentionally
missingOffsets.forEach(ReadReconciliations::push);
missingOffsets.clear();
@@ -239,7 +239,7 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
boolean acceptRemoteSummary(MutationSummary summary, int remoteNode)
{
Log2OffsetsMap.Mutable missingMutations = new Log2OffsetsMap.Mutable();
- MutationTrackingService.instance.collectLocallyMissingMutations(summary, missingMutations);
+ MutationTrackingService.instance().collectLocallyMissingMutations(summary, missingMutations);
// don't request what's already been requested for other remote summaries
// TODO (consider): rely entirely on IncomingMutations deduplication instead
@@ -363,7 +363,7 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
private boolean complete()
{
if (isDataNode())
- MutationTrackingService.instance.localReads().acknowledgeReconcile(id, augmentingOffsets());
+ MutationTrackingService.instance().localReads().acknowledgeReconcile(id, augmentingOffsets());
else
MessagingService.instance().send(Message.out(Verb.READ_RECONCILE_ACK, new ReadReconcileAck(id)), host(dataNode));
@@ -392,7 +392,7 @@ public class ReadReconciliations implements ExpiredStatePurger.Expireable
Offsets.Mutable toPull = new Offsets.Mutable(offsets.logId());
for (ShortMutationId id : offsets)
- if (MutationTrackingService.instance.registerMutationCallback(id, callback))
+ if (MutationTrackingService.instance().registerMutationCallback(id, callback))
toPull.add(id.offset());
if (!toPull.isEmpty())
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/TrackedRead.java b/src/java/org/apache/cassandra/service/reads/tracked/TrackedRead.java
index 261f7aef6d..108262667c 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/TrackedRead.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/TrackedRead.java
@@ -288,7 +288,7 @@ public abstract class TrackedRead<E extends Endpoints<E>, P extends ReplicaPlan.
logger.trace("Locally coordinating {}", readId);
Stage.READ.submit(() -> {
AsyncPromise<TrackedDataResponse> promise =
- MutationTrackingService.instance.localReads().beginRead(readId, ClusterMetadata.current(), command, consistencyLevel, summaryNodes, requestTime, completer);
+ MutationTrackingService.instance().localReads().beginRead(readId, ClusterMetadata.current(), command, consistencyLevel, summaryNodes, requestTime, completer);
promise.addCallback((response, error) -> {
if (error != null)
{
@@ -474,7 +474,7 @@ public abstract class TrackedRead<E extends Endpoints<E>, P extends ReplicaPlan.
// TODO This is 1000% the wrong deadline?
Dispatcher.RequestTime requestTime = new Dispatcher.RequestTime(message.createdAtNanos());
AsyncPromise<TrackedDataResponse> promise =
- MutationTrackingService.instance
+ MutationTrackingService.instance()
.localReads()
.beginRead(readId, metadata, command, consistencyLevel, summaryNodes, requestTime, TrackedLocalReads.Completer.DEFAULT);
promise.addCallback((response, error) -> {
@@ -491,7 +491,7 @@ public abstract class TrackedRead<E extends Endpoints<E>, P extends ReplicaPlan.
@Override
public Future<? extends IReadResponse> executeLocally(Request request, ClusterMetadata metadata, RequestTime requestTime)
{
- return MutationTrackingService.instance
+ return MutationTrackingService.instance()
.localReads()
.beginRead(readId, metadata, command, consistencyLevel, summaryNodes, requestTime, TrackedLocalReads.Completer.DEFAULT);
}
@@ -610,6 +610,7 @@ public abstract class TrackedRead<E extends Endpoints<E>, P extends ReplicaPlan.
@Override
protected void performRead(Message<Request> message, ClusterMetadata metadata)
{
+ MutationTrackingService.ensureEnabled();
message.payload.executeLocally(message, metadata);
}
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/TrackedSummaryResponse.java b/src/java/org/apache/cassandra/service/reads/tracked/TrackedSummaryResponse.java
index ba48d84f3d..5e740f84ff 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/TrackedSummaryResponse.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/TrackedSummaryResponse.java
@@ -25,6 +25,7 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.replication.MutationSummary;
+import org.apache.cassandra.replication.MutationTrackingService;
import java.io.IOException;
@@ -52,6 +53,7 @@ public class TrackedSummaryResponse implements IReadResponse
public static final IVerbHandler<TrackedSummaryResponse> verbHandler = message ->
{
+ MutationTrackingService.ensureEnabled();
TrackedSummaryResponse response = message.payload;
if (logger.isTraceEnabled())
logger.trace("Received summary {} from {}, for {}", response.summary, message.from(), response.readId);
diff --git a/src/java/org/apache/cassandra/streaming/StreamSession.java b/src/java/org/apache/cassandra/streaming/StreamSession.java
index 91489e5b85..3e32e3ba51 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -512,12 +512,12 @@ public class StreamSession
private LogTransferTask createLogTransferTask()
{
- ReconciledLogSnapshot reconciled = MutationTrackingService.instance.snapshotReconciledLogs();
+ ReconciledLogSnapshot reconciled = MutationTrackingService.instance().snapshotReconciledLogs();
// TODO: consider tradeoffs of eagerly reading the index of each segment and filtering out the ones that
// only contain fully reconciled ids vs just filtering out fully reconciled when reading out of the
// snapshot for streaming
- MutationJournal.Snapshot snapshot = MutationJournal.instance.snapshot();
+ MutationJournal.Snapshot snapshot = MutationJournal.instance().snapshot();
try
{
// TODO: grab references to all current segments and the relevant reconciled sets
@@ -763,9 +763,11 @@ public class StreamSession
sessionFailed();
break;
case MUTATION_LOG_STREAM:
+ MutationTrackingService.ensureEnabled();
receiveMutationLog((IncomingMutationLogStreamMessage) message);
break;
case MUTATION_LOG_RECEIVED:
+ MutationTrackingService.ensureEnabled();
mutationLogReceived((MutationLogReceivedMessage) message);
break;
default:
diff --git a/src/java/org/apache/cassandra/streaming/messages/IncomingMutationLogStreamMessage.java b/src/java/org/apache/cassandra/streaming/messages/IncomingMutationLogStreamMessage.java
index fea433419e..5e79bbfac9 100644
--- a/src/java/org/apache/cassandra/streaming/messages/IncomingMutationLogStreamMessage.java
+++ b/src/java/org/apache/cassandra/streaming/messages/IncomingMutationLogStreamMessage.java
@@ -87,7 +87,7 @@ public class IncomingMutationLogStreamMessage extends MutationLogStreamMessage
mutation.apply();
}
- MutationTrackingService.instance.recordFullyReconciledOffsets(header.reconciled);
+ MutationTrackingService.instance().recordFullyReconciledOffsets(header.reconciled);
return new IncomingMutationLogStreamMessage(header, session);
}
diff --git a/src/java/org/apache/cassandra/tcm/Startup.java b/src/java/org/apache/cassandra/tcm/Startup.java
index 5a9553fd45..5a462665cc 100644
--- a/src/java/org/apache/cassandra/tcm/Startup.java
+++ b/src/java/org/apache/cassandra/tcm/Startup.java
@@ -161,14 +161,16 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
.withStorage(LogStorage.SystemKeyspace)
.afterReplay(Startup::scrubDataDirectories,
(metadata) -> StorageService.instance.registerMBeans(),
- MutationTrackingService.instance::start)
+ MutationTrackingService::start)
.withDefaultListeners();
ClusterMetadataService.setInstance(new ClusterMetadataService(new UniformRangePlacement(),
wrapProcessor,
ClusterMetadataService::state,
logSpec));
ClusterMetadataService.instance().log().ready();
- MutationTrackingService.instance.registerTCMListener();
+
+ if (DatabaseDescriptor.getMutationTrackingEnabled())
+ MutationTrackingService.instance().registerMetadataListener();
NodeId nodeId = ClusterMetadata.current().myNodeId();
UUID currentHostId = SystemKeyspace.getLocalHostId();
@@ -290,7 +292,7 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
.withInitialState(emptyFromSystemTables)
.afterReplay(Startup::scrubDataDirectories,
(metadata) -> StorageService.instance.registerMBeans(),
- MutationTrackingService.instance::start)
+ MutationTrackingService::start)
.withStorage(LogStorage.SystemKeyspace)
.withDefaultListeners();
@@ -300,12 +302,17 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
logSpec));
ClusterMetadataService.instance().log().ready();
- MutationTrackingService.instance.registerTCMListener();
+
+ if (DatabaseDescriptor.getMutationTrackingEnabled())
+ MutationTrackingService.instance().registerMetadataListener();
+
initMessaging.run();
try
{
CommitLog.instance.recoverSegmentsOnDisk();
- MutationJournal.instance.replayStaticSegments();
+
+ if (DatabaseDescriptor.getMutationTrackingEnabled())
+ MutationJournal.instance().replayStaticSegments();
}
catch (IOException e)
{
@@ -400,7 +407,7 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
LocalLog.LogSpec logSpec = LocalLog.logSpec()
.afterReplay(Startup::scrubDataDirectories,
(_metadata) -> StorageService.instance.registerMBeans(),
- MutationTrackingService.instance::start)
+ MutationTrackingService::start)
.withPreviousState(prev)
.withInitialState(metadata)
.withStorage(LogStorage.SystemKeyspace)
@@ -413,7 +420,10 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
logSpec));
ClusterMetadataService.instance().log().ready();
- MutationTrackingService.instance.registerTCMListener();
+
+ if (DatabaseDescriptor.getMutationTrackingEnabled())
+ MutationTrackingService.instance().registerMetadataListener();
+
initMessaging.run();
ClusterMetadataService.instance().forceSnapshot(metadata.forceEpoch(metadata.nextEpoch()));
ClusterMetadataService.instance().triggerSnapshot();
diff --git a/test/conf/cassandra-jmx-disabled-sslconfig.yaml b/test/conf/cassandra-jmx-disabled-sslconfig.yaml
index 4708909c96..84a0456491 100644
--- a/test/conf/cassandra-jmx-disabled-sslconfig.yaml
+++ b/test/conf/cassandra-jmx-disabled-sslconfig.yaml
@@ -152,4 +152,8 @@ accord:
enabled: true
journal_directory: build/test/cassandra/accord_journal
queue_shard_count: 2
- command_store_shard_count: 4
\ No newline at end of file
+ command_store_shard_count: 4
+
+mutation_tracking:
+ enabled: true
+ journal_directory: build/test/cassandra/mutation_journal
diff --git a/test/conf/cassandra-jmx-pem-sslconfig.yaml b/test/conf/cassandra-jmx-pem-sslconfig.yaml
index 0e8f204791..62e0251e29 100644
--- a/test/conf/cassandra-jmx-pem-sslconfig.yaml
+++ b/test/conf/cassandra-jmx-pem-sslconfig.yaml
@@ -152,4 +152,8 @@ accord:
enabled: true
journal_directory: build/test/cassandra/accord_journal
queue_shard_count: 2
- command_store_shard_count: 4
\ No newline at end of file
+ command_store_shard_count: 4
+
+mutation_tracking:
+ enabled: true
+ journal_directory: build/test/cassandra/mutation_journal
diff --git a/test/conf/cassandra-jmx-sslconfig-with-passwordfile.yaml b/test/conf/cassandra-jmx-sslconfig-with-passwordfile.yaml
index 6f66b04bbc..6af50e4e1c 100644
--- a/test/conf/cassandra-jmx-sslconfig-with-passwordfile.yaml
+++ b/test/conf/cassandra-jmx-sslconfig-with-passwordfile.yaml
@@ -77,4 +77,8 @@ accord:
enabled: true
journal_directory: build/test/cassandra/accord_journal
queue_shard_count: 2
- command_store_shard_count: 4
\ No newline at end of file
+ command_store_shard_count: 4
+
+mutation_tracking:
+ enabled: true
+ journal_directory: build/test/cassandra/mutation_journal
diff --git a/test/conf/cassandra-jmx-sslconfig.yaml b/test/conf/cassandra-jmx-sslconfig.yaml
index cd2f9fb036..617530a539 100644
--- a/test/conf/cassandra-jmx-sslconfig.yaml
+++ b/test/conf/cassandra-jmx-sslconfig.yaml
@@ -77,4 +77,8 @@ accord:
enabled: true
journal_directory: build/test/cassandra/accord_journal
queue_shard_count: 2
- command_store_shard_count: 4
\ No newline at end of file
+ command_store_shard_count: 4
+
+mutation_tracking:
+ enabled: true
+ journal_directory: build/test/cassandra/mutation_journal
diff --git a/test/conf/cassandra-mtls.yaml b/test/conf/cassandra-mtls.yaml
index e80c2ac296..9593d25896 100644
--- a/test/conf/cassandra-mtls.yaml
+++ b/test/conf/cassandra-mtls.yaml
@@ -89,3 +89,5 @@ authenticator:
validator_class_name: org.apache.cassandra.auth.SpiffeCertificateValidator
accord:
journal_directory: build/test/cassandra/accord_journal
+mutation_tracking:
+ journal_directory: build/test/cassandra/mutation_journal
diff --git a/test/conf/cassandra-murmur.yaml b/test/conf/cassandra-murmur.yaml
index 75a208e5fa..8472d89b97 100644
--- a/test/conf/cassandra-murmur.yaml
+++ b/test/conf/cassandra-murmur.yaml
@@ -44,3 +44,5 @@ sasi_indexes_enabled: true
materialized_views_enabled: true
accord:
journal_directory: build/test/cassandra/accord_journal
+mutation_tracking:
+ journal_directory: build/test/cassandra/mutation_journal
diff --git a/test/conf/cassandra-old.yaml b/test/conf/cassandra-old.yaml
index 000df1148f..be6cc43a26 100644
--- a/test/conf/cassandra-old.yaml
+++ b/test/conf/cassandra-old.yaml
@@ -57,4 +57,6 @@ internode_recv_buff_size_in_bytes: 5
max_hint_window_in_ms: 10800000
cache_load_timeout_seconds: 35
accord:
- journal_directory: build/test/cassandra/accord_journal
+ journal_directory: build/test/cassandra/accord_journal
+mutation_tracking:
+ journal_directory: build/test/cassandra/mutation_journal
diff --git a/test/conf/cassandra-pem-jks-sslcontextfactory.yaml b/test/conf/cassandra-pem-jks-sslcontextfactory.yaml
index 5c27c7b9f9..503a712b7c 100644
--- a/test/conf/cassandra-pem-jks-sslcontextfactory.yaml
+++ b/test/conf/cassandra-pem-jks-sslcontextfactory.yaml
@@ -151,3 +151,5 @@ materialized_views_enabled: true
file_cache_enabled: true
accord:
journal_directory: build/test/cassandra/accord_journal
+mutation_tracking:
+ journal_directory: build/test/cassandra/mutation_journal
\ No newline at end of file
diff --git a/test/conf/cassandra-pem-sslcontextfactory-invalidconfiguration.yaml b/test/conf/cassandra-pem-sslcontextfactory-invalidconfiguration.yaml
index 58c2e9293a..ca0e996a66 100644
--- a/test/conf/cassandra-pem-sslcontextfactory-invalidconfiguration.yaml
+++ b/test/conf/cassandra-pem-sslcontextfactory-invalidconfiguration.yaml
@@ -148,3 +148,5 @@ materialized_views_enabled: true
file_cache_enabled: true
accord:
journal_directory: build/test/cassandra/accord_journal
+mutation_tracking:
+ journal_directory: build/test/cassandra/mutation_journal
\ No newline at end of file
diff --git a/test/conf/cassandra-pem-sslcontextfactory.yaml b/test/conf/cassandra-pem-sslcontextfactory.yaml
index ef68105e94..3166f03aaa 100644
--- a/test/conf/cassandra-pem-sslcontextfactory.yaml
+++ b/test/conf/cassandra-pem-sslcontextfactory.yaml
@@ -152,3 +152,5 @@ materialized_views_enabled: true
file_cache_enabled: true
accord:
journal_directory: build/test/cassandra/accord_journal
+mutation_tracking:
+ journal_directory: build/test/cassandra/mutation_journal
\ No newline at end of file
diff --git a/test/conf/cassandra-seeds.yaml b/test/conf/cassandra-seeds.yaml
index 1049e27fa8..a8f6db6067 100644
--- a/test/conf/cassandra-seeds.yaml
+++ b/test/conf/cassandra-seeds.yaml
@@ -43,3 +43,5 @@ user_defined_functions_enabled: true
scripted_user_defined_functions_enabled: false
accord:
journal_directory: build/test/cassandra/accord_journal
+mutation_tracking:
+ journal_directory: build/test/cassandra/mutation_journal
\ No newline at end of file
diff --git a/test/conf/cassandra-sslcontextfactory-invalidconfiguration.yaml b/test/conf/cassandra-sslcontextfactory-invalidconfiguration.yaml
index 26db768f5c..ebfd0978b1 100644
--- a/test/conf/cassandra-sslcontextfactory-invalidconfiguration.yaml
+++ b/test/conf/cassandra-sslcontextfactory-invalidconfiguration.yaml
@@ -82,3 +82,5 @@ materialized_views_enabled: true
file_cache_enabled: true
accord:
journal_directory: build/test/cassandra/accord_journal
+mutation_tracking:
+ journal_directory: build/test/cassandra/mutation_journal
\ No newline at end of file
diff --git a/test/conf/cassandra-sslcontextfactory.yaml b/test/conf/cassandra-sslcontextfactory.yaml
index 153d2f9245..978b464ba3 100644
--- a/test/conf/cassandra-sslcontextfactory.yaml
+++ b/test/conf/cassandra-sslcontextfactory.yaml
@@ -85,3 +85,5 @@ materialized_views_enabled: true
file_cache_enabled: true
accord:
journal_directory: build/test/cassandra/accord_journal
+mutation_tracking:
+ journal_directory: build/test/cassandra/mutation_journal
\ No newline at end of file
diff --git a/test/conf/cassandra.yaml b/test/conf/cassandra.yaml
index 07200f2d16..cc05eae41d 100644
--- a/test/conf/cassandra.yaml
+++ b/test/conf/cassandra.yaml
@@ -73,6 +73,10 @@ write_tombstone_warn_threshold: 1000
min_tracked_partition_size: 1MiB
min_tracked_partition_tombstone_count: 500
+mutation_tracking:
+ enabled: true
+ journal_directory: build/test/cassandra/mutation_journal
+
accord:
enabled: true
journal_directory: build/test/cassandra/accord_journal
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
index a62e51453e..6407c461c3 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
@@ -789,7 +789,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
CassandraDaemon.getInstanceForTesting().migrateSystemDataIfNeeded();
CommitLog.instance.start();
- MutationJournal.instance.start();
+ MutationJournal.start();
SnapshotManager.instance.start(false);
SnapshotManager.instance.clearExpiredSnapshots();
@@ -840,7 +840,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
NodeId self = ClusterMetadata.current().myNodeId();
if (self != null)
AccordService.localStartup(self);
- MutationJournal.instance.replayStaticSegments();
+
+ if ((Boolean) config.get("mutation_tracking.enabled"))
+ MutationJournal.instance().replayStaticSegments();
}
catch (IOException e)
{
@@ -1057,8 +1059,8 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
});
// The periodic tasks in MTS might hit the CommitLog, so make sure those shut down first...
- error = parallelRun(error, executor, MutationTrackingService.instance::shutdownBlocking);
- error = parallelRun(error, executor, MutationJournal.instance::shutdownBlocking);
+ error = parallelRun(error, executor, MutationTrackingService::shutdown);
+ error = parallelRun(error, executor, MutationJournal::shutdown);
// CommitLog must shut down after Stage, or threads from the latter may attempt to use the former.
// (ex. A Mutation stage thread may attempt to add a mutation to the CommitLog.)
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
index e16320823e..3a6c487b2c 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
@@ -33,6 +33,7 @@ import com.vdurmont.semver4j.Semver;
import org.apache.cassandra.config.AccordSpec;
import org.apache.cassandra.config.CassandraRelevantProperties;
+import org.apache.cassandra.config.MutationTrackingSpec;
import org.apache.cassandra.config.OptionaldPositiveInt;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInstanceConfig;
@@ -43,6 +44,7 @@ import org.apache.cassandra.locator.NetworkTopologyProximity;
import org.apache.cassandra.locator.SimpleSeedProvider;
import static org.apache.cassandra.config.CassandraRelevantProperties.DTEST_ACCORD_ENABLED;
+import static org.apache.cassandra.config.CassandraRelevantProperties.DTEST_MUTATION_TRACKING_ENABLED;
public class InstanceConfig implements IInstanceConfig
{
@@ -78,6 +80,7 @@ public class InstanceConfig implements IInstanceConfig
String hints_directory,
String cdc_raw_directory,
AccordSpec accord,
+ MutationTrackingSpec mutationTracking,
Collection<String> initial_token,
int storage_port,
int native_transport_port,
@@ -104,6 +107,8 @@ public class InstanceConfig implements IInstanceConfig
.set("accord.command_store_shard_count", accord.command_store_shard_count.toString())
.set("accord.expire_txn", accord.expire_txn)
.set("accord.enable_virtual_debug_only_keyspace", "true")
+ .set("mutation_tracking.enabled", mutationTracking.enabled)
+ .set("mutation_tracking.journal_directory", mutationTracking.journal_directory)
.set("partitioner", "org.apache.cassandra.dht.Murmur3Partitioner")
.set("start_native_transport", true)
.set("concurrent_writes", 2)
@@ -332,6 +337,9 @@ public class InstanceConfig implements IInstanceConfig
accordSpec.journal_directory = String.format("%s/node%d/accord_journal", root, nodeNum);
accordSpec.queue_shard_count = new OptionaldPositiveInt(2);
accordSpec.command_store_shard_count = new OptionaldPositiveInt(4);
+ MutationTrackingSpec mutationTrackingSpec = new MutationTrackingSpec();
+ mutationTrackingSpec.enabled = DTEST_MUTATION_TRACKING_ENABLED.getBoolean();
+ mutationTrackingSpec.journal_directory = String.format("%s/node%d/mutation_journal", root, nodeNum);
return new InstanceConfig(nodeNum,
networkTopology,
provisionStrategy.ipAddress(nodeNum),
@@ -346,6 +354,7 @@ public class InstanceConfig implements IInstanceConfig
String.format("%s/node%d/hints", root, nodeNum),
String.format("%s/node%d/cdc", root, nodeNum),
accordSpec,
+ mutationTrackingSpec,
tokens,
provisionStrategy.storagePort(nodeNum),
provisionStrategy.nativeTransportPort(nodeNum),
diff --git a/test/distributed/org/apache/cassandra/distributed/test/MutationTrackingMigrationTest.java b/test/distributed/org/apache/cassandra/distributed/test/MutationTrackingMigrationTest.java
index e4b352101a..51344e893d 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/MutationTrackingMigrationTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/MutationTrackingMigrationTest.java
@@ -112,7 +112,7 @@ public class MutationTrackingMigrationTest extends TestBaseImpl
{
int[] count = new int[1];
- MutationJournal.instance.snapshot().readAll((segment, position, key, buffer, version) -> {
+ MutationJournal.instance().snapshot().readAll((segment, position, key, buffer, version) -> {
count[0]++;
});
diff --git a/test/distributed/org/apache/cassandra/distributed/test/TrackedBatchTest.java b/test/distributed/org/apache/cassandra/distributed/test/TrackedBatchTest.java
index 9b683cf7e2..2fd1aff4a7 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/TrackedBatchTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/TrackedBatchTest.java
@@ -44,8 +44,7 @@ public class TrackedBatchTest extends TestBaseImpl
{
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true"))
+ .with(Feature.GOSSIP))
.start())
{
// Create tracked keyspace
@@ -89,8 +88,7 @@ public class TrackedBatchTest extends TestBaseImpl
{
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true"))
+ .with(Feature.GOSSIP))
.start())
{
// Create tracked keyspace
diff --git a/test/distributed/org/apache/cassandra/distributed/test/TransientRangeMovement2Test.java b/test/distributed/org/apache/cassandra/distributed/test/TransientRangeMovement2Test.java
index 1fd7636b10..29926b36fe 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/TransientRangeMovement2Test.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/TransientRangeMovement2Test.java
@@ -57,7 +57,6 @@ public class TransientRangeMovement2Test extends TestBaseImpl
.withTokenSupplier(new OPPTokens())
.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(4, "dc0", "rack0"))
.withConfig(conf -> conf.set("transient_replication_enabled","true")
- .set("mutation_tracking_enabled", "true")
.set("partitioner", "OrderPreservingPartitioner") // just makes it easier to read the tokens in the log
.with(Feature.NETWORK, Feature.GOSSIP))
.start()))
@@ -112,7 +111,6 @@ public class TransientRangeMovement2Test extends TestBaseImpl
.withTokenSupplier(new OPPTokens())
.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(4, "dc0", "rack0"))
.withConfig(conf -> conf.set("transient_replication_enabled","true")
- .set("mutation_tracking_enabled", "true")
.set("partitioner", "OrderPreservingPartitioner") // just makes it easier to read the tokens in the log
.set("hinted_handoff_enabled", "false")
.with(Feature.NETWORK, Feature.GOSSIP))
@@ -162,7 +160,6 @@ public class TransientRangeMovement2Test extends TestBaseImpl
.withTokenSupplier(new OPPTokens())
.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(4, "dc0", "rack0"))
.withConfig(conf -> conf.set("transient_replication_enabled","true")
- .set("mutation_tracking_enabled", "true")
.set("partitioner", "OrderPreservingPartitioner")
.set("hinted_handoff_enabled", "false")
.with(Feature.NETWORK, Feature.GOSSIP))
diff --git a/test/distributed/org/apache/cassandra/distributed/test/TransientRangeMovementTest.java b/test/distributed/org/apache/cassandra/distributed/test/TransientRangeMovementTest.java
index 059e86e382..0aa41df55f 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/TransientRangeMovementTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/TransientRangeMovementTest.java
@@ -73,7 +73,6 @@ public class TransientRangeMovementTest extends TestBaseImpl
.withTokenSupplier(new OPPTokens())
.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(4, "dc0", "rack0"))
.withConfig(conf -> conf.set("transient_replication_enabled","true")
- .set("mutation_tracking_enabled", "true")
.set("partitioner", "OrderPreservingPartitioner")
.set("hinted_handoff_enabled", "false")
.with(Feature.NETWORK, Feature.GOSSIP))
@@ -111,7 +110,6 @@ public class TransientRangeMovementTest extends TestBaseImpl
.withTokenSupplier(new OPPTokensReplace())
.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(4, "dc0", "rack0"))
.withConfig(conf -> conf.set("transient_replication_enabled","true")
- .set("mutation_tracking_enabled", "true")
.set("partitioner", "OrderPreservingPartitioner")
.with(Feature.NETWORK, Feature.GOSSIP))
.start()))
@@ -173,7 +171,6 @@ public class TransientRangeMovementTest extends TestBaseImpl
.withTokenSupplier(new OPPTokens())
.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(4, "dc0", "rack0"))
.withConfig(conf -> conf.set("transient_replication_enabled","true")
- .set("mutation_tracking_enabled", "true")
.set("partitioner", "OrderPreservingPartitioner")
.set("hinted_handoff_enabled", "false")
.with(Feature.NETWORK, Feature.GOSSIP))
diff --git a/test/distributed/org/apache/cassandra/distributed/test/WitnessAlwaysReadsFullReplicaTest.java b/test/distributed/org/apache/cassandra/distributed/test/WitnessAlwaysReadsFullReplicaTest.java
index fea9b29544..342beea4bd 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/WitnessAlwaysReadsFullReplicaTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/WitnessAlwaysReadsFullReplicaTest.java
@@ -82,7 +82,6 @@ public class WitnessAlwaysReadsFullReplicaTest extends SharedClusterTestBase
setupCluster(6, builder -> builder.withRacks(2, 1, 3) // 2 DCs with 3 nodes each
.withConfig(cfg -> cfg.with(Feature.NETWORK)
.with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
.set("transient_replication_enabled", "true")
.set("dynamic_snitch", false) // Disable dynamic snitch
.set("node_proximity", TransientFirstProximity.class.getName()))); // Use our custom proximity
diff --git a/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkWithMutationTrackingTest.java b/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkWithMutationTrackingTest.java
index d5ea279521..d5aeea8f38 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkWithMutationTrackingTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkWithMutationTrackingTest.java
@@ -94,7 +94,6 @@ public class MultiNodeTableWalkWithMutationTrackingTest extends MultiNodeTableWa
protected void clusterConfig(IInstanceConfig c)
{
super.clusterConfig(c);
- c.set("mutation_tracking_enabled", true);
IGNORED_ISSUES.remove(AF_MULTI_NODE_MULTI_COLUMN_AND_NODE_LOCAL_WRITES);
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/TrackedHostReplacementTest.java b/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/TrackedHostReplacementTest.java
index 589738a009..df8315272d 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/TrackedHostReplacementTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/TrackedHostReplacementTest.java
@@ -64,7 +64,7 @@ public class TrackedHostReplacementTest extends TestBaseImpl
cluster.stream()
.filter(node -> !node.isShutdown())
.forEach(node -> node.runOnInstance(() -> {
- MutationTrackingService.instance.pauseOffsetBroadcast(pause);
+ MutationTrackingService.instance().pauseOffsetBroadcast(pause);
}));
}
@@ -334,7 +334,7 @@ public class TrackedHostReplacementTest extends TestBaseImpl
{
cluster.stream().filter(node -> !node.isShutdown()).forEach( node -> {
node.runOnInstance(() -> {
- MutationJournal.instance.advanceSegment();
+ MutationJournal.instance().advanceSegment();
});
});
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/MetadataChangeSimulationTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/MetadataChangeSimulationTest.java
index e2e0ddcbcd..b13cab2660 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/log/MetadataChangeSimulationTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/log/MetadataChangeSimulationTest.java
@@ -89,7 +89,6 @@ public class MetadataChangeSimulationTest extends CMSTestBase
{
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
DatabaseDescriptor.setTransientReplicationEnabledUnsafe(true);
- DatabaseDescriptor.setMutationTrackingEnabled(true);
}
@Test
diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/NTSSimulationTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/NTSSimulationTest.java
index 76498d69e5..c0ce61226b 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/log/NTSSimulationTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/log/NTSSimulationTest.java
@@ -40,7 +40,6 @@ public class NTSSimulationTest extends CMSTestBase
{
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
DatabaseDescriptor.setTransientReplicationEnabledUnsafe(true);
- DatabaseDescriptor.setMutationTrackingEnabled(true);
}
@Before
diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/OperationalEquivalenceTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/OperationalEquivalenceTest.java
index 4e81b2afe7..2f3e9f81ef 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/log/OperationalEquivalenceTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/log/OperationalEquivalenceTest.java
@@ -62,7 +62,6 @@ public class OperationalEquivalenceTest extends CMSTestBase
static
{
DatabaseDescriptor.setTransientReplicationEnabledUnsafe(true);
- DatabaseDescriptor.setMutationTrackingEnabled(true);
}
@Test
diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/SimpleStrategySimulationTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/SimpleStrategySimulationTest.java
index c68674baab..b8f98f68a0 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/log/SimpleStrategySimulationTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/log/SimpleStrategySimulationTest.java
@@ -40,7 +40,6 @@ public class SimpleStrategySimulationTest extends CMSTestBase
{
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
DatabaseDescriptor.setTransientReplicationEnabledUnsafe(true);
- DatabaseDescriptor.setMutationTrackingEnabled(true);
}
@Before
diff --git a/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingServiceTest.java b/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingServiceTest.java
index 4f75a00ea8..f305a635d4 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingServiceTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingServiceTest.java
@@ -46,7 +46,7 @@ public class MutationTrackingServiceTest extends TestBaseImpl
// Test that shards remain the same object (no reconfiguration)
Boolean shardsUnchanged = cluster.get(1).callOnInstance(() -> {
- var service = MutationTrackingService.instance;
+ var service = MutationTrackingService.instance();
Object initialShards = MutationTrackingService.TestAccess.getKeyspaceShards(service, KS_NAME);
Object newShards = MutationTrackingService.TestAccess.getKeyspaceShards(service, KS_NAME);
return initialShards == newShards; // Same object reference
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingBounceTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingBounceTest.java
index 3e0a1ad260..935cd3db65 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingBounceTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingBounceTest.java
@@ -74,7 +74,7 @@ public class MutationTrackingBounceTest extends FuzzTestBase
history.insert(pk);
if (++counter % 10 == 0)
- cluster.get(1).runOnInstance(() -> MutationJournal.instance.closeCurrentSegmentForTestingIfNonEmpty());
+ cluster.get(1).runOnInstance(() -> MutationJournal.instance().closeCurrentSegmentForTestingIfNonEmpty());
}
ClusterUtils.stopUnchecked(cluster.get(1));
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingBounce_ValidateRunnable.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingBounce_ValidateRunnable.java
index 0afb402b1d..88a16e41af 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingBounce_ValidateRunnable.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingBounce_ValidateRunnable.java
@@ -44,7 +44,7 @@ public class MutationTrackingBounce_ValidateRunnable implements IIsolatedExecuto
public void run()
{
AtomicInteger counter = new AtomicInteger();
- MutationJournal.instance.replay(new DeserializedRecordConsumer<>(MutationJournal.MutationSerializer.INSTANCE)
+ MutationJournal.instance().replay(new DeserializedRecordConsumer<>(MutationJournal.MutationSerializer.INSTANCE)
{
Set<ShortMutationId> seen = new HashSet<>();
@Override
@@ -55,7 +55,7 @@ public class MutationTrackingBounce_ValidateRunnable implements IIsolatedExecuto
for (PartitionUpdate partitionUpdate : mutation.getPartitionUpdates())
{
- if (!MutationTrackingService.instance.createSummaryForKey(partitionUpdate.partitionKey(), partitionUpdate.metadata().id, false)
+ if (!MutationTrackingService.instance().createSummaryForKey(partitionUpdate.partitionKey(), partitionUpdate.metadata().id, false)
.contains(key))
{
throw new AssertionError(String.format("Mutation %s should have been witnessed (%s)", mutation, key));
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingCasForwardingTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingCasForwardingTest.java
index 4276067307..5fd1e8393e 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingCasForwardingTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingCasForwardingTest.java
@@ -84,7 +84,6 @@ public class MutationTrackingCasForwardingTest extends TestBaseImpl
try (Cluster cluster = Cluster.build(4)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
.with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
.set("transient_replication_enabled", "true")
.set("paxos_variant", paxosVariant))
.start())
@@ -183,7 +182,7 @@ public class MutationTrackingCasForwardingTest extends TestBaseImpl
if (replicaNode != blockedReplicaNode) {
final int nodeNum = replicaNode;
cluster.get(nodeNum).runOnInstance(() -> {
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(
Util.dk(1),
ColumnFamilyStore.getIfExists(keyspaceName, "tbl").metadata.id,
true);
@@ -198,7 +197,7 @@ public class MutationTrackingCasForwardingTest extends TestBaseImpl
// The blocked replica node should not have the mutation yet due to filtering
cluster.get(blockedReplicaNode).runOnInstance(() -> {
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(
Util.dk(1),
ColumnFamilyStore.getIfExists(keyspaceName, "tbl").metadata.id,
true);
@@ -229,7 +228,7 @@ public class MutationTrackingCasForwardingTest extends TestBaseImpl
// Now the blocked replica node should have the mutation propagated via mutation tracking
// Note: When using replica coordinator, propagation behavior may differ
cluster.get(blockedReplicaNode).runOnInstance(() -> {
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(
Util.dk(1),
ColumnFamilyStore.getIfExists(keyspaceName, "tbl").metadata.id,
true);
@@ -249,7 +248,7 @@ public class MutationTrackingCasForwardingTest extends TestBaseImpl
final int nodeNum = replicaNodes[i];
final int arrayIndex = i;
cluster.get(nodeNum).runOnInstance(() -> {
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(
Util.dk(1),
ColumnFamilyStore.getIfExists(keyspaceName, "tbl").metadata.id,
true);
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingCommitAndPrepareTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingCommitAndPrepareTest.java
index 4b4e81cf20..dc18e12fb6 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingCommitAndPrepareTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingCommitAndPrepareTest.java
@@ -76,7 +76,6 @@ public class MutationTrackingCommitAndPrepareTest extends TestBaseImpl
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
.with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
.set("paxos_variant", "v2_without_linearizable_reads"))
.start())
{
@@ -137,7 +136,7 @@ public class MutationTrackingCommitAndPrepareTest extends TestBaseImpl
{
TableMetadata table = Schema.instance.getTableMetadata(keyspaceName, "tbl");
DecoratedKey dk = Murmur3Partitioner.instance.decorateKey(ByteBufferUtil.bytes(1));
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(dk, table.id, false);
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(dk, table.id, false);
logger.info("Node 1 mutation summary: {} coordinator logs", summary.size());
@@ -164,7 +163,7 @@ public class MutationTrackingCommitAndPrepareTest extends TestBaseImpl
{
TableMetadata table = Schema.instance.getTableMetadata(keyspaceName, "tbl");
DecoratedKey dk = Murmur3Partitioner.instance.decorateKey(ByteBufferUtil.bytes(1));
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(dk, table.id, true);
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(dk, table.id, true);
logger.info("Node {} mutation summary size: {}", nodeId, summary.size());
// All nodes should eventually have the mutation
@@ -194,7 +193,6 @@ public class MutationTrackingCommitAndPrepareTest extends TestBaseImpl
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
.with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
.set("transient_replication_enabled", "true")
.set("paxos_variant", "v2_without_linearizable_reads"))
.start())
@@ -259,7 +257,7 @@ public class MutationTrackingCommitAndPrepareTest extends TestBaseImpl
int nodeId = i;
cluster.get(nodeId).runOnInstance(() ->
{
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(
Util.dk(1),
ColumnFamilyStore.getIfExists(keyspaceName, "tbl").metadata.id,
true);
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingFeatureFlagTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingFeatureFlagTest.java
new file mode 100644
index 0000000000..ea2ed39cde
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingFeatureFlagTest.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.tracking;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.YamlConfigurationLoader;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.db.virtual.MutationTrackingTables;
+import org.apache.cassandra.db.virtual.SystemViewsKeyspace;
+import org.apache.cassandra.db.virtual.VirtualTable;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.utils.AssertionUtils;
+
+import static org.apache.cassandra.distributed.api.Feature.NETWORK;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+
+import static org.apache.cassandra.replication.MutationTrackingService.DISABLED_MESSAGE;
+
+public class MutationTrackingFeatureFlagTest extends TestBaseImpl
+{
+ @Test
+ public void shouldHideMutationTracking() throws IOException
+ {
+ try (Cluster cluster = builder().withNodes(3).withConfig(c -> c.with(NETWORK).set("mutation_tracking.enabled", false)).start())
+ {
+ // We shouldn't be able to create a tracked keyspace:
+ Assertions.assertThatThrownBy(() -> cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE +
+ " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3}" +
+ " AND replication_type='tracked'"))
+ .has(AssertionUtils.isThrowableInstanceof(InvalidRequestException.class))
+ .hasMessage(DISABLED_MESSAGE);
+
+ cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3} AND replication_type='untracked'");
+
+ // We shouldn't be able to alter a keyspace to make it tracked:
+ Assertions.assertThatThrownBy(() -> cluster.schemaChange("ALTER KEYSPACE " + KEYSPACE + " WITH replication_type='tracked'"))
+ .has(AssertionUtils.isThrowableInstanceof(InvalidRequestException.class))
+ .hasMessage(DISABLED_MESSAGE);
+
+ // Mutation tracking system keyspace tables should not be present:
+ assertNull(cluster.get(1).callOnInstance(() -> Schema.instance.getTableMetadata(SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.HOST_LOG_ID)));
+ assertNull(cluster.get(1).callOnInstance(() -> Schema.instance.getTableMetadata(SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.SHARDS)));
+ assertNull(cluster.get(1).callOnInstance(() -> Schema.instance.getTableMetadata(SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.COORDINATOR_LOGS)));
+
+ // Make sure virtual tables don't exist:
+ IIsolatedExecutor.SerializableCallable<Stream<VirtualTable>> hasMutationTrackingTables =
+ () -> SystemViewsKeyspace.instance.tables().stream().filter(t -> t.getClass().equals(MutationTrackingTables.MutationTrackingShardsTable.class)
+ || t.getClass().equals(MutationTrackingTables.MutationJournalTable.class));
+ List<VirtualTable> tables = cluster.get(1).callOnInstance(hasMutationTrackingTables).collect(Collectors.toList());
+ assertEquals(Collections.emptyList(), tables);
+
+ String journalDirectoryPath = cluster.get(1).callOnInstance(DatabaseDescriptor::getMutationTrackingJournalDirectory);
+ File journalDirectory = new File(journalDirectoryPath);
+ assertFalse("Journal directory should not be created when mutation tracking is disabled", journalDirectory.exists());
+ }
+ }
+
+ @Test
+ public void shouldFailOnBounceWhenDisabledAfterEnabled() throws Exception
+ {
+ try (Cluster cluster = builder().withNodes(1)
+ .withConfig(c -> c.with(Feature.NETWORK).set("mutation_tracking.enabled", true))
+ .start())
+ {
+ // Create a tracked keyspace/table while MT is enabled:
+ cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1} AND replication_type='tracked'"));
+ cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (k int primary key, v int)"));
+
+ IInvokableInstance bouncingNode = cluster.get(1);
+ bouncingNode.shutdown().get();
+ bouncingNode.config().set("mutation_tracking.enabled", false);
+ Config badConfig = YamlConfigurationLoader.fromMap(bouncingNode.config().getParams(), true, Config.class);
+ Config.setOverrideLoadConfig(() -> badConfig);
+
+ Assertions.assertThatThrownBy(() -> bouncingNode.startup(cluster))
+ .has(AssertionUtils.isThrowableInstanceof(IllegalStateException.class))
+ .hasMessage(DISABLED_MESSAGE);
+ }
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingLogPersisterTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingLogPersisterTest.java
index 5d0785454e..9d8349a639 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingLogPersisterTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingLogPersisterTest.java
@@ -46,8 +46,7 @@ public class MutationTrackingLogPersisterTest extends FuzzTestBase
{
try (Cluster cluster = builder().withNodes(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true"))
+ .with(Feature.GOSSIP))
.start())
{
int tables = 3;
@@ -81,16 +80,16 @@ public class MutationTrackingLogPersisterTest extends FuzzTestBase
history.insert(pk);
if (++counter % 10 == 0)
- cluster.get(1).runOnInstance(() -> MutationJournal.instance.closeCurrentSegmentForTestingIfNonEmpty());
+ cluster.get(1).runOnInstance(() -> MutationJournal.instance().closeCurrentSegmentForTestingIfNonEmpty());
}
cluster.forEach(i -> i.nodetoolResult("flush", KEYSPACE).asserts().success());
- cluster.forEach(i -> i.runOnInstance(() -> MutationTrackingService.instance.persistLogStateForTesting()));
- cluster.forEach(i -> i.runOnInstance(() -> MutationTrackingService.instance.broadcastOffsetsForTesting()));
- cluster.forEach(i -> i.runOnInstance(() -> MutationTrackingService.instance.persistLogStateForTesting()));
+ cluster.forEach(i -> i.runOnInstance(() -> MutationTrackingService.instance().persistLogStateForTesting()));
+ cluster.forEach(i -> i.runOnInstance(() -> MutationTrackingService.instance().broadcastOffsetsForTesting()));
+ cluster.forEach(i -> i.runOnInstance(() -> MutationTrackingService.instance().persistLogStateForTesting()));
cluster.forEach(i -> i.runOnInstance(() -> {
- int staticSegments = MutationJournal.instance.countStaticSegmentsForTesting();
+ int staticSegments = MutationJournal.instance().countStaticSegmentsForTesting();
Assert.assertEquals("Expected no static segments after log persister runs", 0, staticSegments);
}));
});
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingMetricsTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingMetricsTest.java
index 54d3c6b973..7b659087ad 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingMetricsTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingMetricsTest.java
@@ -57,9 +57,9 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
cluster.schemaChange(String.format(CREATE_TABLE, KEYSPACE));
// Get initial write-time discovery counts on all nodes
- long initialNode1Count = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance.writeTimeOffsetsDiscovered.getCount());
- long initialNode2Count = cluster.get(2).callOnInstance(() -> MutationTrackingMetrics.instance.writeTimeOffsetsDiscovered.getCount());
- long initialNode3Count = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance.writeTimeOffsetsDiscovered.getCount());
+ long initialNode1Count = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance().writeTimeOffsetsDiscovered.getCount());
+ long initialNode2Count = cluster.get(2).callOnInstance(() -> MutationTrackingMetrics.instance().writeTimeOffsetsDiscovered.getCount());
+ long initialNode3Count = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance().writeTimeOffsetsDiscovered.getCount());
// Perform writes with QUORUM - each write goes to at least 2 replicas
int numWrites = 10;
@@ -75,18 +75,18 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
.atMost(Duration.ofSeconds(5))
.pollInterval(Duration.ofMillis(100))
.until(() -> {
- long node1Delta = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance.writeTimeOffsetsDiscovered.getCount()) - initialNode1Count;
- long node2Delta = cluster.get(2).callOnInstance(() -> MutationTrackingMetrics.instance.writeTimeOffsetsDiscovered.getCount()) - initialNode2Count;
- long node3Delta = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance.writeTimeOffsetsDiscovered.getCount()) - initialNode3Count;
+ long node1Delta = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance().writeTimeOffsetsDiscovered.getCount()) - initialNode1Count;
+ long node2Delta = cluster.get(2).callOnInstance(() -> MutationTrackingMetrics.instance().writeTimeOffsetsDiscovered.getCount()) - initialNode2Count;
+ long node3Delta = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance().writeTimeOffsetsDiscovered.getCount()) - initialNode3Count;
long totalDiscovered = node1Delta + node2Delta + node3Delta;
return node1Delta > 0 && node2Delta > 0 && node3Delta > 0 && totalDiscovered >= (long) numWrites * 3;
});
// Verify final counts
- long afterNode1Count = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance.writeTimeOffsetsDiscovered.getCount());
- long afterNode2Count = cluster.get(2).callOnInstance(() -> MutationTrackingMetrics.instance.writeTimeOffsetsDiscovered.getCount());
- long afterNode3Count = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance.writeTimeOffsetsDiscovered.getCount());
+ long afterNode1Count = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance().writeTimeOffsetsDiscovered.getCount());
+ long afterNode2Count = cluster.get(2).callOnInstance(() -> MutationTrackingMetrics.instance().writeTimeOffsetsDiscovered.getCount());
+ long afterNode3Count = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance().writeTimeOffsetsDiscovered.getCount());
long node1Delta = afterNode1Count - initialNode1Count;
long node2Delta = afterNode2Count - initialNode2Count;
@@ -124,7 +124,7 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
cluster.schemaChange(String.format(CREATE_TABLE, KEYSPACE));
// Record initial broadcast metrics on receiving node 3 since we are next going to block this node to from receiving mutations
- long initialNode3Count = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance.broadcastOffsetsDiscovered.getCount());
+ long initialNode3Count = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance().broadcastOffsetsDiscovered.getCount());
// Block node 3 from receiving mutation writes (but allow broadcast messages)
cluster.filters().verbs(Verb.MUTATION_REQ.id).to(3).drop();
@@ -146,7 +146,7 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
// Broadcast offsets from node 1 to other nodes
// This tells node 3 about mutations it's missing
- cluster.get(1).runOnInstance(() -> MutationTrackingService.instance.broadcastOffsetsForTesting());
+ cluster.get(1).runOnInstance(() -> MutationTrackingService.instance().broadcastOffsetsForTesting());
// Wait for broadcasts to propagate to node 3
long[] previousCount = {0};
@@ -154,7 +154,7 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
.atMost(Duration.ofSeconds(5))
.pollInterval(Duration.ofMillis(100))
.until(() -> {
- long currentCount = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance.broadcastOffsetsDiscovered.getCount());
+ long currentCount = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance().broadcastOffsetsDiscovered.getCount());
boolean hasDiscoveredOffsets = currentCount > initialNode3Count;
boolean isStable = hasDiscoveredOffsets && currentCount == previousCount[0];
previousCount[0] = currentCount;
@@ -162,10 +162,10 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
});
// Get the count after first broadcast
- long afterFirstBroadcast = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance.broadcastOffsetsDiscovered.getCount());
+ long afterFirstBroadcast = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance().broadcastOffsetsDiscovered.getCount());
// Broadcast the same offsets again (duplicate) - should NOT increment metric
- cluster.get(1).runOnInstance(() -> MutationTrackingService.instance.broadcastOffsetsForTesting());
+ cluster.get(1).runOnInstance(() -> MutationTrackingService.instance().broadcastOffsetsForTesting());
// Wait for duplicate broadcast to propagate, then verify metric stayed the same
// We poll to ensure the broadcast had time to arrive, then check it didn't increment
@@ -174,7 +174,7 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
.atMost(Duration.ofSeconds(2))
.pollInterval(Duration.ofMillis(100))
.until(() -> {
- long count = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance.broadcastOffsetsDiscovered.getCount());
+ long count = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance().broadcastOffsetsDiscovered.getCount());
return count == afterFirstBroadcast; // Should remain at the same value (duplicate doesn't increment)
});
@@ -203,7 +203,7 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
.isEqualTo(numWrites);
// Check metrics after reconciliation - if reconciliation worked, broadcasts happened
- long afterNode3Count = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance.broadcastOffsetsDiscovered.getCount());
+ long afterNode3Count = cluster.get(3).callOnInstance(() -> MutationTrackingMetrics.instance().broadcastOffsetsDiscovered.getCount());
long node3Delta = afterNode3Count - initialNode3Count;
// Node 3 was blocked before and now must have applied broadcast offsets
@@ -226,7 +226,7 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
cluster.schemaChange(String.format(CREATE_TABLE, KEYSPACE));
// Get initial metric value from coordinator node
- long initialSize = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance.readSummarySize.getCount());
+ long initialSize = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance().readSummarySize.getCount());
// Insert test data
int numWrites = 10;
@@ -247,7 +247,7 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
// Verify metric incremented by at least twice the number of reads as
// each read creates TWO summaries: initial (before read) + secondary (after read)
// This is to detect concurrent writes during read execution for proper reconciliation
- long afterSize = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance.readSummarySize.getCount());
+ long afterSize = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance().readSummarySize.getCount());
long delta = afterSize - initialSize;
assertThat(delta)
@@ -269,7 +269,7 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
cluster.schemaChange(String.format(CREATE_TABLE, KEYSPACE));
// Get initial unreconciled count (should be 0)
- long initialCount = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance.unreconciledMutationCount.getValue());
+ long initialCount = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance().unreconciledMutationCount.getValue());
assertThat(initialCount)
.as("Initial unreconciled count should be 0")
.isEqualTo(0L);
@@ -286,7 +286,7 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
}
// Node 1 should now have unreconciled mutations (since node 3 didn't get them)
- long afterWrites = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance.unreconciledMutationCount.getValue());
+ long afterWrites = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance().unreconciledMutationCount.getValue());
assertThat(afterWrites)
.as("Expected %d unreconciled mutations (node 3 blocked)", numWrites)
.isEqualTo((long) numWrites);
@@ -306,12 +306,12 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
.atMost(Duration.ofSeconds(5))
.pollInterval(Duration.ofMillis(100))
.until(() -> {
- long count = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance.unreconciledMutationCount.getValue());
+ long count = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance().unreconciledMutationCount.getValue());
return count == 0;
});
// Verify reconciliation actually happened
- long afterReconcile = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance.unreconciledMutationCount.getValue());
+ long afterReconcile = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance().unreconciledMutationCount.getValue());
assertThat(afterReconcile)
.as("Unreconciled count should be 0 after reconciliation")
.isEqualTo(0L);
@@ -331,7 +331,7 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
cluster.schemaChange(String.format(CREATE_TABLE, KEYSPACE));
// Get initial disk space - would be 2 * 1024 * 1024 as 2 segements are allocated by default
- long initialSpace = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance.journalDiskSpaceUsed.getValue());
+ long initialSpace = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance().journalDiskSpaceUsed.getValue());
// Write enough data to fill 1MiB segment and force new segment creation
int numWrites = 200;
@@ -343,11 +343,11 @@ public class MutationTrackingMetricsTest extends TestBaseImpl
// Close segment every 20 writes to create multiple segments
if (i % 20 == 0 && i > 0)
- cluster.get(1).runOnInstance(() -> MutationJournal.instance.closeCurrentSegmentForTestingIfNonEmpty());
+ cluster.get(1).runOnInstance(() -> MutationJournal.instance().closeCurrentSegmentForTestingIfNonEmpty());
}
// Verify disk space increased
- long afterWrites = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance.journalDiskSpaceUsed.getValue());
+ long afterWrites = cluster.get(1).callOnInstance(() -> MutationTrackingMetrics.instance().journalDiskSpaceUsed.getValue());
assertThat(afterWrites)
.as("Disk space should increase after writes: before=%d", initialSpace)
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPartitionReadTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPartitionReadTest.java
index 119ec61523..3c1d16bfb0 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPartitionReadTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPartitionReadTest.java
@@ -49,7 +49,6 @@ public class MutationTrackingPartitionReadTest extends TestBaseImpl
cluster = Cluster.build()
.withNodes(REPLICAS)
.withConfig(cfg -> cfg.with(Feature.NETWORK, Feature.GOSSIP)
- .set("mutation_tracking_enabled", true)
.set("hinted_handoff_enabled", false))
.start();
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java
index ae49c00383..0e83bc07ac 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java
@@ -88,7 +88,7 @@ public class MutationTrackingPendingReadTest
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
.with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
+ .set("mutation_tracking.enabled", true)
.set("write_request_timeout", "1000ms"))
.start())
{
@@ -121,7 +121,7 @@ public class MutationTrackingPendingReadTest
Assert.assertEquals(1, secondIds.offsetCount());
// create a mutation
- MutationId id = MutationTrackingService.instance.nextMutationId(keyspaceName, dk.getToken());
+ MutationId id = MutationTrackingService.instance().nextMutationId(keyspaceName, dk.getToken());
SimpleBuilders.MutationBuilder builder = new SimpleBuilders.MutationBuilder(id, keyspaceName, dk);
PartitionUpdate.SimpleBuilder tableBuilder = builder.update(metadata);
tableBuilder.row(bytes(1)).add("v", 1);
@@ -144,7 +144,7 @@ public class MutationTrackingPendingReadTest
initialSummary = command.createMutationSummary(false);
ReadExecutionController controller = command.executionController(false);
- MutationTrackingService.instance.startWriting(mutation);
+ MutationTrackingService.instance().startWriting(mutation);
read = command.beginTrackedRead(controller);
// Create another summary once initial data has been read fully. We do this to catch
@@ -204,7 +204,7 @@ public class MutationTrackingPendingReadTest
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
.with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
+ .set("mutation_tracking.enabled", true)
.set("write_request_timeout", "1000ms"))
.start())
{
@@ -229,12 +229,12 @@ public class MutationTrackingPendingReadTest
int nowInSeconds = (int) FBUtilities.nowInSeconds();
SinglePartitionReadCommand command = SinglePartitionReadCommand.fullPartitionRead(metadata, nowInSeconds, dk);
-// try (ListeningPendingRead pendingRead = (ListeningPendingRead) MutationTrackingService.instance.startReading(command))
+// try (ListeningPendingRead pendingRead = (ListeningPendingRead) MutationTrackingService.instance().startReading(command))
// {
// Assert.assertTrue(pendingRead.mutationIds().isEmpty());
//
// // create and apply a mutation
-// MutationId id = MutationTrackingService.instance.nextMutationId(keyspaceName, dk.getToken());
+// MutationId id = MutationTrackingService.instance().nextMutationId(keyspaceName, dk.getToken());
// SimpleBuilders.MutationBuilder builder = new SimpleBuilders.MutationBuilder(id, keyspaceName, dk);
// PartitionUpdate.SimpleBuilder tableBuilder = builder.update(metadata);
// tableBuilder.row(bytes(1)).add("v", 1);
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingRangeReadTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingRangeReadTest.java
index 194637253d..f6ad1899a2 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingRangeReadTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingRangeReadTest.java
@@ -46,9 +46,7 @@ public class MutationTrackingRangeReadTest extends TestBaseImpl
{
cluster = Cluster.build()
.withNodes(REPLICAS)
- .withConfig(cfg -> cfg.with(Feature.NETWORK, Feature.GOSSIP)
- .set("mutation_tracking_enabled", true)
- .set("hinted_handoff_enabled", false))
+ .withConfig(cfg -> cfg.with(Feature.NETWORK, Feature.GOSSIP).set("hinted_handoff_enabled", false))
.start();
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingReadReconciliationTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingReadReconciliationTest.java
index 37f6c6a46f..5689bcca33 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingReadReconciliationTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingReadReconciliationTest.java
@@ -113,9 +113,8 @@ public class MutationTrackingReadReconciliationTest extends TestBaseImpl
{
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
- .set("write_request_timeout", "1000ms"))
+ .with(Feature.GOSSIP)
+ .set("write_request_timeout", "1000ms"))
.start())
{
String keyspaceName = "basic_reconciliation_test";
@@ -191,9 +190,8 @@ public class MutationTrackingReadReconciliationTest extends TestBaseImpl
{
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
- .set("write_request_timeout", "1000ms"))
+ .with(Feature.GOSSIP)
+ .set("write_request_timeout", "1000ms"))
.start())
{
String keyspaceName = "basic_reconciliation_test";
@@ -268,9 +266,8 @@ public class MutationTrackingReadReconciliationTest extends TestBaseImpl
{
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
- .set("write_request_timeout", "1000ms"))
+ .with(Feature.GOSSIP)
+ .set("write_request_timeout", "1000ms"))
.start())
{
String keyspaceName = "basic_reconciliation_test";
@@ -344,9 +341,8 @@ public class MutationTrackingReadReconciliationTest extends TestBaseImpl
{
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
- .set("write_request_timeout", "1000ms"))
+ .with(Feature.GOSSIP)
+ .set("write_request_timeout", "1000ms"))
.start())
{
String keyspaceName = "basic_reconciliation_test";
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingTest.java
index b160458cd9..32fdd25055 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingTest.java
@@ -70,9 +70,7 @@ public class MutationTrackingTest extends TestBaseImpl
public void testBasicWritePath() throws Throwable
{
try (Cluster cluster = Cluster.build(3)
- .withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true"))
+ .withConfig(cfg -> cfg.with(Feature.NETWORK).with(Feature.GOSSIP))
.start())
{
@@ -94,7 +92,7 @@ public class MutationTrackingTest extends TestBaseImpl
cluster.get(1).runOnInstance(() -> {
TableMetadata table = Schema.instance.getTableMetadata(keyspaceName, "tbl");
DecoratedKey dk = Murmur3Partitioner.instance.decorateKey(ByteBufferUtil.bytes(1));
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(dk, table.id, false);
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(dk, table.id, false);
CoordinatorLogId logId = getOnlyLogId(summary);
Offsets summaryIds = summaryIdSpace(summary.get(logId));
assertEquals(1, summaryIds.offsetCount());
@@ -117,9 +115,7 @@ public class MutationTrackingTest extends TestBaseImpl
private void testWitnessPaxosReads(String paxosVariant) throws Throwable
{
try (Cluster cluster = Cluster.build(3)
- .withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
+ .withConfig(cfg -> cfg.with(Feature.NETWORK).with(Feature.GOSSIP)
.set("transient_replication_enabled", "true")
.set("paxos_variant", paxosVariant))
.start())
@@ -142,13 +138,13 @@ public class MutationTrackingTest extends TestBaseImpl
// Two nodes should know about the mutation
for (int i = 1; i <= 2; i++)
cluster.get(i).runOnInstance(() -> {
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(Util.dk(1), ColumnFamilyStore.getIfExists(keyspaceName, "tbl").metadata.id, true);
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(Util.dk(1), ColumnFamilyStore.getIfExists(keyspaceName, "tbl").metadata.id, true);
assertEquals(1, summary.size());
});
// Filter should stop the witness from getting the mutation so we can test pushing the mutation summary to the witness
cluster.get(3).runOnInstance(() -> {
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(Util.dk(1), ColumnFamilyStore.getIfExists(keyspaceName, "tbl").metadata.id, true);
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(Util.dk(1), ColumnFamilyStore.getIfExists(keyspaceName, "tbl").metadata.id, true);
assertEquals(0, summary.size());
});
@@ -180,7 +176,7 @@ public class MutationTrackingTest extends TestBaseImpl
// The read at SERIAL should propagate the mutation to the witness
cluster.get(3).runOnInstance(() -> {
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(Util.dk(1), ColumnFamilyStore.getIfExists(keyspaceName, "tbl").metadata.id, true);
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(Util.dk(1), ColumnFamilyStore.getIfExists(keyspaceName, "tbl").metadata.id, true);
assertEquals(1, summary.size());
});
}
@@ -203,10 +199,8 @@ public class MutationTrackingTest extends TestBaseImpl
private void testWitnessBatchWrites(boolean logged) throws Throwable
{
try (Cluster cluster = Cluster.build(3)
- .withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
- .set("transient_replication_enabled", "true"))
+ .withConfig(cfg -> cfg.with(Feature.NETWORK).with(Feature.GOSSIP)
+ .set("transient_replication_enabled", "true"))
.start())
{
cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH replication = " +
@@ -239,7 +233,7 @@ public class MutationTrackingTest extends TestBaseImpl
cluster.get(1).runOnInstance(() -> {
TableMetadata table = Schema.instance.getTableMetadata(keyspaceName, "tbl");
DecoratedKey dk = Murmur3Partitioner.instance.decorateKey(ByteBufferUtil.bytes(1));
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(dk, table.id, false);
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(dk, table.id, false);
CoordinatorLogId logId = getOnlyLogId(summary);
Offsets summaryIds = summaryIdSpace(summary.get(logId));
@@ -292,11 +286,9 @@ public class MutationTrackingTest extends TestBaseImpl
{
String paxosVariantFinal = paxosVariant == null ? "v1" : paxosVariant;
try (Cluster cluster = Cluster.build(3)
- .withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
- .set("transient_replication_enabled", "true")
- .set("paxos_variant", paxosVariantFinal))
+ .withConfig(cfg -> cfg.with(Feature.NETWORK).with(Feature.GOSSIP)
+ .set("transient_replication_enabled", "true")
+ .set("paxos_variant", paxosVariantFinal))
.start())
{
cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH replication = " +
@@ -328,7 +320,7 @@ public class MutationTrackingTest extends TestBaseImpl
cluster.get(1).runOnInstance(() -> {
TableMetadata table = Schema.instance.getTableMetadata(keyspaceName, "tbl");
DecoratedKey dk = Murmur3Partitioner.instance.decorateKey(ByteBufferUtil.bytes(1));
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(dk, table.id, false);
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(dk, table.id, false);
CoordinatorLogId logId = getOnlyLogId(summary);
Offsets summaryIds = summaryIdSpace(summary.get(logId));
@@ -358,9 +350,8 @@ public class MutationTrackingTest extends TestBaseImpl
{
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
- .set("write_request_timeout", "1000ms"))
+ .with(Feature.GOSSIP)
+ .set("write_request_timeout", "1000ms"))
.start())
{
cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH replication = " +
@@ -394,9 +385,8 @@ public class MutationTrackingTest extends TestBaseImpl
{
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
- .set("write_request_timeout", "1000ms"))
+ .with(Feature.GOSSIP)
+ .set("write_request_timeout", "1000ms"))
.start())
{
String keyspaceName = KEYSPACE;
@@ -411,7 +401,7 @@ public class MutationTrackingTest extends TestBaseImpl
cluster.filters().verbs(Verb.MUTATION_REQ.id).to(3).drop();
// pause reconciler temporarily
- cluster.get(1).runOnInstance(() -> MutationTrackingService.instance.pauseActiveReconciler());
+ cluster.get(1).runOnInstance(() -> MutationTrackingService.instance().pauseActiveReconciler());
// issue a write - should fail on node 3
cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.tbl (k, v) VALUES (1, 1)"), ConsistencyLevel.QUORUM);
@@ -422,21 +412,21 @@ public class MutationTrackingTest extends TestBaseImpl
{
TableMetadata table = Schema.instance.getTableMetadata(keyspaceName, "tbl");
DecoratedKey dk = Murmur3Partitioner.instance.decorateKey(ByteBufferUtil.bytes(1));
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(dk, table.id, false);
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(dk, table.id, false);
CoordinatorLogId logId = getOnlyLogId(summary);
Assert.assertEquals(1, summary.get(logId).unreconciled.offsetCount());
Assert.assertEquals(0, summary.get(logId).reconciled.offsetCount());
});
// resume the reconciler
- cluster.get(1).runOnInstance(() -> MutationTrackingService.instance.resumeActiveReconciler());
+ cluster.get(1).runOnInstance(() -> MutationTrackingService.instance().resumeActiveReconciler());
Thread.sleep(1000); // wait for reconiciler to do its job
cluster.get(1).runOnInstance(() ->
{
TableMetadata table = Schema.instance.getTableMetadata(keyspaceName, "tbl");
DecoratedKey dk = Murmur3Partitioner.instance.decorateKey(ByteBufferUtil.bytes(1));
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(dk, table.id, false);
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(dk, table.id, false);
CoordinatorLogId logId = getOnlyLogId(summary);
Assert.assertEquals(0, summary.get(logId).unreconciled.offsetCount());
Assert.assertEquals(1, summary.get(logId).reconciled.offsetCount());
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingUtils.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingUtils.java
index 2cf832fb08..6f0bff174f 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingUtils.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingUtils.java
@@ -124,7 +124,7 @@ public class MutationTrackingUtils
public static MutationSummary summaryForKey(String keyspaceName, String tableName, DecoratedKey dk)
{
TableMetadata table = Schema.instance.getTableMetadata(keyspaceName, tableName);
- return MutationTrackingService.instance.createSummaryForKey(dk, table.id, false);
+ return MutationTrackingService.instance().createSummaryForKey(dk, table.id, false);
}
public static MutationSummary summaryForTable(String keyspaceName, String tableName)
@@ -161,7 +161,7 @@ public class MutationTrackingUtils
public static MutationSummary summaryForRange(String keyspaceName, String tableName, Range<Token> range)
{
TableMetadata table = Schema.instance.getTableMetadata(keyspaceName, tableName);
- return MutationTrackingService.instance.createSummaryForRange(range, table.id, false);
+ return MutationTrackingService.instance().createSummaryForRange(range, table.id, false);
}
public static Offsets summaryIdSpace(CoordinatorSummary summary)
@@ -296,7 +296,7 @@ public class MutationTrackingUtils
public static Mutation createMutation(TableMetadata tableMetadata, int k, int v)
{
DecoratedKey key = tableMetadata.partitioner.decorateKey(ByteBufferUtil.bytes(k));
- MutationId mutationId = MutationTrackingService.instance.nextMutationId(tableMetadata.keyspace, key.getToken());
+ MutationId mutationId = MutationTrackingService.instance().nextMutationId(tableMetadata.keyspace, key.getToken());
SimpleBuilders.MutationBuilder builder = new SimpleBuilders.MutationBuilder(mutationId, tableMetadata.keyspace, key);
PartitionUpdate.SimpleBuilder partition = builder.update(tableMetadata);
partition.row().add("v", v);
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingWriteForwardingTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingWriteForwardingTest.java
index 6209c1c1b6..fd534888eb 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingWriteForwardingTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingWriteForwardingTest.java
@@ -64,9 +64,8 @@ public class MutationTrackingWriteForwardingTest extends TestBaseImpl
// TODO: disable background reconciliation so we can test that writes are reconciling immediately
try (Cluster cluster = Cluster.build(NODES)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true")
- .set("write_request_timeout", "1000ms"))
+ .with(Feature.GOSSIP)
+ .set("write_request_timeout", "1000ms"))
.withNodeIdTopology(topology)
.start())
{
@@ -97,7 +96,7 @@ public class MutationTrackingWriteForwardingTest extends TestBaseImpl
Token token = DatabaseDescriptor.getPartitioner().getMinimumToken();
Range<Token> fullRange = new Range<>(token, token);
TableId tableId = Schema.instance.getTableMetadata(keyspaceName, tableName).id;
- MutationSummary summary = MutationTrackingService.instance.createSummaryForRange(fullRange, tableId, true);
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForRange(fullRange, tableId, true);
return summary.reconciledIds();
});
@@ -105,7 +104,7 @@ public class MutationTrackingWriteForwardingTest extends TestBaseImpl
Token token = DatabaseDescriptor.getPartitioner().getMinimumToken();
Range<Token> fullRange = new Range<>(token, token);
TableId tableId = Schema.instance.getTableMetadata(keyspaceName, tableName).id;
- MutationSummary summary = MutationTrackingService.instance.createSummaryForRange(fullRange, tableId, true);
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForRange(fullRange, tableId, true);
return summary.unreconciledIds();
});
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/OffsetBroadcastTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/OffsetBroadcastTest.java
index 5cd9c8cb2c..b14ec37285 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/OffsetBroadcastTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/OffsetBroadcastTest.java
@@ -44,9 +44,7 @@ public class OffsetBroadcastTest extends TestBaseImpl
public void testBroadcastOffsets() throws Throwable
{
try (Cluster cluster = Cluster.build(3)
- .withConfig(cfg -> cfg.with(Feature.NETWORK)
- .with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", "true"))
+ .withConfig(cfg -> cfg.with(Feature.NETWORK).with(Feature.GOSSIP))
.start())
{
@@ -61,14 +59,14 @@ public class OffsetBroadcastTest extends TestBaseImpl
cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.tbl (k, v) VALUES (1, 1)"), ConsistencyLevel.QUORUM);
for (int i = 1; i <= cluster.size(); ++i)
- cluster.get(i).runOnInstance(() -> MutationTrackingService.instance.broadcastOffsetsForTesting());
+ cluster.get(i).runOnInstance(() -> MutationTrackingService.instance().broadcastOffsetsForTesting());
for (int i = 1; i <= cluster.size(); ++i)
{
cluster.get(i).runOnInstance(() -> {
TableMetadata table = Schema.instance.getTableMetadata(keyspaceName, "tbl");
DecoratedKey dk = Murmur3Partitioner.instance.decorateKey(ByteBufferUtil.bytes(1));
- MutationSummary summary = MutationTrackingService.instance.createSummaryForKey(dk, table.id, false);
+ MutationSummary summary = MutationTrackingService.instance().createSummaryForKey(dk, table.id, false);
MutationSummary.CoordinatorSummary coordinatorSummary = summary.get(getOnlyLogId(summary));
Assert.assertEquals(1, coordinatorSummary.reconciled.offsetCount());
Assert.assertEquals(0, coordinatorSummary.unreconciled.offsetCount());
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedTransferTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedTransferTestBase.java
index 5b5a3b5a45..e5ce559e8c 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedTransferTestBase.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedTransferTestBase.java
@@ -82,7 +82,6 @@ public abstract class TrackedTransferTestBase extends TestBaseImpl
protected static final Consumer<IInstanceConfig> CONFIG = cfg -> cfg.with(Feature.NETWORK)
.with(Feature.GOSSIP)
- .set("mutation_tracking_enabled", true)
.set("write_request_timeout", "1000ms")
.set("autocompaction_on_startup_enabled", false)
.set("repair_request_timeout", "2s")
@@ -222,7 +221,7 @@ public abstract class TrackedTransferTestBase extends TestBaseImpl
MutationSummary summary;
try
{
- summary = MutationTrackingService.instance.createSummaryForKey(key, tableId, false);
+ summary = MutationTrackingService.instance().createSummaryForKey(key, tableId, false);
}
catch (UnknownShardException e)
{
@@ -394,8 +393,8 @@ public abstract class TrackedTransferTestBase extends TestBaseImpl
// Activation ID must be persisted and broadcast across all peers in the cluster for any to mark as persisted + reconciled
cluster.forEach(i -> {
i.runOnInstance(() -> {
- MutationTrackingService.instance.persistLogStateForTesting();
- MutationTrackingService.instance.broadcastOffsetsForTesting();
+ MutationTrackingService.instance().persistLogStateForTesting();
+ MutationTrackingService.instance().broadcastOffsetsForTesting();
});
});
diff --git a/test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java b/test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java
index a2975090b7..d22a469c2d 100644
--- a/test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java
+++ b/test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java
@@ -134,7 +134,7 @@ public abstract class CommitLogStressTest
}
SchemaLoader.loadSchema();
- MutationJournal.instance.start();
+ MutationJournal.start();
SchemaLoader.schemaDefinition(""); // leave def. blank to maintain old behaviour
CommitLog.instance.stopUnsafe(true);
diff --git a/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java b/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java
index 2134a1a954..33008c7641 100644
--- a/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java
+++ b/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java
@@ -923,7 +923,7 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
// Enable mutation tracking and transient replication for tracked modes
if (builder.replicationConfig.isTracked())
- config.set("mutation_tracking_enabled", "true");
+ config.set("mutation_tracking.enabled", true);
if (builder.replicationConfig.withWitnesses())
config.set("transient_replication_enabled", "true");
diff --git a/test/unit/org/apache/cassandra/ServerTestUtils.java b/test/unit/org/apache/cassandra/ServerTestUtils.java
index 84e42694aa..448ac0468e 100644
--- a/test/unit/org/apache/cassandra/ServerTestUtils.java
+++ b/test/unit/org/apache/cassandra/ServerTestUtils.java
@@ -156,7 +156,7 @@ public final class ServerTestUtils
daemonInitialization();
// Need to happen after daemonInitialization for config to be set, but before CFS initialization
- MutationJournal.instance.start();
+ MutationJournal.start();
if (isServerPrepared)
return;
diff --git a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java
index 5a6e2086ff..f24bfa21ce 100644
--- a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java
+++ b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java
@@ -168,6 +168,7 @@ public class DatabaseDescriptorRefTest
"org.apache.cassandra.config.GuardrailsOptions$ConsistencyLevels",
"org.apache.cassandra.config.GuardrailsOptions$TableProperties",
"org.apache.cassandra.config.JMXServerOptions",
+ "org.apache.cassandra.config.MutationTrackingSpec",
"org.apache.cassandra.config.OptionaldPositiveInt",
"org.apache.cassandra.config.ParameterizedClass",
"org.apache.cassandra.config.RepairConfig",
diff --git a/test/unit/org/apache/cassandra/db/CleanupTransientTest.java b/test/unit/org/apache/cassandra/db/CleanupTransientTest.java
index 6c83418f91..b560f6a300 100644
--- a/test/unit/org/apache/cassandra/db/CleanupTransientTest.java
+++ b/test/unit/org/apache/cassandra/db/CleanupTransientTest.java
@@ -75,7 +75,6 @@ public class CleanupTransientTest extends CassandraTestBase
@BeforeClass
public static void setup() throws Exception
{
- DatabaseDescriptor.setMutationTrackingEnabled(true);
DatabaseDescriptor.setTransientReplicationEnabledUnsafe(true);
StorageService ss = StorageService.instance;
final int RING_SIZE = 2;
@@ -89,7 +88,7 @@ public class CleanupTransientTest extends CassandraTestBase
endpointTokens.add(RandomPartitioner.instance.midpoint(RandomPartitioner.MINIMUM, new RandomPartitioner.BigIntegerToken(RandomPartitioner.MAXIMUM)));
Util.createInitialRing(endpointTokens, keyTokens, hosts, hostIds, RING_SIZE);
- MutationJournal.instance.start();
+ MutationJournal.start();
SchemaLoader.createKeyspace(KEYSPACE1,
KeyspaceParams.simpleWitness("2/1"),
SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD1),
@@ -160,7 +159,7 @@ public class CleanupTransientTest extends CassandraTestBase
.clustering(COLUMN)
.add(colName, VALUE)
.build();
- mutation.withMutationId(MutationTrackingService.instance.nextMutationId(cfs.metadata().keyspace, mutation.key().getToken()))
+ mutation.withMutationId(MutationTrackingService.instance().nextMutationId(cfs.metadata().keyspace, mutation.key().getToken()))
.applyUnsafe();
}
diff --git a/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java b/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java
index b19cc1190e..cdb1f433e5 100644
--- a/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java
+++ b/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java
@@ -96,7 +96,7 @@ public class CoordinatorLogOffsetsLifecycleTest
private static Mutation createMutation(TableMetadata tableMetadata, int k, int v)
{
DecoratedKey key = tableMetadata.partitioner.decorateKey(ByteBufferUtil.bytes(k));
- MutationId mutationId = MutationTrackingService.instance.nextMutationId(tableMetadata.keyspace, key.getToken());
+ MutationId mutationId = MutationTrackingService.instance().nextMutationId(tableMetadata.keyspace, key.getToken());
SimpleBuilders.MutationBuilder builder = new SimpleBuilders.MutationBuilder(mutationId, tableMetadata.keyspace, key);
PartitionUpdate.SimpleBuilder partition = builder.update(tableMetadata);
partition.row().add("v", v);
@@ -185,7 +185,7 @@ public class CoordinatorLogOffsetsLifecycleTest
// flush 1
{
- MutationTrackingService.instance.persistLogStateForTesting();
+ MutationTrackingService.instance().persistLogStateForTesting();
cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
View view = cfs.getTracker().getView();
@@ -219,7 +219,7 @@ public class CoordinatorLogOffsetsLifecycleTest
// flush 2
{
- MutationTrackingService.instance.persistLogStateForTesting();
+ MutationTrackingService.instance().persistLogStateForTesting();
cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
View view = cfs.getTracker().getView();
diff --git a/test/unit/org/apache/cassandra/db/MutationTrackingMutationVerbHandlerMigrationTest.java b/test/unit/org/apache/cassandra/db/MutationTrackingMutationVerbHandlerMigrationTest.java
index 025e0e9031..47d9e81fb4 100644
--- a/test/unit/org/apache/cassandra/db/MutationTrackingMutationVerbHandlerMigrationTest.java
+++ b/test/unit/org/apache/cassandra/db/MutationTrackingMutationVerbHandlerMigrationTest.java
@@ -75,7 +75,7 @@ public class MutationTrackingMutationVerbHandlerMigrationTest
public static void init() throws Exception
{
ServerTestUtils.prepareServerNoRegister();
- MutationJournal.instance.start();
+ MutationJournal.start();
TableMetadata trackedTable = TableMetadata.builder(TRACKED_KEYSPACE, TABLE)
.addPartitionKeyColumn("pk", Int32Type.instance)
@@ -103,7 +103,6 @@ public class MutationTrackingMutationVerbHandlerMigrationTest
ServerTestUtils.markCMS();
StorageService.instance.unsafeSetInitialized();
- DatabaseDescriptor.setMutationTrackingEnabled(true);
}
@Before
diff --git a/test/unit/org/apache/cassandra/db/MutationVerbHandlerOutOfRangeTest.java b/test/unit/org/apache/cassandra/db/MutationVerbHandlerOutOfRangeTest.java
index e8ac3b209e..fa082f5c84 100644
--- a/test/unit/org/apache/cassandra/db/MutationVerbHandlerOutOfRangeTest.java
+++ b/test/unit/org/apache/cassandra/db/MutationVerbHandlerOutOfRangeTest.java
@@ -77,11 +77,10 @@ public class MutationVerbHandlerOutOfRangeTest
public static void init() throws Exception
{
ServerTestUtils.prepareServerNoRegister();
- MutationJournal.instance.start();
+ MutationJournal.start();
SchemaLoader.schemaDefinition(TEST_NAME);
ServerTestUtils.markCMS();
StorageService.instance.unsafeSetInitialized();
- org.apache.cassandra.config.DatabaseDescriptor.setMutationTrackingEnabled(true);
}
@Before
diff --git a/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerOutOfRangeTest.java b/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerOutOfRangeTest.java
index fda501a447..2e95db7d23 100644
--- a/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerOutOfRangeTest.java
+++ b/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerOutOfRangeTest.java
@@ -77,7 +77,7 @@ public class ReadCommandVerbHandlerOutOfRangeTest
public static void init() throws Throwable
{
ServerTestUtils.prepareServerNoRegister();
- MutationJournal.instance.start();
+ MutationJournal.start();
SchemaLoader.schemaDefinition(TEST_NAME);
ServerTestUtils.markCMS();
metadata_nonreplicated = Schema.instance.getTableMetadata(KEYSPACE_NONREPLICATED, TABLE);
diff --git a/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerTest.java b/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerTest.java
index d2f3e52957..4b6c9a2605 100644
--- a/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerTest.java
+++ b/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerTest.java
@@ -73,7 +73,7 @@ public class ReadCommandVerbHandlerTest
public static void init() throws Throwable
{
ServerTestUtils.prepareServerNoRegister();
- MutationJournal.instance.start();
+ MutationJournal.start();
SchemaLoader.schemaDefinition(TEST_NAME);
metadata = Schema.instance.getTableMetadata(KEYSPACE, TABLE);
metadata_with_transient = Schema.instance.getTableMetadata(KEYSPACE_WITH_TRANSIENT, TABLE);
diff --git a/test/unit/org/apache/cassandra/db/virtual/MutationJournalTableTest.java b/test/unit/org/apache/cassandra/db/virtual/MutationJournalTableTest.java
index 42f40fe7e7..e5578cf526 100644
--- a/test/unit/org/apache/cassandra/db/virtual/MutationJournalTableTest.java
+++ b/test/unit/org/apache/cassandra/db/virtual/MutationJournalTableTest.java
@@ -43,7 +43,7 @@ public class MutationJournalTableTest extends CQLTester
public static void setUpClass()
{
CQLTester.setUpClass();
- MutationJournalTable table = new MutationJournalTable(KS_NAME);
+ MutationTrackingTables.MutationJournalTable table = new MutationTrackingTables.MutationJournalTable(KS_NAME);
VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(KS_NAME, ImmutableList.of(table)));
}
@@ -57,7 +57,7 @@ public class MutationJournalTableTest extends CQLTester
public void testSelectAll() throws Throwable
{
// Start the mutation journal
- MutationJournal.instance.start();
+ MutationJournal.start();
// Write data to trigger journal writes
for (int i = 0; i < 100; i++)
diff --git a/test/unit/org/apache/cassandra/db/virtual/MutationTrackingShardsTableTest.java b/test/unit/org/apache/cassandra/db/virtual/MutationTrackingShardsTableTest.java
index 221daa2353..3e0eb1c4f7 100644
--- a/test/unit/org/apache/cassandra/db/virtual/MutationTrackingShardsTableTest.java
+++ b/test/unit/org/apache/cassandra/db/virtual/MutationTrackingShardsTableTest.java
@@ -48,7 +48,7 @@ public class MutationTrackingShardsTableTest extends CQLTester
public static void setUpClass()
{
CQLTester.setUpClass();
- MutationTrackingShardsTable table = new MutationTrackingShardsTable(KS_NAME);
+ MutationTrackingTables.MutationTrackingShardsTable table = new MutationTrackingTables.MutationTrackingShardsTable(KS_NAME);
VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(KS_NAME, ImmutableList.of(table)));
}
@@ -56,8 +56,8 @@ public class MutationTrackingShardsTableTest extends CQLTester
public void setUp()
{
// Start required services for mutation tracking
- MutationJournal.instance.start();
- MutationTrackingService.instance.start(ClusterMetadata.current());
+ MutationJournal.start();
+ MutationTrackingService.start(ClusterMetadata.current());
// Create a tracked keyspace
schemaChange("CREATE KEYSPACE IF NOT EXISTS tracked_ks WITH replication = " +
@@ -73,7 +73,7 @@ public class MutationTrackingShardsTableTest extends CQLTester
public void tearDown() throws InterruptedException
{
// Shutdown the service to prevent test hanging
- MutationTrackingService.instance.shutdownBlocking();
+ MutationTrackingService.shutdown();
}
@Test
diff --git a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java
index f95b131fe0..9105ad63b8 100644
--- a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java
+++ b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java
@@ -94,7 +94,7 @@ public class BootStrapperTest
oldPartitioner = StorageService.instance.setPartitionerUnsafe(Murmur3Partitioner.instance);
ServerTestUtils.prepareServerNoRegister();
SchemaLoader.startGossiper();
- MutationJournal.instance.start();
+ MutationJournal.start();
SchemaLoader.schemaDefinition("BootStrapperTest");
RangeStreamer.ALIVE_PREDICATE = Predicates.alwaysTrue();
ServerTestUtils.markCMS();
diff --git a/test/unit/org/apache/cassandra/locator/SimpleStrategyTest.java b/test/unit/org/apache/cassandra/locator/SimpleStrategyTest.java
index 393142041e..1befd06822 100644
--- a/test/unit/org/apache/cassandra/locator/SimpleStrategyTest.java
+++ b/test/unit/org/apache/cassandra/locator/SimpleStrategyTest.java
@@ -83,7 +83,6 @@ public class SimpleStrategyTest extends CassandraTestBase
@Before
public void defineSchema()
{
- DatabaseDescriptor.setMutationTrackingEnabled(true);
DatabaseDescriptor.setTransientReplicationEnabledUnsafe(true);
recreateCMS();
SchemaLoader.createKeyspace(KEYSPACE1, KeyspaceParams.simple(1));
diff --git a/test/unit/org/apache/cassandra/repair/RepairMessageVerbHandlerOutOfRangeTest.java b/test/unit/org/apache/cassandra/repair/RepairMessageVerbHandlerOutOfRangeTest.java
index e0e6b2ffcc..655bfd6b64 100644
--- a/test/unit/org/apache/cassandra/repair/RepairMessageVerbHandlerOutOfRangeTest.java
+++ b/test/unit/org/apache/cassandra/repair/RepairMessageVerbHandlerOutOfRangeTest.java
@@ -93,7 +93,7 @@ public class RepairMessageVerbHandlerOutOfRangeTest
SchemaLoader.loadSchema();
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
ServerTestUtils.recreateCMS();
- MutationJournal.instance.start();
+ MutationJournal.start();
SchemaLoader.schemaDefinition(TEST_NAME);
ClusterMetadataTestHelper.register(broadcastAddress);
ServerTestUtils.markCMS();
diff --git a/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java b/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java
index f2d6387bfb..026d5394fd 100644
--- a/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java
+++ b/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java
@@ -246,7 +246,7 @@ public class CoordinatorLogOffsetsTest
public void reconciledBounds() throws InterruptedException, ExecutionException {
DatabaseDescriptor.daemonInitialization();
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
- MutationJournal.instance.start();
+ MutationJournal.start();
String ks = "ks";
String tbl = "tbl";
@@ -274,7 +274,7 @@ public class CoordinatorLogOffsetsTest
ClusterMetadataTestHelper.commit(new AlterSchema(SchemaTransformations.addTable(tableMetadata, false)));
CommitLog.instance.start();
- MutationTrackingService.instance.start(metadata);
+ MutationTrackingService.start(metadata);
// Eventually, will also run perturbations before checking isReconciled (like log truncation, durability, etc.)
// to ensure that we don't prune data required to check what's been reconciled
@@ -282,45 +282,45 @@ public class CoordinatorLogOffsetsTest
// Applied at all replicas
{
Mutation mutation = MutationTrackingUtils.createMutation(tableMetadata, 1, 1);
- MutationTrackingService.instance.startWriting(mutation);
+ MutationTrackingService.instance().startWriting(mutation);
- MutationTrackingService.instance.finishWriting(mutation);
- MutationTrackingService.instance.receivedWriteResponse(mutation.id(), addr2);
- MutationTrackingService.instance.receivedWriteResponse(mutation.id(), addr3);
- MutationTrackingService.instance.persistLogStateForTesting();
+ MutationTrackingService.instance().finishWriting(mutation);
+ MutationTrackingService.instance().receivedWriteResponse(mutation.id(), addr2);
+ MutationTrackingService.instance().receivedWriteResponse(mutation.id(), addr3);
+ MutationTrackingService.instance().persistLogStateForTesting();
ImmutableCoordinatorLogOffsets logOffsets = new ImmutableCoordinatorLogOffsets.Builder()
.add(mutation.id())
.build();
Range<Token> range = getShardRange(mutation);
List<? extends Offsets> offsets = Collections.singletonList(logOffsets.mutations().offsets(mutation.id().logId()));
- MutationTrackingService.instance.updateReplicatedOffsets(ks, range, offsets, true, addr2);
- MutationTrackingService.instance.updateReplicatedOffsets(ks, range, offsets, true, addr3);
+ MutationTrackingService.instance().updateReplicatedOffsets(ks, range, offsets, true, addr2);
+ MutationTrackingService.instance().updateReplicatedOffsets(ks, range, offsets, true, addr3);
- Assertions.assertThat(MutationTrackingService.instance.isDurablyReconciled(logOffsets)).isTrue();
+ Assertions.assertThat(MutationTrackingService.instance().isDurablyReconciled(logOffsets)).isTrue();
}
// Applied locally but not on remote replicas
{
Mutation mutation = MutationTrackingUtils.createMutation(tableMetadata, 2, 2);
- MutationTrackingService.instance.startWriting(mutation);
- MutationTrackingService.instance.finishWriting(mutation);
- MutationTrackingService.instance.persistLogStateForTesting();
+ MutationTrackingService.instance().startWriting(mutation);
+ MutationTrackingService.instance().finishWriting(mutation);
+ MutationTrackingService.instance().persistLogStateForTesting();
ImmutableCoordinatorLogOffsets logOffsets = new ImmutableCoordinatorLogOffsets.Builder()
.add(mutation.id())
.build();
- Assertions.assertThat(MutationTrackingService.instance.isDurablyReconciled(logOffsets)).isFalse();
+ Assertions.assertThat(MutationTrackingService.instance().isDurablyReconciled(logOffsets)).isFalse();
}
// Applied on remote replicas but not locally
{
Mutation mutation = MutationTrackingUtils.createMutation(tableMetadata, 3, 3);
- MutationTrackingService.instance.startWriting(mutation);
+ MutationTrackingService.instance().startWriting(mutation);
- MutationTrackingService.instance.receivedWriteResponse(mutation.id(), addr2);
- MutationTrackingService.instance.receivedWriteResponse(mutation.id(), addr3);
- MutationTrackingService.instance.persistLogStateForTesting();
+ MutationTrackingService.instance().receivedWriteResponse(mutation.id(), addr2);
+ MutationTrackingService.instance().receivedWriteResponse(mutation.id(), addr3);
+ MutationTrackingService.instance().persistLogStateForTesting();
ImmutableCoordinatorLogOffsets logOffsets = new ImmutableCoordinatorLogOffsets.Builder()
.add(mutation.id())
@@ -328,21 +328,21 @@ public class CoordinatorLogOffsetsTest
Range<Token> range = getShardRange(mutation);
List<? extends Offsets> offsets = Collections.singletonList(logOffsets.mutations().offsets(mutation.id().logId()));
- MutationTrackingService.instance.updateReplicatedOffsets(ks, range, offsets, true, addr2);
- MutationTrackingService.instance.updateReplicatedOffsets(ks, range, offsets, true, addr3);
+ MutationTrackingService.instance().updateReplicatedOffsets(ks, range, offsets, true, addr2);
+ MutationTrackingService.instance().updateReplicatedOffsets(ks, range, offsets, true, addr3);
- Assertions.assertThat(MutationTrackingService.instance.isDurablyReconciled(logOffsets)).isFalse();
+ Assertions.assertThat(MutationTrackingService.instance().isDurablyReconciled(logOffsets)).isFalse();
}
// If no replicas are aware of a log, it should be considered unreconciled out of caution
{
Mutation mutation = MutationTrackingUtils.createMutation(tableMetadata, 4, 4);
- MutationTrackingService.instance.startWriting(mutation);
+ MutationTrackingService.instance().startWriting(mutation);
- MutationTrackingService.instance.finishWriting(mutation);
- MutationTrackingService.instance.receivedWriteResponse(mutation.id(), addr2);
- MutationTrackingService.instance.receivedWriteResponse(mutation.id(), addr3);
- MutationTrackingService.instance.persistLogStateForTesting();
+ MutationTrackingService.instance().finishWriting(mutation);
+ MutationTrackingService.instance().receivedWriteResponse(mutation.id(), addr2);
+ MutationTrackingService.instance().receivedWriteResponse(mutation.id(), addr3);
+ MutationTrackingService.instance().persistLogStateForTesting();
MutationId fakeMutationId = new MutationId(CoordinatorLogId.asLong(111, 222), MutationId.sequenceId(333, 444));
Assertions.assertThat(metadata.directory.version(new NodeId(fakeMutationId.hostId()))).isNull();
@@ -352,19 +352,19 @@ public class CoordinatorLogOffsetsTest
ImmutableCoordinatorLogOffsets.Builder logOffsetsBuilder = new ImmutableCoordinatorLogOffsets.Builder();
logOffsetsBuilder.add(fakeMutationId);
- Assertions.assertThatThrownBy(() -> MutationTrackingService.instance.isDurablyReconciled(logOffsetsBuilder.build()))
+ Assertions.assertThatThrownBy(() -> MutationTrackingService.instance().isDurablyReconciled(logOffsetsBuilder.build()))
.hasSameClassAs(new IllegalStateException())
.hasMessageMatching("Could not find shard for logId \\d+");
}
- MutationTrackingService.instance.shutdownBlocking();
+ MutationTrackingService.shutdown();
CommitLog.instance.stopUnsafe(true);
}
private Range<Token> getShardRange(Mutation mutation)
{
Map<String, Range<Token>> ksRanges = new HashMap<>();
- MutationTrackingService.instance.forEachKeyspace(shards -> {
+ MutationTrackingService.instance().forEachKeyspace(shards -> {
Shard shard = shards.lookUp(mutation);
ksRanges.put(shard.keyspace, shard.range);
});
diff --git a/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java b/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java
index 7da22a6c1f..d58faa8831 100644
--- a/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java
+++ b/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java
@@ -68,7 +68,7 @@ public class CoordinatorLogTest
.addClusteringColumn("ck", UTF8Type.instance)
.addRegularColumn("value", UTF8Type.instance)
.build());
- MutationJournal.instance.start();
+ MutationJournal.start();
}
private static Token tk(String key)
@@ -176,8 +176,8 @@ public class CoordinatorLogTest
Mutation mutation2 = createMutation(new MutationId(logId.asLong(), MutationId.sequenceId(2, 0)));
unreconciled.addDirectly(mutation1);
unreconciled.addDirectly(mutation2);
- MutationJournal.instance.write(mutation1.id(), mutation1);
- MutationJournal.instance.write(mutation2.id(), mutation2);
+ MutationJournal.instance().write(mutation1.id(), mutation1);
+ MutationJournal.instance().write(mutation2.id(), mutation2);
CoordinatorLog log =
CoordinatorLog.recreate(KEYSPACE, range, LOCAL_HOST_ID, logId, PARTICIPANTS, witnessed, witnessed, unreconciled);
diff --git a/test/unit/org/apache/cassandra/replication/MutationJournalTest.java b/test/unit/org/apache/cassandra/replication/MutationJournalTest.java
index 72017bc8a1..60c4e39f53 100644
--- a/test/unit/org/apache/cassandra/replication/MutationJournalTest.java
+++ b/test/unit/org/apache/cassandra/replication/MutationJournalTest.java
@@ -74,7 +74,7 @@ public class MutationJournalTest
directory.deleteRecursiveOnExit();
journal = new MutationJournal(directory, TestParams.MUTATION_JOURNAL);
- journal.start();
+ journal.startInternal();
}
@AfterClass
diff --git a/test/unit/org/apache/cassandra/replication/ShardTest.java b/test/unit/org/apache/cassandra/replication/ShardTest.java
index d8c62554f1..37b1ea8193 100644
--- a/test/unit/org/apache/cassandra/replication/ShardTest.java
+++ b/test/unit/org/apache/cassandra/replication/ShardTest.java
@@ -56,7 +56,7 @@ public class ShardTest
.addClusteringColumn("ck", UTF8Type.instance)
.addRegularColumn("value", UTF8Type.instance)
.build());
- MutationJournal.instance.start();
+ MutationJournal.start();
}
private static Token tk(String key)
diff --git a/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java b/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java
index 281c8da079..d283eadede 100644
--- a/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java
+++ b/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java
@@ -422,7 +422,7 @@ public class UnreconciledMutationsTest
@Test
public void testLoadFromJournal()
{
- MutationJournal.instance.start();
+ MutationJournal.start();
CoordinatorLogId logId = new CoordinatorLogId(1, 1);
@@ -438,8 +438,8 @@ public class UnreconciledMutationsTest
Mutation mutation6 = createMutation(6, 6, 6);
Mutation mutation7 = createMutation(7, 7, 7);
- MutationJournal.instance.write(mutation6.id(), mutation6);
- MutationJournal.instance.write(mutation7.id(), mutation7);
+ MutationJournal.instance().write(mutation6.id(), mutation6);
+ MutationJournal.instance().write(mutation7.id(), mutation7);
Offsets.Mutable loadedOffsets = new Offsets.Mutable(logId);
UnreconciledMutations unreconciled = UnreconciledMutations.loadFromJournal(witnessed, 1);
diff --git a/test/unit/org/apache/cassandra/service/BootstrapTransientTest.java b/test/unit/org/apache/cassandra/service/BootstrapTransientTest.java
index b61a394057..d7aa637f93 100644
--- a/test/unit/org/apache/cassandra/service/BootstrapTransientTest.java
+++ b/test/unit/org/apache/cassandra/service/BootstrapTransientTest.java
@@ -96,7 +96,6 @@ public class BootstrapTransientTest extends CassandraTestBase
{
DatabaseDescriptor.daemonInitialization();
DatabaseDescriptor.setPartitionerUnsafe(OrderPreservingPartitioner.instance);
- DatabaseDescriptor.setMutationTrackingEnabled(true);
DatabaseDescriptor.setTransientReplicationEnabledUnsafe(true);
address02 = InetAddressAndPort.getByName("127.0.0.2");
address03 = InetAddressAndPort.getByName("127.0.0.3");
diff --git a/test/unit/org/apache/cassandra/service/JoinTokenRingTest.java b/test/unit/org/apache/cassandra/service/JoinTokenRingTest.java
index 55b5e8dce8..92643d3a68 100644
--- a/test/unit/org/apache/cassandra/service/JoinTokenRingTest.java
+++ b/test/unit/org/apache/cassandra/service/JoinTokenRingTest.java
@@ -46,7 +46,7 @@ public class JoinTokenRingTest
DatabaseDescriptor.daemonInitialization();
ServerTestUtils.prepareServerNoRegister();
SchemaLoader.startGossiper();
- MutationJournal.instance.start();
+ MutationJournal.start();
SchemaLoader.schemaDefinition("JoinTokenRingTest");
}
diff --git a/test/unit/org/apache/cassandra/service/PaxosStateTest.java b/test/unit/org/apache/cassandra/service/PaxosStateTest.java
index 4d25c18530..1d3825d882 100644
--- a/test/unit/org/apache/cassandra/service/PaxosStateTest.java
+++ b/test/unit/org/apache/cassandra/service/PaxosStateTest.java
@@ -69,7 +69,7 @@ public class PaxosStateTest
public static void setUpClass() throws Throwable
{
SchemaLoader.loadSchema();
- MutationJournal.instance.start();
+ MutationJournal.start();
SchemaLoader.schemaDefinition("PaxosStateTest");
}
diff --git a/test/unit/org/apache/cassandra/service/StorageServiceTest.java b/test/unit/org/apache/cassandra/service/StorageServiceTest.java
index 4616df8b5b..deb31d729c 100644
--- a/test/unit/org/apache/cassandra/service/StorageServiceTest.java
+++ b/test/unit/org/apache/cassandra/service/StorageServiceTest.java
@@ -75,7 +75,7 @@ public class StorageServiceTest extends TestBaseImpl
SimpleLocationProvider.LOCATION,
NodeVersion.CURRENT));
CommitLog.instance.start();
- MutationJournal.instance.start();
+ MutationJournal.start();
}
@Before
diff --git a/test/unit/org/apache/cassandra/service/WriteResponseHandlerTransientTest.java b/test/unit/org/apache/cassandra/service/WriteResponseHandlerTransientTest.java
index 37358d4328..aac5e8b86b 100644
--- a/test/unit/org/apache/cassandra/service/WriteResponseHandlerTransientTest.java
+++ b/test/unit/org/apache/cassandra/service/WriteResponseHandlerTransientTest.java
@@ -90,8 +90,7 @@ public class WriteResponseHandlerTransientTest
public static void setupClass() throws Throwable
{
SchemaLoader.loadSchema();
- MutationJournal.instance.start();
- DatabaseDescriptor.setMutationTrackingEnabled(true);
+ MutationJournal.start();
DatabaseDescriptor.setTransientReplicationEnabledUnsafe(true);
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
diff --git a/test/unit/org/apache/cassandra/service/paxos/PaxosStateTest.java b/test/unit/org/apache/cassandra/service/paxos/PaxosStateTest.java
index e080e4fc11..c8092d739e 100644
--- a/test/unit/org/apache/cassandra/service/paxos/PaxosStateTest.java
+++ b/test/unit/org/apache/cassandra/service/paxos/PaxosStateTest.java
@@ -64,7 +64,7 @@ public class PaxosStateTest
public static void setUpClass() throws Throwable
{
SchemaLoader.loadSchema();
- MutationJournal.instance.start();
+ MutationJournal.start();
SchemaLoader.schemaDefinition("PaxosStateTest");
metadata = Keyspace.open("PaxosStateTestKeyspace1").getColumnFamilyStore("Standard1").metadata.get();
metadata.withSwapped(metadata.params.unbuild().gcGraceSeconds(3600).build());
diff --git a/test/unit/org/apache/cassandra/service/paxos/PaxosVerbHandlerOutOfRangeTest.java b/test/unit/org/apache/cassandra/service/paxos/PaxosVerbHandlerOutOfRangeTest.java
index c8cd6b5b06..9f6191d074 100644
--- a/test/unit/org/apache/cassandra/service/paxos/PaxosVerbHandlerOutOfRangeTest.java
+++ b/test/unit/org/apache/cassandra/service/paxos/PaxosVerbHandlerOutOfRangeTest.java
@@ -76,7 +76,7 @@ public class PaxosVerbHandlerOutOfRangeTest // PaxosV1 out of range tests - V2 i
public static void init() throws Exception
{
ServerTestUtils.prepareServerNoRegister();
- MutationJournal.instance.start();
+ MutationJournal.start();
SchemaLoader.schemaDefinition(TEST_NAME);
ServerTestUtils.markCMS();
StorageService.instance.unsafeSetInitialized();
diff --git a/test/unit/org/apache/cassandra/service/reads/repair/RepairedDataVerifierTest.java b/test/unit/org/apache/cassandra/service/reads/repair/RepairedDataVerifierTest.java
index 4e04b3e659..d091c80aa8 100644
--- a/test/unit/org/apache/cassandra/service/reads/repair/RepairedDataVerifierTest.java
+++ b/test/unit/org/apache/cassandra/service/reads/repair/RepairedDataVerifierTest.java
@@ -62,7 +62,7 @@ public class RepairedDataVerifierTest
public static void init()
{
SchemaLoader.loadSchema();
- MutationJournal.instance.start();
+ MutationJournal.start();
SchemaLoader.schemaDefinition(TEST_NAME);
DatabaseDescriptor.reportUnconfirmedRepairedDataMismatches(true);
}
diff --git a/test/unit/org/apache/cassandra/streaming/StreamReaderTest.java b/test/unit/org/apache/cassandra/streaming/StreamReaderTest.java
index a448c81900..45dcd381e1 100644
--- a/test/unit/org/apache/cassandra/streaming/StreamReaderTest.java
+++ b/test/unit/org/apache/cassandra/streaming/StreamReaderTest.java
@@ -99,7 +99,7 @@ public class StreamReaderTest
DatabaseDescriptor.daemonInitialization();
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
ServerTestUtils.prepareServerNoRegister();
- MutationJournal.instance.start();
+ MutationJournal.start();
SchemaLoader.schemaDefinition(TEST_NAME);
ServerTestUtils.markCMS();
}
diff --git a/test/unit/org/apache/cassandra/streaming/StreamSessionOwnedRangesTest.java b/test/unit/org/apache/cassandra/streaming/StreamSessionOwnedRangesTest.java
index 79ff929384..b5d22a1645 100644
--- a/test/unit/org/apache/cassandra/streaming/StreamSessionOwnedRangesTest.java
+++ b/test/unit/org/apache/cassandra/streaming/StreamSessionOwnedRangesTest.java
@@ -78,7 +78,7 @@ public class StreamSessionOwnedRangesTest
SchemaLoader.loadSchema();
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
ServerTestUtils.recreateCMS();
- MutationJournal.instance.start();
+ MutationJournal.start();
SchemaLoader.schemaDefinition(TEST_NAME);
ClusterMetadataTestHelper.register(broadcastAddress);
ServerTestUtils.markCMS();
diff --git a/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java b/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java
index 12bd501a86..a82b516d77 100644
--- a/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java
+++ b/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java
@@ -59,7 +59,7 @@ public class AlterSchemaMutationTrackingTest
CassandraRelevantProperties.PARTITIONER.setString(Murmur3Partitioner.class.getName());
ServerTestUtils.daemonInitialization();
ServerTestUtils.prepareServer();
- MutationJournal.instance.start();
+ MutationJournal.start();
partitioner = (Murmur3Partitioner) DatabaseDescriptor.getPartitioner();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org