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

Fix inconsistencies in cassandra-stress load balancing policy 

patch by Stefania Alborghetti; reviewed by Jake Luciani for CASSANDRA-12919
This commit is contained in:
Stefania Alborghetti 2016-11-17 11:36:22 +08:00
parent 595f1da9ef
commit 12c63cf4a4
2 changed files with 20 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.11
* Fix inconsistencies in cassandra-stress load balancing policy (CASSANDRA-12919)
* Fix validation of non-frozen UDT cells (CASSANDRA-12916)
* AnticompactionRequestSerializer serializedSize is incorrect (CASSANDRA-12934)

33
tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java
View File

@ -25,6 +25,7 @@ import javax.net.ssl.SSLContext;
import com.datastax.driver.core.*;
import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;
import com.datastax.driver.core.policies.LoadBalancingPolicy;
import com.datastax.driver.core.policies.TokenAwarePolicy;
import com.datastax.driver.core.policies.WhiteListPolicy;
import io.netty.util.internal.logging.InternalLoggerFactory;
import io.netty.util.internal.logging.Slf4JLoggerFactory;
@ -70,19 +71,8 @@ public class JavaDriverClient
this.password = settings.mode.password;
this.authProvider = settings.mode.authProvider;
this.encryptionOptions = encryptionOptions;
DCAwareRoundRobinPolicy.Builder policyBuilder = DCAwareRoundRobinPolicy.builder();
if (settings.node.datacenter != null)
policyBuilder.withLocalDc(settings.node.datacenter);
if (settings.node.isWhiteList)
loadBalancingPolicy = new WhiteListPolicy(policyBuilder.build(), settings.node.resolveAll(settings.port.nativePort));
else if (settings.node.datacenter != null)
loadBalancingPolicy = policyBuilder.build();
else
loadBalancingPolicy = null;
connectionsPerHost = settings.mode.connectionsPerHost == null ? 8 : settings.mode.connectionsPerHost;
this.loadBalancingPolicy = loadBalancingPolicy(settings);
this.connectionsPerHost = settings.mode.connectionsPerHost == null ? 8 : settings.mode.connectionsPerHost;
int maxThreadCount = 0;
if (settings.rate.auto)
@ -97,6 +87,22 @@ public class JavaDriverClient
maxPendingPerConnection = settings.mode.maxPendingPerConnection == null ? Math.max(128, requestsPerConnection ) : settings.mode.maxPendingPerConnection;
}
private LoadBalancingPolicy loadBalancingPolicy(StressSettings settings)
{
DCAwareRoundRobinPolicy.Builder policyBuilder = DCAwareRoundRobinPolicy.builder();
if (settings.node.datacenter != null)
policyBuilder.withLocalDc(settings.node.datacenter);
LoadBalancingPolicy ret = null;
if (settings.node.datacenter != null)
ret = policyBuilder.build();
if (settings.node.isWhiteList)
ret = new WhiteListPolicy(ret == null ? policyBuilder.build() : ret, settings.node.resolveAll(settings.port.nativePort));
return new TokenAwarePolicy(ret == null ? policyBuilder.build() : ret);
}
public PreparedStatement prepare(String query)
{
PreparedStatement stmt = stmts.get(query);
@ -185,7 +191,6 @@ public class JavaDriverClient
public ResultSet executePrepared(PreparedStatement stmt, List<Object> queryParams, org.apache.cassandra.db.ConsistencyLevel consistency)
{
stmt.setConsistencyLevel(from(consistency));
BoundStatement bstmt = stmt.bind((Object[]) queryParams.toArray(new Object[queryParams.size()]));
return getSession().execute(bstmt);