Upgrade to 4.0 fails with NullPointerException

Patch by Ariel Weisberg; Reviewed by Dinesh Joshi for CASSANDRA-14820

doc/source/cql/ddl.rst

@@ -463,7 +463,7 @@ A table supports the following options:
 | ``speculative_retry``          | *simple* | 99PERCENTILE| :ref:`Speculative retry options                           |
 |                                |          |             | <speculative-retry-options>`.                             |
 +--------------------------------+----------+-------------+-----------------------------------------------------------+
-| ``speculative_write_threshold``| *simple* | 99PERCENTILE| :ref:`Speculative retry options                           |
+| ``additional_write_policy``    | *simple* | 99PERCENTILE| :ref:`Speculative retry options                           |
 |                                |          |             | <speculative-retry-options>`.                             |
 +--------------------------------+----------+-------------+-----------------------------------------------------------+
 | ``gc_grace_seconds``           | *simple* | 864000      | Time to wait before garbage collecting tombstones         |
@@ -496,7 +496,7 @@ Speculative retry options
 By default, Cassandra read coordinators only query as many replicas as necessary to satisfy
 consistency levels: one for consistency level ``ONE``, a quorum for ``QUORUM``, and so on.
 ``speculative_retry`` determines when coordinators may query additional replicas, which is useful
-when replicas are slow or unresponsive.  ``speculative_write_threshold`` specifies the threshold at which
+when replicas are slow or unresponsive.  ``additional_write_policy`` specifies the threshold at which
 a cheap quorum write will be upgraded to include transient replicas.  The following are legal values (case-insensitive):
 
 ============================ ======================== =============================================================================

pylib/cqlshlib/cql3handling.py

@@ -49,7 +49,7 @@ class Cql3ParsingRuleSet(CqlParsingRuleSet):
         ('max_index_interval', None),
         ('default_time_to_live', None),
         ('speculative_retry', None),
-        ('speculative_write_threshold', None),
+        ('additional_write_policy', None),
         ('memtable_flush_period_in_ms', None),
         ('cdc', None),
         ('read_repair', None),

pylib/cqlshlib/test/test_cqlsh_completion.py

@@ -594,7 +594,7 @@ class TestCqlshCompletion(CqlshCompletionCase):
                                      'memtable_flush_period_in_ms',
                                      'CLUSTERING',
                                      'COMPACT', 'caching', 'comment',
-                                     'min_index_interval', 'speculative_retry', 'speculative_write_threshold', 'cdc'])
+                                     'min_index_interval', 'speculative_retry', 'additional_write_policy', 'cdc'])
         self.trycompletions(prefix + ' new_table (col_a int PRIMARY KEY) WITH ',
                             choices=['bloom_filter_fp_chance', 'compaction',
                                      'compression',
@@ -603,7 +603,7 @@ class TestCqlshCompletion(CqlshCompletionCase):
                                      'memtable_flush_period_in_ms',
                                      'CLUSTERING',
                                      'COMPACT', 'caching', 'comment',
-                                     'min_index_interval', 'speculative_retry', 'speculative_write_threshold', 'cdc'])
+                                     'min_index_interval', 'speculative_retry', 'additional_write_policy', 'cdc'])
         self.trycompletions(prefix + ' new_table (col_a int PRIMARY KEY) WITH bloom_filter_fp_chance ',
                             immediate='= ')
         self.trycompletions(prefix + ' new_table (col_a int PRIMARY KEY) WITH bloom_filter_fp_chance = ',
@@ -650,7 +650,7 @@ class TestCqlshCompletion(CqlshCompletionCase):
                                      'memtable_flush_period_in_ms',
                                      'CLUSTERING',
                                      'COMPACT', 'caching', 'comment',
-                                     'min_index_interval', 'speculative_retry', 'speculative_write_threshold', 'cdc'])
+                                     'min_index_interval', 'speculative_retry', 'additional_write_policy', 'cdc'])
         self.trycompletions(prefix + " new_table (col_a int PRIMARY KEY) WITH compaction = "
                             + "{'class': 'DateTieredCompactionStrategy', '",
                             choices=['base_time_seconds', 'max_sstable_age_days',

pylib/cqlshlib/test/test_cqlsh_output.py

@@ -623,7 +623,7 @@ class TestCqlshOutput(BaseTestCase):
                 AND memtable_flush_period_in_ms = 0
                 AND min_index_interval = 128
                 AND speculative_retry = '99PERCENTILE'
-                AND speculative_write_threshold = '99PERCENTILE';
+                AND additional_write_policy = '99PERCENTILE';
 
         """ % quote_name(get_keyspace()))
 

src/java/org/apache/cassandra/cql3/statements/schema/TableAttributes.java

@@ -128,8 +128,8 @@ public final class TableAttributes extends PropertyDefinitions
         if (hasOption(Option.SPECULATIVE_RETRY))
             builder.speculativeRetry(SpeculativeRetryPolicy.fromString(getString(Option.SPECULATIVE_RETRY)));
 
-        if (hasOption(Option.SPECULATIVE_WRITE_THRESHOLD))
+        if (hasOption(Option.ADDITIONAL_WRITE_POLICY))
-            builder.speculativeWriteThreshold(SpeculativeRetryPolicy.fromString(getString(Option.SPECULATIVE_WRITE_THRESHOLD)));
+            builder.additionalWritePolicy(SpeculativeRetryPolicy.fromString(getString(Option.ADDITIONAL_WRITE_POLICY)));
 
         if (hasOption(Option.CRC_CHECK_CHANCE))
             builder.crcCheckChance(getDouble(Option.CRC_CHECK_CHANCE));

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -204,7 +204,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
     public final TableMetrics metric;
     public volatile long sampleReadLatencyNanos;
-    public volatile long transientWriteLatencyNanos;
+    public volatile long additionalWriteLatencyNanos;
 
     private final CassandraTableWriteHandler writeHandler;
     private final CassandraStreamManager streamManager;
@@ -384,7 +384,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         metric = new TableMetrics(this);
         fileIndexGenerator.set(generation);
         sampleReadLatencyNanos = TimeUnit.MILLISECONDS.toNanos(DatabaseDescriptor.getReadRpcTimeout() / 2);
-        transientWriteLatencyNanos = TimeUnit.MILLISECONDS.toNanos(DatabaseDescriptor.getWriteRpcTimeout() / 2);
+        additionalWriteLatencyNanos = TimeUnit.MILLISECONDS.toNanos(DatabaseDescriptor.getWriteRpcTimeout() / 2);
 
         logger.info("Initializing {}.{}", keyspace.getName(), name);
 
@@ -455,7 +455,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         try
         {
             sampleReadLatencyNanos = metadata().params.speculativeRetry.calculateThreshold(metric.coordinatorReadLatency.getSnapshot(), sampleReadLatencyNanos);
-            transientWriteLatencyNanos = metadata().params.speculativeWriteThreshold.calculateThreshold(metric.coordinatorWriteLatency.getSnapshot(), transientWriteLatencyNanos);
+            additionalWriteLatencyNanos = metadata().params.additionalWritePolicy.calculateThreshold(metric.coordinatorWriteLatency.getSnapshot(), additionalWriteLatencyNanos);
         }
         catch (Throwable e)
         {

src/java/org/apache/cassandra/db/TableCQLHelper.java

@@ -310,7 +310,7 @@ public class TableCQLHelper
         builder.append("\n\tAND max_index_interval = ").append(tableParams.maxIndexInterval);
         builder.append("\n\tAND memtable_flush_period_in_ms = ").append(tableParams.memtableFlushPeriodInMs);
         builder.append("\n\tAND speculative_retry = '").append(tableParams.speculativeRetry).append("'");
-        builder.append("\n\tAND speculative_write_threshold = '").append(tableParams.speculativeWriteThreshold).append("'");
+        builder.append("\n\tAND additional_write_policy = '").append(tableParams.additionalWritePolicy).append("'");
         builder.append("\n\tAND comment = ").append(singleQuote(tableParams.comment));
         builder.append("\n\tAND caching = ").append(toCQL(tableParams.caching.asMap()));
         builder.append("\n\tAND compaction = ").append(toCQL(tableParams.compaction.asMap()));

src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java

@@ -104,7 +104,7 @@ public class KeyspaceMetrics
     /** Needed to speculate, but didn't have enough replicas **/
     public final Counter speculativeInsufficientReplicas;
     /** Needed to write to a transient replica to satisfy quorum **/
-    public final Counter speculativeWrites;
+    public final Counter additionalWrites;
     /** Number of started repairs as coordinator on this keyspace */
     public final Counter repairsStarted;
     /** Number of completed repairs as coordinator on this keyspace */
@@ -292,7 +292,7 @@ public class KeyspaceMetrics
         speculativeRetries = createKeyspaceCounter("SpeculativeRetries", metric -> metric.speculativeRetries.getCount());
         speculativeFailedRetries = createKeyspaceCounter("SpeculativeFailedRetries", metric -> metric.speculativeFailedRetries.getCount());
         speculativeInsufficientReplicas = createKeyspaceCounter("SpeculativeInsufficientReplicas", metric -> metric.speculativeInsufficientReplicas.getCount());
-        speculativeWrites = createKeyspaceCounter("SpeculativeWrites", metric -> metric.speculativeWrites.getCount());
+        additionalWrites = createKeyspaceCounter("AdditionalWrites", metric -> metric.additionalWrites.getCount());
         repairsStarted = createKeyspaceCounter("RepairJobsStarted", metric -> metric.repairsStarted.getCount());
         repairsCompleted = createKeyspaceCounter("RepairJobsCompleted", metric -> metric.repairsCompleted.getCount());
         repairTime = Metrics.timer(factory.createMetricName("RepairTime"));

src/java/org/apache/cassandra/metrics/TableMetrics.java

@@ -214,8 +214,8 @@ public class TableMetrics
     public final Counter speculativeInsufficientReplicas;
     public final Gauge<Long> speculativeSampleLatencyNanos;
 
-    public final Counter speculativeWrites;
+    public final Counter additionalWrites;
-    public final Gauge<Long> speculativeWriteLatencyNanos;
+    public final Gauge<Long> additionalWriteLatencyNanos;
 
     /**
      * Metrics for inconsistencies detected between repaired data sets across replicas. These
@@ -843,8 +843,8 @@ public class TableMetrics
         speculativeInsufficientReplicas = createTableCounter("SpeculativeInsufficientReplicas");
         speculativeSampleLatencyNanos = createTableGauge("SpeculativeSampleLatencyNanos", () -> cfs.sampleReadLatencyNanos);
 
-        speculativeWrites = createTableCounter("SpeculativeWrites");
+        additionalWrites = createTableCounter("AdditionalWrites");
-        speculativeWriteLatencyNanos = createTableGauge("SpeculativeWriteLatencyNanos", () -> cfs.transientWriteLatencyNanos);
+        additionalWriteLatencyNanos = createTableGauge("AdditionalWriteLatencyNanos", () -> cfs.additionalWriteLatencyNanos);
 
         keyCacheHitRate = Metrics.register(factory.createMetricName("KeyCacheHitRate"),
                                            aliasFactory.createMetricName("KeyCacheHitRate"),

src/java/org/apache/cassandra/schema/SchemaKeyspace.java

@@ -137,7 +137,7 @@ public final class SchemaKeyspace
               + "min_index_interval int,"
               + "read_repair_chance double," // no longer used, left for drivers' sake
               + "speculative_retry text,"
-              + "speculative_write_threshold text,"
+              + "additional_write_policy text,"
               + "cdc boolean,"
               + "read_repair text,"
               + "PRIMARY KEY ((keyspace_name), table_name))");
@@ -204,7 +204,7 @@ public final class SchemaKeyspace
               + "min_index_interval int,"
               + "read_repair_chance double," // no longer used, left for drivers' sake
               + "speculative_retry text,"
-              + "speculative_write_threshold text,"
+              + "additional_write_policy text,"
               + "cdc boolean,"
               + "read_repair text,"
               + "PRIMARY KEY ((keyspace_name), view_name))");
@@ -565,7 +565,7 @@ public final class SchemaKeyspace
                .add("min_index_interval", params.minIndexInterval)
                .add("read_repair_chance", 0.0) // no longer used, left for drivers' sake
                .add("speculative_retry", params.speculativeRetry.toString())
-               .add("speculative_write_threshold", params.speculativeWriteThreshold.toString())
+               .add("additional_write_policy", params.additionalWritePolicy.toString())
                .add("crc_check_chance", params.crcCheckChance)
                .add("caching", params.caching.asMap())
                .add("compaction", params.compaction.asMap())
@@ -994,7 +994,9 @@ public final class SchemaKeyspace
                           .minIndexInterval(row.getInt("min_index_interval"))
                           .crcCheckChance(row.getDouble("crc_check_chance"))
                           .speculativeRetry(SpeculativeRetryPolicy.fromString(row.getString("speculative_retry")))
-                          .speculativeWriteThreshold(SpeculativeRetryPolicy.fromString(row.getString("speculative_write_threshold")))
+                          .additionalWritePolicy(row.has("additional_write_policy") ?
+                                                     SpeculativeRetryPolicy.fromString(row.getString("additional_write_policy")) :
+                                                     SpeculativeRetryPolicy.fromString("99PERCENTILE"))
                           .cdc(row.has("cdc") && row.getBoolean("cdc"))
                           .readRepair(getReadRepairStrategy(row))
                           .build();

src/java/org/apache/cassandra/schema/TableMetadata.java

@@ -834,9 +834,9 @@ public final class TableMetadata
             return this;
         }
 
-        public Builder speculativeWriteThreshold(SpeculativeRetryPolicy val)
+        public Builder additionalWritePolicy(SpeculativeRetryPolicy val)
         {
-            params.speculativeWriteThreshold(val);
+            params.additionalWritePolicy(val);
             return this;
         }
 

src/java/org/apache/cassandra/schema/TableParams.java

@@ -51,7 +51,7 @@ public final class TableParams
         MEMTABLE_FLUSH_PERIOD_IN_MS,
         MIN_INDEX_INTERVAL,
         SPECULATIVE_RETRY,
-        SPECULATIVE_WRITE_THRESHOLD,
+        ADDITIONAL_WRITE_POLICY,
         CRC_CHECK_CHANCE,
         CDC,
         READ_REPAIR;
@@ -72,7 +72,7 @@ public final class TableParams
     public final int minIndexInterval;
     public final int maxIndexInterval;
     public final SpeculativeRetryPolicy speculativeRetry;
-    public final SpeculativeRetryPolicy speculativeWriteThreshold;
+    public final SpeculativeRetryPolicy additionalWritePolicy;
     public final CachingParams caching;
     public final CompactionParams compaction;
     public final CompressionParams compression;
@@ -93,7 +93,7 @@ public final class TableParams
         minIndexInterval = builder.minIndexInterval;
         maxIndexInterval = builder.maxIndexInterval;
         speculativeRetry = builder.speculativeRetry;
-        speculativeWriteThreshold = builder.speculativeWriteThreshold;
+        additionalWritePolicy = builder.additionalWritePolicy;
         caching = builder.caching;
         compaction = builder.compaction;
         compression = builder.compression;
@@ -121,7 +121,7 @@ public final class TableParams
                             .memtableFlushPeriodInMs(params.memtableFlushPeriodInMs)
                             .minIndexInterval(params.minIndexInterval)
                             .speculativeRetry(params.speculativeRetry)
-                            .speculativeWriteThreshold(params.speculativeWriteThreshold)
+                            .additionalWritePolicy(params.additionalWritePolicy)
                             .extensions(params.extensions)
                             .cdc(params.cdc)
                             .readRepair(params.readRepair);
@@ -264,7 +264,7 @@ public final class TableParams
         private int minIndexInterval = 128;
         private int maxIndexInterval = 2048;
         private SpeculativeRetryPolicy speculativeRetry = PercentileSpeculativeRetryPolicy.NINETY_NINE_P;
-        private SpeculativeRetryPolicy speculativeWriteThreshold = PercentileSpeculativeRetryPolicy.NINETY_NINE_P;
+        private SpeculativeRetryPolicy additionalWritePolicy = PercentileSpeculativeRetryPolicy.NINETY_NINE_P;
         private CachingParams caching = CachingParams.DEFAULT;
         private CompactionParams compaction = CompactionParams.DEFAULT;
         private CompressionParams compression = CompressionParams.DEFAULT;
@@ -335,9 +335,9 @@ public final class TableParams
             return this;
         }
 
-        public Builder speculativeWriteThreshold(SpeculativeRetryPolicy val)
+        public Builder additionalWritePolicy(SpeculativeRetryPolicy val)
         {
-            speculativeWriteThreshold = val;
+            additionalWritePolicy = val;
             return this;
         }
 

src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java

@@ -298,7 +298,7 @@ public abstract class AbstractWriteResponseHandler<T> implements IAsyncCallbackW
                                               .map(Schema.instance::getColumnFamilyStoreInstance)
                                               .collect(Collectors.toList());
         for (ColumnFamilyStore cf : cfs)
-            timeout = Math.min(timeout, cf.transientWriteLatencyNanos);
+            timeout = Math.min(timeout, cf.additionalWriteLatencyNanos);
 
         // no latency information, or we're overloaded
         if (timeout > TimeUnit.MILLISECONDS.toNanos(mutation.getTimeout()))
@@ -309,7 +309,7 @@ public abstract class AbstractWriteResponseHandler<T> implements IAsyncCallbackW
             if (!condition.await(timeout, TimeUnit.NANOSECONDS))
             {
                 for (ColumnFamilyStore cf : cfs)
-                    cf.metric.speculativeWrites.inc();
+                    cf.metric.additionalWrites.inc();
 
                 writePerformer.apply(mutation, replicaPlan.withContact(uncontacted),
                                      (AbstractWriteResponseHandler<IMutation>) this,

src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java

@@ -74,7 +74,7 @@ public class BlockingReadRepair<E extends Endpoints<E>, P extends ReplicaPlan.Fo
     {
         for (BlockingPartitionRepair repair: repairs)
         {
-            repair.maybeSendAdditionalWrites(cfs.transientWriteLatencyNanos, TimeUnit.NANOSECONDS);
+            repair.maybeSendAdditionalWrites(cfs.additionalWriteLatencyNanos, TimeUnit.NANOSECONDS);
         }
     }
 

test/unit/org/apache/cassandra/db/TableCQLHelperTest.java

@@ -256,7 +256,7 @@ public class TableCQLHelperTest extends CQLTester
                .maxIndexInterval(7)
                .memtableFlushPeriod(8)
                .speculativeRetry(AlwaysSpeculativeRetryPolicy.INSTANCE)
-               .speculativeWriteThreshold(NeverSpeculativeRetryPolicy.INSTANCE)
+               .additionalWritePolicy(NeverSpeculativeRetryPolicy.INSTANCE)
                .extensions(ImmutableMap.of("ext1", ByteBuffer.wrap("val1".getBytes())))
                .recordColumnDrop(ColumnMetadata.regularColumn(keyspace, table, "reg1", AsciiType.instance),
                                  FBUtilities.timestampMicros());
@@ -274,7 +274,7 @@ public class TableCQLHelperTest extends CQLTester
         "\tAND max_index_interval = 7\n" +
         "\tAND memtable_flush_period_in_ms = 8\n" +
         "\tAND speculative_retry = 'ALWAYS'\n" +
-        "\tAND speculative_write_threshold = 'NEVER'\n" +
+        "\tAND additional_write_policy = 'NEVER'\n" +
         "\tAND comment = 'comment'\n" +
         "\tAND caching = { 'keys': 'ALL', 'rows_per_partition': 'NONE' }\n" +
         "\tAND compaction = { 'max_threshold': '32', 'min_threshold': '4', 'sstable_size_in_mb': '1', 'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy' }\n" +

test/unit/org/apache/cassandra/service/reads/repair/AbstractReadRepairTest.java

@@ -205,7 +205,7 @@ public abstract  class AbstractReadRepairTest
         cfs = ks.getColumnFamilyStore("tbl");
 
         cfs.sampleReadLatencyNanos = 0;
-        cfs.transientWriteLatencyNanos = 0;
+        cfs.additionalWriteLatencyNanos = 0;
 
         target1 = InetAddressAndPort.getByName("127.0.0.255");
         target2 = InetAddressAndPort.getByName("127.0.0.254");
@@ -243,7 +243,7 @@ public abstract  class AbstractReadRepairTest
         assert configured : "configureClass must be called in a @BeforeClass method";
 
         cfs.sampleReadLatencyNanos = 0;
-        cfs.transientWriteLatencyNanos = 0;
+        cfs.additionalWriteLatencyNanos = 0;
     }
 
     static ReplicaPlan.ForRangeRead replicaPlan(ConsistencyLevel consistencyLevel, EndpointsForRange replicas)