Skip to content

Commit fd32cd6

Browse files
szymon-miezalszymon-miezal
committed
STAR-1903: ConsistencyLevel.THREE means ALL_BUT_ONE
Add a system property (dse.consistency_level.three_means_all_but_one) that changes the behavior of ConsistencyLevel.THREE to mean ALL_BUT_ONE, i.e. that all replicas except for at most one in the cluster (across all DCs) must accept the write for it to be successful.
1 parent fad831e commit fd32cd6

3 files changed

Lines changed: 209 additions & 1 deletion

File tree

‎src/java/org/apache/cassandra/config/CassandraRelevantProperties.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -387,7 +387,11 @@ public enum CassandraRelevantProperties
387387
CUSTOM_READ_OBSERVER_FACTORY("cassandra.custom_read_observer_factory_class"),
388388

389389
// Allows skipping advising the OS to free cached pages associated with commitlog flushing
390-
COMMITLOG_SKIP_FILE_ADVICE("cassandra.commitlog.skip_file_advice");
390+
COMMITLOG_SKIP_FILE_ADVICE("cassandra.commitlog.skip_file_advice"),
391+
392+
// Changes the semantic of the "THREE" consistency level to mean "all but one"
393+
// i.e. that all replicas except for at most one in the cluster (across all DCs) must accept the write for it to be successful.
394+
THREE_MEANS_ALL_BUT_ONE("dse.consistency_level.three_means_all_but_one", "false");
391395

392396
CassandraRelevantProperties(String key, String defaultVal)
393397
{

‎src/java/org/apache/cassandra/db/ConsistencyLevel.java‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import javax.annotation.Nullable;
2323

2424
import com.carrotsearch.hppc.ObjectIntHashMap;
25+
import org.apache.cassandra.config.CassandraRelevantProperties;
2526
import org.apache.cassandra.guardrails.Guardrails;
2627
import org.apache.cassandra.locator.Endpoints;
2728
import org.apache.cassandra.schema.TableMetadata;
@@ -50,6 +51,8 @@ public enum ConsistencyLevel
5051
LOCAL_ONE (10, true),
5152
NODE_LOCAL (11, true);
5253

54+
public static final boolean THREE_MEANS_ALL_BUT_ONE = CassandraRelevantProperties.THREE_MEANS_ALL_BUT_ONE.getBoolean();
55+
5356
// Used by the binary protocol
5457
public final int code;
5558
private final boolean isDCLocal;
@@ -79,6 +82,16 @@ private ConsistencyLevel(int code, boolean isDCLocal)
7982
this.isDCLocal = isDCLocal;
8083
}
8184

85+
@Override
86+
public String toString()
87+
{
88+
if (this == THREE && THREE_MEANS_ALL_BUT_ONE)
89+
{
90+
return "THREE (ALL_BUT_ONE)";
91+
}
92+
return super.toString();
93+
}
94+
8295
public static ConsistencyLevel fromString(String str)
8396
{
8497
return valueOf(str.toUpperCase(Locale.US));
@@ -96,6 +109,12 @@ public static int quorumFor(AbstractReplicationStrategy replicationStrategy)
96109
return (replicationStrategy.getReplicationFactor().allReplicas / 2) + 1;
97110
}
98111

112+
static int allButOneFor(AbstractReplicationStrategy replicationStrategy)
113+
{
114+
int rf = replicationStrategy.getReplicationFactor().fullReplicas;
115+
return rf <= 1 ? rf : rf - 1;
116+
}
117+
99118
public static int localQuorumFor(AbstractReplicationStrategy replicationStrategy, String dc)
100119
{
101120
return (replicationStrategy instanceof NetworkTopologyStrategy)
@@ -145,6 +164,10 @@ public int blockFor(AbstractReplicationStrategy replicationStrategy)
145164
case TWO:
146165
return 2;
147166
case THREE:
167+
if (THREE_MEANS_ALL_BUT_ONE)
168+
{
169+
return allButOneFor(replicationStrategy);
170+
}
148171
return 3;
149172
case QUORUM:
150173
case SERIAL:

‎test/unit/org/apache/cassandra/db/ConsistencyLevelTest.java‎

Lines changed: 181 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,181 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.cassandra.db;
20+
21+
import java.net.InetAddress;
22+
import java.net.UnknownHostException;
23+
import java.util.Collections;
24+
import java.util.HashMap;
25+
import java.util.Map;
26+
27+
import com.google.common.collect.HashMultimap;
28+
import com.google.common.collect.Multimap;
29+
import org.junit.After;
30+
import org.junit.AfterClass;
31+
import org.junit.BeforeClass;
32+
import org.junit.Test;
33+
34+
import org.apache.cassandra.config.DatabaseDescriptor;
35+
import org.apache.cassandra.dht.OrderPreservingPartitioner;
36+
import org.apache.cassandra.dht.Token;
37+
import org.apache.cassandra.locator.AbstractNetworkTopologySnitch;
38+
import org.apache.cassandra.locator.AbstractReplicationStrategy;
39+
import org.apache.cassandra.locator.EverywhereStrategy;
40+
import org.apache.cassandra.locator.IEndpointSnitch;
41+
import org.apache.cassandra.locator.InetAddressAndPort;
42+
import org.apache.cassandra.locator.LocalStrategy;
43+
import org.apache.cassandra.locator.NetworkTopologyStrategy;
44+
import org.apache.cassandra.locator.SimpleStrategy;
45+
import org.apache.cassandra.locator.TokenMetadata;
46+
import org.apache.cassandra.utils.Pair;
47+
48+
import static org.junit.Assert.*;
49+
50+
public class ConsistencyLevelTest
51+
{
52+
private static final String KS = "test";
53+
private static final Map<String, String> RACK = new HashMap<>(), DATACENTER = new HashMap<>();
54+
private static final IEndpointSnitch SNITCH = new AbstractNetworkTopologySnitch()
55+
{
56+
@Override
57+
public String getRack(InetAddressAndPort endpoint)
58+
{
59+
return RACK.getOrDefault(endpoint.getHostAddress(false), "RC1");
60+
}
61+
62+
@Override
63+
public String getDatacenter(InetAddressAndPort endpoint)
64+
{
65+
return DATACENTER.getOrDefault(endpoint.getHostAddress(false), "DC1");
66+
}
67+
};
68+
69+
@BeforeClass
70+
public static void setSnitch()
71+
{
72+
DatabaseDescriptor.daemonInitialization();
73+
DatabaseDescriptor.setEndpointSnitch(SNITCH);
74+
}
75+
76+
@AfterClass
77+
public static void resetSnitch()
78+
{
79+
DatabaseDescriptor.setEndpointSnitch(null);
80+
}
81+
82+
@After
83+
public void resetSnitchState()
84+
{
85+
RACK.clear();
86+
DATACENTER.clear();
87+
}
88+
89+
@Test
90+
public void allButOne_shouldBe_2_forReplicationFactor_3()
91+
{
92+
testAllButOne(simpleStrategy(3), 2);
93+
}
94+
95+
@Test
96+
public void allButOne_shouldBe_1_forReplicationFactor_2()
97+
{
98+
testAllButOne(simpleStrategy(2), 1);
99+
}
100+
101+
@Test
102+
public void allButOne_shouldBe_1_forReplicationFactor_1()
103+
{
104+
testAllButOne(simpleStrategy(1), 1);
105+
}
106+
107+
@Test
108+
public void allButOne_shouldBe_1_forLocalStrategy()
109+
{
110+
testAllButOne(localStrategy(), 1);
111+
}
112+
113+
@Test
114+
public void allButOne_shouldBe_8_forReplicationFactor_3_3_3()
115+
{
116+
testAllButOne(networkTopologyStrategy(3, 3, 3), 8);
117+
}
118+
119+
@Test
120+
public void allButOne_shouldBe_11_forEverywhereStrategyOnClusterOf_12() throws Exception
121+
{
122+
testAllButOne(everywhereStrategy(
123+
dc(1, Pair.create("192.168.0.1", "A"), Pair.create("192.168.0.2", "E"), Pair.create("192.168.0.3", "H"),
124+
Pair.create("192.168.0.4", "C"), Pair.create("192.168.0.5", "I"), Pair.create("192.168.0.6", "J")),
125+
dc(2, Pair.create("192.168.1.1", "B"), Pair.create("192.168.1.2", "G"), Pair.create("192.168.1.3", "L"),
126+
Pair.create("192.168.1.4", "D"), Pair.create("192.168.1.5", "F"), Pair.create("192.168.1.6", "K"))),
127+
11);
128+
}
129+
130+
private void testAllButOne(AbstractReplicationStrategy replicationStrategy, int expected)
131+
{
132+
// when
133+
int blockFor = ConsistencyLevel.allButOneFor(replicationStrategy);
134+
135+
// then
136+
assertEquals("number of nodes to block for", expected, blockFor);
137+
}
138+
139+
private static NetworkTopologyStrategy networkTopologyStrategy(int... dc)
140+
{
141+
Map<String, String> config = new HashMap<>();
142+
for (int i = 0; i < dc.length; i++)
143+
{
144+
config.put("DC" + i, Integer.toString(dc[i]));
145+
}
146+
return new NetworkTopologyStrategy(KS, new TokenMetadata(), SNITCH, config);
147+
}
148+
149+
private static AbstractReplicationStrategy simpleStrategy(int replicationFactory)
150+
{
151+
Map<String, String> config = Collections.singletonMap("replication_factor", Integer.toString(replicationFactory));
152+
return new SimpleStrategy(KS, new TokenMetadata(), SNITCH, config);
153+
}
154+
155+
@SafeVarargs
156+
private static AbstractReplicationStrategy everywhereStrategy(Multimap<InetAddressAndPort, Token>... dcs)
157+
{
158+
TokenMetadata metadata = new TokenMetadata();
159+
for (Multimap<InetAddressAndPort, Token> dc : dcs)
160+
{
161+
metadata.updateNormalTokens(dc);
162+
}
163+
return new EverywhereStrategy(KS, metadata, SNITCH, Collections.emptyMap());
164+
}
165+
166+
private static AbstractReplicationStrategy localStrategy()
167+
{
168+
return new LocalStrategy(KS, new TokenMetadata(), SNITCH, Collections.emptyMap());
169+
}
170+
171+
private static Multimap<InetAddressAndPort, Token> dc(int id, Pair<String, String>... addressToken) throws UnknownHostException
172+
{
173+
Multimap<InetAddressAndPort, Token> dc = HashMultimap.create();
174+
for (Pair<String, String> pair : addressToken)
175+
{
176+
DATACENTER.put(pair.left, "DC" + id);
177+
dc.put(InetAddressAndPort.getByName(pair.left), new OrderPreservingPartitioner.StringToken(pair.right));
178+
}
179+
return dc;
180+
}
181+
}

0 commit comments

Comments
 (0)