mirror of https://github.com/apache/cassandra
Merge branch 'cassandra-3.0' into cassandra-3.11
This commit is contained in:
Sam Tunnicliffe
commit
3040965668
|
|
@ -3,7 +3,8 @@
|
|||
* Fix cassandra-env.sh to use $CASSANDRA_CONF to find cassandra-jaas.config (CASSANDRA-14305)
|
||||
* Fixed nodetool cfstats printing index name twice (CASSANDRA-14903)
|
||||
* Add flag to disable SASI indexes, and warnings on creation (CASSANDRA-14866)
|
||||
Merged from 3.0:
|
||||
Merged from 3.0:a
|
||||
* Add ability to cap max negotiable protocol version (CASSANDRA-15193)
|
||||
* Gossip tokens on startup if available (CASSANDRA-15335)
|
||||
* Fix resource leak in CompressedSequentialWriter (CASSANDRA-15340)
|
||||
* Fix bad merge that reverted CASSANDRA-14993 (CASSANDRA-15289)
|
||||
|
|
|
|||
5
NEWS.txt
5
NEWS.txt
|
|
@ -49,6 +49,11 @@ Upgrading
|
|||
---------
|
||||
- repair_session_max_tree_depth setting has been added to cassandra.yaml to allow operators to reduce
|
||||
merkle tree size if repair is creating too much heap pressure. See CASSANDRA-14096 for details.
|
||||
- native_transport_max_negotiable_protocol_version has been added to cassandra.yaml to allow operators to
|
||||
enforce an upper limit on the version of the native protocol that servers will negotiate with clients.
|
||||
This can be used during upgrades from 2.1 to 3.0 to prevent errors due to incompatible paging state formats
|
||||
between the two versions. See CASSANDRA-15193 for details.
|
||||
|
||||
|
||||
Experimental features
|
||||
---------------------
|
||||
|
|
|
|||
|
|
@ -622,9 +622,9 @@ class Shell(cmd.Cmd):
|
|||
result, = self.session.execute("select * from system.local where key = 'local'")
|
||||
vers = {
|
||||
'build': result['release_version'],
|
||||
'protocol': result['native_protocol_version'],
|
||||
'cql': result['cql_version'],
|
||||
}
|
||||
vers['protocol'] = self.conn.protocol_version
|
||||
self.connection_versions = vers
|
||||
|
||||
def get_keyspace_names(self):
|
||||
|
|
|
|||
|
|
@ -163,7 +163,7 @@ public class Config
|
|||
public boolean native_transport_flush_in_batches_legacy = true;
|
||||
public volatile long native_transport_max_concurrent_requests_in_bytes_per_ip = -1L;
|
||||
public volatile long native_transport_max_concurrent_requests_in_bytes = -1L;
|
||||
|
||||
public Integer native_transport_max_negotiable_protocol_version = Integer.MIN_VALUE;
|
||||
|
||||
@Deprecated
|
||||
public int thrift_max_message_length_in_mb = 16;
|
||||
|
|
|
|||
|
|
@ -61,6 +61,8 @@ import org.apache.cassandra.scheduler.NoScheduler;
|
|||
import org.apache.cassandra.security.EncryptionContext;
|
||||
import org.apache.cassandra.service.CacheService.CacheType;
|
||||
import org.apache.cassandra.thrift.ThriftServer.ThriftServerType;
|
||||
import org.apache.cassandra.transport.ProtocolVersion;
|
||||
import org.apache.cassandra.transport.ProtocolVersionLimit;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
|
|
@ -728,6 +730,21 @@ public class DatabaseDescriptor
|
|||
throw new ConfigurationException("Encryption must be enabled in client_encryption_options for native_transport_port_ssl", false);
|
||||
}
|
||||
|
||||
// If max protocol version has been set, just validate it's within an acceptable range
|
||||
if (conf.native_transport_max_negotiable_protocol_version != Integer.MIN_VALUE)
|
||||
{
|
||||
try
|
||||
{
|
||||
ProtocolVersion.decode(conf.native_transport_max_negotiable_protocol_version, ProtocolVersionLimit.SERVER_DEFAULT);
|
||||
logger.info("Native transport max negotiable version statically limited to {}", conf.native_transport_max_negotiable_protocol_version);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
throw new ConfigurationException("Invalid setting for native_transport_max_negotiable_protocol_version; " +
|
||||
ProtocolVersion.invalidVersionMessage(conf.native_transport_max_negotiable_protocol_version));
|
||||
}
|
||||
}
|
||||
|
||||
if (conf.max_value_size_in_mb <= 0)
|
||||
throw new ConfigurationException("max_value_size_in_mb must be positive", false);
|
||||
else if (conf.max_value_size_in_mb >= 2048)
|
||||
|
|
@ -1849,6 +1866,11 @@ public class DatabaseDescriptor
|
|||
return conf.native_transport_flush_in_batches_legacy;
|
||||
}
|
||||
|
||||
public static int getNativeProtocolMaxVersionOverride()
|
||||
{
|
||||
return conf.native_transport_max_negotiable_protocol_version;
|
||||
}
|
||||
|
||||
public static double getCommitLogSyncBatchWindow()
|
||||
{
|
||||
return conf.commitlog_sync_batch_window_in_ms;
|
||||
|
|
|
|||
|
|
@ -757,6 +757,18 @@ public final class SystemKeyspace
|
|||
return executorService.submit((Runnable) () -> executeInternal(String.format(req, PEERS, columnName), ep, value));
|
||||
}
|
||||
|
||||
public static void updatePeerReleaseVersion(final InetAddress ep, final Object value, Runnable postUpdateTask, ExecutorService executorService)
|
||||
{
|
||||
if (ep.equals(FBUtilities.getBroadcastAddress()))
|
||||
return;
|
||||
|
||||
String req = "INSERT INTO system.%s (peer, release_version) VALUES (?, ?)";
|
||||
executorService.execute(() -> {
|
||||
executeInternal(String.format(req, PEERS), ep, value);
|
||||
postUpdateTask.run();
|
||||
});
|
||||
}
|
||||
|
||||
public static synchronized void updateHintsDropped(InetAddress ep, UUID timePeriod, int value)
|
||||
{
|
||||
// with 30 day TTL
|
||||
|
|
@ -852,6 +864,37 @@ public final class SystemKeyspace
|
|||
return hostIdMap;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return a map of IP address to C* version. If an invalid version string, or no version
|
||||
* at all is stored for a given peer IP, then NULL_VERSION will be reported for that peer
|
||||
*/
|
||||
public static Map<InetAddress, CassandraVersion> loadPeerVersions()
|
||||
{
|
||||
Map<InetAddress, CassandraVersion> releaseVersionMap = new HashMap<>();
|
||||
for (UntypedResultSet.Row row : executeInternal("SELECT peer, release_version FROM system." + PEERS))
|
||||
{
|
||||
InetAddress peer = row.getInetAddress("peer");
|
||||
if (row.has("release_version"))
|
||||
{
|
||||
try
|
||||
{
|
||||
releaseVersionMap.put(peer, new CassandraVersion(row.getString("release_version")));
|
||||
}
|
||||
catch (IllegalArgumentException e)
|
||||
{
|
||||
logger.info("Invalid version string found for {}", peer);
|
||||
releaseVersionMap.put(peer, NULL_VERSION);
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
logger.info("No version string found for {}", peer);
|
||||
releaseVersionMap.put(peer, NULL_VERSION);
|
||||
}
|
||||
}
|
||||
return releaseVersionMap;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get preferred IP for given endpoint if it is known. Otherwise this returns given endpoint itself.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -707,6 +707,16 @@ public class CassandraDaemon
|
|||
return nativeTransportService != null ? nativeTransportService.isRunning() : false;
|
||||
}
|
||||
|
||||
public int getMaxNativeProtocolVersion()
|
||||
{
|
||||
return nativeTransportService.getMaxProtocolVersion();
|
||||
}
|
||||
|
||||
public void refreshMaxNativeProtocolVersion()
|
||||
{
|
||||
if (nativeTransportService != null)
|
||||
nativeTransportService.refreshMaxNegotiableProtocolVersion();
|
||||
}
|
||||
|
||||
/**
|
||||
* A convenience method to stop and destroy the daemon in one shot.
|
||||
|
|
|
|||
|
|
@ -35,6 +35,7 @@ import io.netty.util.concurrent.EventExecutor;
|
|||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.metrics.AuthMetrics;
|
||||
import org.apache.cassandra.metrics.ClientMetrics;
|
||||
import org.apache.cassandra.transport.ConfiguredLimit;
|
||||
import org.apache.cassandra.transport.Message;
|
||||
import org.apache.cassandra.transport.Server;
|
||||
|
||||
|
|
@ -50,6 +51,7 @@ public class NativeTransportService
|
|||
|
||||
private boolean initialized = false;
|
||||
private EventLoopGroup workerGroup;
|
||||
private ConfiguredLimit protocolVersionLimit;
|
||||
|
||||
/**
|
||||
* Creates netty thread pools and event loops.
|
||||
|
|
@ -71,12 +73,15 @@ public class NativeTransportService
|
|||
logger.info("Netty using Java NIO event loop");
|
||||
}
|
||||
|
||||
protocolVersionLimit = ConfiguredLimit.newLimit();
|
||||
|
||||
int nativePort = DatabaseDescriptor.getNativeTransportPort();
|
||||
int nativePortSSL = DatabaseDescriptor.getNativeTransportPortSSL();
|
||||
InetAddress nativeAddr = DatabaseDescriptor.getRpcAddress();
|
||||
|
||||
org.apache.cassandra.transport.Server.Builder builder = new org.apache.cassandra.transport.Server.Builder()
|
||||
.withEventLoopGroup(workerGroup)
|
||||
.withProtocolVersionLimit(protocolVersionLimit)
|
||||
.withHost(nativeAddr);
|
||||
|
||||
if (!DatabaseDescriptor.getClientEncryptionOptions().enabled)
|
||||
|
|
@ -141,6 +146,20 @@ public class NativeTransportService
|
|||
Message.Dispatcher.shutdown();
|
||||
}
|
||||
|
||||
public int getMaxProtocolVersion()
|
||||
{
|
||||
return protocolVersionLimit.getMaxVersion().asInt();
|
||||
}
|
||||
|
||||
public void refreshMaxNegotiableProtocolVersion()
|
||||
{
|
||||
// lowering the max negotiable protocol version is only safe if we haven't already
|
||||
// allowed clients to connect with a higher version. This still allows the max
|
||||
// version to be raised, as that is safe.
|
||||
if (initialized)
|
||||
protocolVersionLimit.updateMaxSupportedVersion();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return intend to use epoll bassed event looping
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -441,6 +441,23 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
|
|||
return daemon.isNativeTransportRunning();
|
||||
}
|
||||
|
||||
public int getMaxNativeProtocolVersion()
|
||||
{
|
||||
if (daemon == null)
|
||||
{
|
||||
throw new IllegalStateException("No configured daemon");
|
||||
}
|
||||
return daemon.getMaxNativeProtocolVersion();
|
||||
}
|
||||
|
||||
private void refreshMaxNativeProtocolVersion()
|
||||
{
|
||||
if (daemon != null)
|
||||
{
|
||||
daemon.refreshMaxNativeProtocolVersion();
|
||||
}
|
||||
}
|
||||
|
||||
public void stopTransports()
|
||||
{
|
||||
if (isGossipActive())
|
||||
|
|
@ -2050,7 +2067,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
|
|||
switch (state)
|
||||
{
|
||||
case RELEASE_VERSION:
|
||||
SystemKeyspace.updatePeerInfo(endpoint, "release_version", value.value, executor);
|
||||
SystemKeyspace.updatePeerReleaseVersion(endpoint, value.value, this::refreshMaxNativeProtocolVersion, executor);
|
||||
break;
|
||||
case DC:
|
||||
updateTopology(endpoint);
|
||||
|
|
@ -2127,7 +2144,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
|
|||
switch (entry.getKey())
|
||||
{
|
||||
case RELEASE_VERSION:
|
||||
SystemKeyspace.updatePeerInfo(endpoint, "release_version", entry.getValue().value, executor);
|
||||
SystemKeyspace.updatePeerReleaseVersion(endpoint, entry.getValue().value, this::refreshMaxNativeProtocolVersion, executor);
|
||||
break;
|
||||
case DC:
|
||||
SystemKeyspace.updatePeerInfo(endpoint, "data_center", entry.getValue().value, executor);
|
||||
|
|
|
|||
|
|
@ -693,4 +693,7 @@ public interface StorageServiceMBean extends NotificationEmitter
|
|||
* @return true if the node successfully starts resuming. (this does not mean bootstrap streaming was success.)
|
||||
*/
|
||||
public boolean resumeBootstrap();
|
||||
|
||||
/** Returns the max version that this node will negotiate for native protocol connections */
|
||||
public int getMaxNativeProtocolVersion();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -251,7 +251,7 @@ public class Client extends SimpleClient
|
|||
// Parse options.
|
||||
String host = args[0];
|
||||
int port = Integer.parseInt(args[1]);
|
||||
ProtocolVersion version = args.length == 3 ? ProtocolVersion.decode(Integer.parseInt(args[2])) : ProtocolVersion.CURRENT;
|
||||
ProtocolVersion version = args.length == 3 ? ProtocolVersion.decode(Integer.parseInt(args[2]), ProtocolVersionLimit.SERVER_DEFAULT) : ProtocolVersion.CURRENT;
|
||||
|
||||
ClientEncryptionOptions encryptionOptions = new ClientEncryptionOptions();
|
||||
System.out.println("CQL binary protocol console " + host + "@" + port + " using native protocol version " + version);
|
||||
|
|
|
|||
|
|
@ -0,0 +1,112 @@
|
|||
/*
|
||||
* 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.transport;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.db.SystemKeyspace;
|
||||
import org.apache.cassandra.utils.CassandraVersion;
|
||||
|
||||
public abstract class ConfiguredLimit implements ProtocolVersionLimit
|
||||
{
|
||||
private static final Logger logger = LoggerFactory.getLogger(ConfiguredLimit.class);
|
||||
static final String DISABLE_MAX_PROTOCOL_AUTO_OVERRIDE = "cassandra.disable_max_protocol_auto_override";
|
||||
static final CassandraVersion MIN_VERSION_FOR_V4 = new CassandraVersion("3.0.0");
|
||||
|
||||
public abstract ProtocolVersion getMaxVersion();
|
||||
public abstract void updateMaxSupportedVersion();
|
||||
|
||||
public static ConfiguredLimit newLimit()
|
||||
{
|
||||
if (Boolean.getBoolean(DISABLE_MAX_PROTOCOL_AUTO_OVERRIDE))
|
||||
return new StaticLimit(ProtocolVersion.MAX_SUPPORTED_VERSION);
|
||||
|
||||
int fromConfig = DatabaseDescriptor.getNativeProtocolMaxVersionOverride();
|
||||
return fromConfig != Integer.MIN_VALUE
|
||||
? new StaticLimit(ProtocolVersion.decode(fromConfig, ProtocolVersionLimit.SERVER_DEFAULT))
|
||||
: new DynamicLimit(ProtocolVersion.MAX_SUPPORTED_VERSION);
|
||||
}
|
||||
|
||||
private static class StaticLimit extends ConfiguredLimit
|
||||
{
|
||||
private final ProtocolVersion maxVersion;
|
||||
private StaticLimit(ProtocolVersion maxVersion)
|
||||
{
|
||||
this.maxVersion = maxVersion;
|
||||
logger.info("Native transport max negotiable version statically limited to {}", maxVersion);
|
||||
}
|
||||
|
||||
public ProtocolVersion getMaxVersion()
|
||||
{
|
||||
return maxVersion;
|
||||
}
|
||||
|
||||
public void updateMaxSupportedVersion()
|
||||
{
|
||||
// statically configured, so this is a no-op
|
||||
}
|
||||
}
|
||||
|
||||
private static class DynamicLimit extends ConfiguredLimit
|
||||
{
|
||||
private volatile ProtocolVersion maxVersion;
|
||||
private DynamicLimit(ProtocolVersion initialLimit)
|
||||
{
|
||||
maxVersion = initialLimit;
|
||||
maybeUpdateVersion(true);
|
||||
}
|
||||
|
||||
public ProtocolVersion getMaxVersion()
|
||||
{
|
||||
return maxVersion;
|
||||
}
|
||||
|
||||
public void updateMaxSupportedVersion()
|
||||
{
|
||||
maybeUpdateVersion(false);
|
||||
}
|
||||
|
||||
private void maybeUpdateVersion(boolean allowLowering)
|
||||
{
|
||||
boolean enforceV3Cap = SystemKeyspace.loadPeerVersions()
|
||||
.values()
|
||||
.stream()
|
||||
.anyMatch(v -> v.compareTo(MIN_VERSION_FOR_V4) < 0);
|
||||
|
||||
if (!enforceV3Cap)
|
||||
{
|
||||
maxVersion = ProtocolVersion.MAX_SUPPORTED_VERSION;
|
||||
return;
|
||||
}
|
||||
|
||||
if (ProtocolVersion.V3.isSmallerThan(maxVersion) && !allowLowering)
|
||||
{
|
||||
logger.info("Detected peers which do not fully support protocol V4, but V4 was previously negotiable. " +
|
||||
"Not enforcing cap as this can cause issues for older client versions. After the next " +
|
||||
"restart the server will apply the cap");
|
||||
return;
|
||||
}
|
||||
|
||||
logger.info("Detected peers which do not fully support protocol V4. Capping max negotiable version to V3");
|
||||
maxVersion = ProtocolVersion.V3;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -142,10 +142,12 @@ public class Frame
|
|||
private int tooLongStreamId;
|
||||
|
||||
private final Connection.Factory factory;
|
||||
private final ProtocolVersionLimit versionCap;
|
||||
|
||||
public Decoder(Connection.Factory factory)
|
||||
public Decoder(Connection.Factory factory, ProtocolVersionLimit versionCap)
|
||||
{
|
||||
this.factory = factory;
|
||||
this.versionCap = versionCap;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -172,7 +174,7 @@ public class Frame
|
|||
int firstByte = buffer.getByte(idx++);
|
||||
Message.Direction direction = Message.Direction.extractFromVersion(firstByte);
|
||||
int versionNum = firstByte & PROTOCOL_VERSION_MASK;
|
||||
ProtocolVersion version = ProtocolVersion.decode(versionNum);
|
||||
ProtocolVersion version = ProtocolVersion.decode(versionNum, versionCap);
|
||||
|
||||
// Wait until we have the complete header
|
||||
if (readableBytes < Header.LENGTH)
|
||||
|
|
|
|||
|
|
@ -322,11 +322,19 @@ public abstract class Message
|
|||
@ChannelHandler.Sharable
|
||||
public static class ProtocolEncoder extends MessageToMessageEncoder<Message>
|
||||
{
|
||||
private final ProtocolVersionLimit versionCap;
|
||||
|
||||
ProtocolEncoder(ProtocolVersionLimit versionCap)
|
||||
{
|
||||
this.versionCap = versionCap;
|
||||
}
|
||||
|
||||
public void encode(ChannelHandlerContext ctx, Message message, List results)
|
||||
{
|
||||
Connection connection = ctx.channel().attr(Connection.attributeKey).get();
|
||||
// The only case the connection can be null is when we send the initial STARTUP message (client side thus)
|
||||
ProtocolVersion version = connection == null ? ProtocolVersion.CURRENT : connection.getVersion();
|
||||
ProtocolVersion version = connection == null ? versionCap.getMaxVersion() : connection.getVersion();
|
||||
|
||||
EnumSet<Frame.Header.Flag> flags = EnumSet.noneOf(Frame.Header.Flag.class);
|
||||
|
||||
Codec<Message> codec = (Codec<Message>)message.type.codec;
|
||||
|
|
|
|||
|
|
@ -84,9 +84,9 @@ public enum ProtocolVersion implements Comparable<ProtocolVersion>
|
|||
return ret;
|
||||
}
|
||||
|
||||
public static ProtocolVersion decode(int versionNum)
|
||||
public static ProtocolVersion decode(int versionNum, ProtocolVersionLimit ceiling)
|
||||
{
|
||||
ProtocolVersion ret = versionNum >= MIN_SUPPORTED_VERSION.num && versionNum <= MAX_SUPPORTED_VERSION.num
|
||||
ProtocolVersion ret = versionNum >= MIN_SUPPORTED_VERSION.num && versionNum <= ceiling.getMaxVersion().num
|
||||
? SUPPORTED_VERSIONS[versionNum - MIN_SUPPORTED_VERSION.num]
|
||||
: null;
|
||||
|
||||
|
|
@ -102,7 +102,7 @@ public enum ProtocolVersion implements Comparable<ProtocolVersion>
|
|||
}
|
||||
|
||||
// If the version is invalid reply with the highest version that we support
|
||||
throw new ProtocolException(invalidVersionMessage(versionNum), MAX_SUPPORTED_VERSION);
|
||||
throw new ProtocolException(invalidVersionMessage(versionNum), ceiling.getMaxVersion());
|
||||
}
|
||||
|
||||
return ret;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,27 @@
|
|||
/*
|
||||
* 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.transport;
|
||||
|
||||
@FunctionalInterface
|
||||
public interface ProtocolVersionLimit
|
||||
{
|
||||
public ProtocolVersion getMaxVersion();
|
||||
|
||||
public static final ProtocolVersionLimit SERVER_DEFAULT = () -> ProtocolVersion.MAX_SUPPORTED_VERSION;
|
||||
}
|
||||
|
|
@ -82,11 +82,14 @@ public class Server implements CassandraDaemon.Server
|
|||
private final AtomicBoolean isRunning = new AtomicBoolean(false);
|
||||
|
||||
private EventLoopGroup workerGroup;
|
||||
private final ProtocolVersionLimit protocolVersionLimit;
|
||||
|
||||
private Server (Builder builder)
|
||||
{
|
||||
this.socket = builder.getSocket();
|
||||
this.useSSL = builder.useSSL;
|
||||
this.protocolVersionLimit = builder.getProtocolVersionLimit();
|
||||
|
||||
if (builder.workerGroup != null)
|
||||
{
|
||||
workerGroup = builder.workerGroup;
|
||||
|
|
@ -184,6 +187,7 @@ public class Server implements CassandraDaemon.Server
|
|||
private InetAddress hostAddr;
|
||||
private int port = -1;
|
||||
private InetSocketAddress socket;
|
||||
private ProtocolVersionLimit versionLimit;
|
||||
|
||||
public Builder withSSL(boolean useSSL)
|
||||
{
|
||||
|
|
@ -211,6 +215,19 @@ public class Server implements CassandraDaemon.Server
|
|||
return this;
|
||||
}
|
||||
|
||||
public Builder withProtocolVersionLimit(ProtocolVersionLimit limit)
|
||||
{
|
||||
this.versionLimit = limit;
|
||||
return this;
|
||||
}
|
||||
|
||||
ProtocolVersionLimit getProtocolVersionLimit()
|
||||
{
|
||||
if (versionLimit == null)
|
||||
throw new IllegalArgumentException("Missing protocol version limiter");
|
||||
return versionLimit;
|
||||
}
|
||||
|
||||
public Server build()
|
||||
{
|
||||
return new Server(this);
|
||||
|
|
@ -323,7 +340,6 @@ public class Server implements CassandraDaemon.Server
|
|||
{
|
||||
// Stateless handlers
|
||||
private static final Message.ProtocolDecoder messageDecoder = new Message.ProtocolDecoder();
|
||||
private static final Message.ProtocolEncoder messageEncoder = new Message.ProtocolEncoder();
|
||||
private static final Frame.Decompressor frameDecompressor = new Frame.Decompressor();
|
||||
private static final Frame.Compressor frameCompressor = new Frame.Compressor();
|
||||
private static final Frame.Encoder frameEncoder = new Frame.Encoder();
|
||||
|
|
@ -351,14 +367,14 @@ public class Server implements CassandraDaemon.Server
|
|||
|
||||
//pipeline.addLast("debug", new LoggingHandler());
|
||||
|
||||
pipeline.addLast("frameDecoder", new Frame.Decoder(server.connectionFactory));
|
||||
pipeline.addLast("frameDecoder", new Frame.Decoder(server.connectionFactory, server.protocolVersionLimit));
|
||||
pipeline.addLast("frameEncoder", frameEncoder);
|
||||
|
||||
pipeline.addLast("frameDecompressor", frameDecompressor);
|
||||
pipeline.addLast("frameCompressor", frameCompressor);
|
||||
|
||||
pipeline.addLast("messageDecoder", messageDecoder);
|
||||
pipeline.addLast("messageEncoder", messageEncoder);
|
||||
pipeline.addLast("messageEncoder", new Message.ProtocolEncoder(server.protocolVersionLimit));
|
||||
|
||||
pipeline.addLast("executor", new Message.Dispatcher(DatabaseDescriptor.useNativeTransportLegacyFlusher(),
|
||||
EndpointPayloadTracker.get(((InetSocketAddress) channel.remoteAddress()).getAddress())));
|
||||
|
|
|
|||
|
|
@ -259,7 +259,7 @@ public class SimpleClient implements Closeable
|
|||
|
||||
// Stateless handlers
|
||||
private static final Message.ProtocolDecoder messageDecoder = new Message.ProtocolDecoder();
|
||||
private static final Message.ProtocolEncoder messageEncoder = new Message.ProtocolEncoder();
|
||||
private static final Message.ProtocolEncoder messageEncoder = new Message.ProtocolEncoder(ProtocolVersionLimit.SERVER_DEFAULT);
|
||||
private static final Frame.Decompressor frameDecompressor = new Frame.Decompressor();
|
||||
private static final Frame.Compressor frameCompressor = new Frame.Compressor();
|
||||
private static final Frame.Encoder frameEncoder = new Frame.Encoder();
|
||||
|
|
@ -282,7 +282,7 @@ public class SimpleClient implements Closeable
|
|||
channel.attr(Connection.attributeKey).set(connection);
|
||||
|
||||
ChannelPipeline pipeline = channel.pipeline();
|
||||
pipeline.addLast("frameDecoder", new Frame.Decoder(connectionFactory));
|
||||
pipeline.addLast("frameDecoder", new Frame.Decoder(connectionFactory, ProtocolVersionLimit.SERVER_DEFAULT));
|
||||
pipeline.addLast("frameEncoder", frameEncoder);
|
||||
|
||||
pipeline.addLast("frameDecompressor", frameDecompressor);
|
||||
|
|
|
|||
|
|
@ -62,8 +62,10 @@ import org.apache.cassandra.serializers.TypeSerializer;
|
|||
import org.apache.cassandra.service.ClientState;
|
||||
import org.apache.cassandra.service.QueryState;
|
||||
import org.apache.cassandra.service.StorageService;
|
||||
import org.apache.cassandra.transport.*;
|
||||
import org.apache.cassandra.transport.ProtocolVersion;
|
||||
import org.apache.cassandra.transport.ConfiguredLimit;
|
||||
import org.apache.cassandra.transport.Event;
|
||||
import org.apache.cassandra.transport.Server;
|
||||
import org.apache.cassandra.transport.messages.ResultMessage;
|
||||
import org.apache.cassandra.utils.ByteBufferUtil;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
|
|
@ -91,6 +93,7 @@ public abstract class CQLTester
|
|||
protected static final InetAddress nativeAddr;
|
||||
private static final Map<ProtocolVersion, Cluster> clusters = new HashMap<>();
|
||||
private static final Map<ProtocolVersion, Session> sessions = new HashMap<>();
|
||||
protected static ConfiguredLimit protocolVersionLimit;
|
||||
|
||||
private static boolean isServerPrepared = false;
|
||||
|
||||
|
|
@ -353,11 +356,43 @@ public abstract class CQLTester
|
|||
if (server != null)
|
||||
return;
|
||||
|
||||
prepareNetwork();
|
||||
initializeNetwork();
|
||||
}
|
||||
|
||||
protected static void prepareNetwork()
|
||||
{
|
||||
SystemKeyspace.finishStartup();
|
||||
StorageService.instance.initServer();
|
||||
SchemaLoader.startGossiper();
|
||||
}
|
||||
|
||||
server = new Server.Builder().withHost(nativeAddr).withPort(nativePort).build();
|
||||
protected static void reinitializeNetwork()
|
||||
{
|
||||
if (server != null && server.isRunning())
|
||||
{
|
||||
server.stop();
|
||||
server = null;
|
||||
}
|
||||
List<CloseFuture> futures = new ArrayList<>();
|
||||
for (Cluster cluster : clusters.values())
|
||||
futures.add(cluster.closeAsync());
|
||||
for (Session session : sessions.values())
|
||||
futures.add(session.closeAsync());
|
||||
FBUtilities.waitOnFutures(futures);
|
||||
clusters.clear();
|
||||
sessions.clear();
|
||||
|
||||
initializeNetwork();
|
||||
}
|
||||
|
||||
private static void initializeNetwork()
|
||||
{
|
||||
protocolVersionLimit = ConfiguredLimit.newLimit();
|
||||
server = new Server.Builder().withHost(nativeAddr)
|
||||
.withPort(nativePort)
|
||||
.withProtocolVersionLimit(protocolVersionLimit)
|
||||
.build();
|
||||
ClientMetrics.instance.init(Collections.singleton(server));
|
||||
server.start();
|
||||
|
||||
|
|
@ -366,9 +401,12 @@ public abstract class CQLTester
|
|||
if (clusters.containsKey(version))
|
||||
continue;
|
||||
|
||||
if (version.isGreaterThan(protocolVersionLimit.getMaxVersion()))
|
||||
continue;
|
||||
|
||||
Cluster cluster = Cluster.builder()
|
||||
.addContactPoints(nativeAddr)
|
||||
.withClusterName("Test Cluster")
|
||||
.withClusterName("Test Cluster-" + version.name())
|
||||
.withPort(nativePort)
|
||||
.withProtocolVersion(com.datastax.driver.core.ProtocolVersion.fromInt(version.asInt()))
|
||||
.build();
|
||||
|
|
@ -379,6 +417,14 @@ public abstract class CQLTester
|
|||
}
|
||||
}
|
||||
|
||||
protected void updateMaxNegotiableProtocolVersion()
|
||||
{
|
||||
if (protocolVersionLimit == null)
|
||||
throw new IllegalStateException("Native transport server has not been initialized");
|
||||
|
||||
protocolVersionLimit.updateMaxSupportedVersion();
|
||||
}
|
||||
|
||||
protected void dropPerTestKeyspace() throws Throwable
|
||||
{
|
||||
execute(String.format("DROP KEYSPACE IF EXISTS %s", KEYSPACE_PER_TEST));
|
||||
|
|
|
|||
|
|
@ -0,0 +1,111 @@
|
|||
/*
|
||||
* 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.transport;
|
||||
|
||||
import java.net.InetAddress;
|
||||
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.apache.cassandra.cql3.CQLTester;
|
||||
|
||||
import static org.apache.cassandra.transport.ProtocolTestHelper.cleanupPeers;
|
||||
import static org.apache.cassandra.transport.ProtocolTestHelper.setStaticLimitInConfig;
|
||||
import static org.apache.cassandra.transport.ProtocolTestHelper.setupPeer;
|
||||
import static org.apache.cassandra.transport.ProtocolTestHelper.updatePeerInfo;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
public class DynamicLimitTest
|
||||
{
|
||||
@BeforeClass
|
||||
public static void setup()
|
||||
{
|
||||
CQLTester.prepareServer();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void disableDynamicLimitWithSystemProperty() throws Throwable
|
||||
{
|
||||
// Dynamic limiting of the max negotiable protocol version can be
|
||||
// disabled with a system property
|
||||
|
||||
// ensure that no static limit is configured
|
||||
setStaticLimitInConfig(null);
|
||||
|
||||
// set the property which disables dynamic limiting
|
||||
System.setProperty(ConfiguredLimit.DISABLE_MAX_PROTOCOL_AUTO_OVERRIDE, "true");
|
||||
// insert a legacy peer into system.peers and also
|
||||
InetAddress peer = null;
|
||||
try
|
||||
{
|
||||
peer = setupPeer("127.1.0.1", "2.2.0");
|
||||
ConfiguredLimit limit = ConfiguredLimit.newLimit();
|
||||
assertEquals(ProtocolVersion.MAX_SUPPORTED_VERSION, limit.getMaxVersion());
|
||||
|
||||
// clearing the property after the limit has been returned has no effect
|
||||
System.clearProperty(ConfiguredLimit.DISABLE_MAX_PROTOCOL_AUTO_OVERRIDE);
|
||||
limit.updateMaxSupportedVersion();
|
||||
assertEquals(ProtocolVersion.MAX_SUPPORTED_VERSION, limit.getMaxVersion());
|
||||
|
||||
// a new limit should now be dynamic
|
||||
limit = ConfiguredLimit.newLimit();
|
||||
assertEquals(ProtocolVersion.V3, limit.getMaxVersion());
|
||||
}
|
||||
finally
|
||||
{
|
||||
System.clearProperty(ConfiguredLimit.DISABLE_MAX_PROTOCOL_AUTO_OVERRIDE);
|
||||
cleanupPeers(peer);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void disallowLoweringMaxVersion() throws Throwable
|
||||
{
|
||||
// Lowering the max version once connections have been established is a problem
|
||||
// for some clients. So for a dynamic limit, if notifications of peer versions
|
||||
// trigger a change to the max version, it's only allowed to increase the max
|
||||
// negotiable version
|
||||
|
||||
InetAddress peer = null;
|
||||
try
|
||||
{
|
||||
// ensure that no static limit is configured
|
||||
setStaticLimitInConfig(null);
|
||||
ConfiguredLimit limit = ConfiguredLimit.newLimit();
|
||||
assertEquals(ProtocolVersion.MAX_SUPPORTED_VERSION, limit.getMaxVersion());
|
||||
|
||||
peer = setupPeer("127.1.0.1", "3.0.0");
|
||||
limit.updateMaxSupportedVersion();
|
||||
assertEquals(ProtocolVersion.MAX_SUPPORTED_VERSION, limit.getMaxVersion());
|
||||
|
||||
// learn that peer doesn't actually fully support V4, behaviour should remain the same
|
||||
updatePeerInfo(peer, "2.2.0");
|
||||
limit.updateMaxSupportedVersion();
|
||||
assertEquals(ProtocolVersion.MAX_SUPPORTED_VERSION, limit.getMaxVersion());
|
||||
|
||||
// finally learn that peer2 has been upgraded, just for completeness
|
||||
updatePeerInfo(peer, "3.3.0");
|
||||
limit.updateMaxSupportedVersion();
|
||||
assertEquals(ProtocolVersion.MAX_SUPPORTED_VERSION, limit.getMaxVersion());
|
||||
|
||||
} finally {
|
||||
cleanupPeers(peer);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -52,7 +52,7 @@ public class ProtocolErrorTest {
|
|||
|
||||
public void testInvalidProtocolVersion(int version) throws Exception
|
||||
{
|
||||
Frame.Decoder dec = new Frame.Decoder(null);
|
||||
Frame.Decoder dec = new Frame.Decoder(null, ProtocolVersionLimit.SERVER_DEFAULT);
|
||||
|
||||
List<Object> results = new ArrayList<>();
|
||||
byte[] frame = new byte[] {
|
||||
|
|
@ -80,7 +80,7 @@ public class ProtocolErrorTest {
|
|||
public void testInvalidProtocolVersionShortFrame() throws Exception
|
||||
{
|
||||
// test for CASSANDRA-11464
|
||||
Frame.Decoder dec = new Frame.Decoder(null);
|
||||
Frame.Decoder dec = new Frame.Decoder(null, ProtocolVersionLimit.SERVER_DEFAULT);
|
||||
|
||||
List<Object> results = new ArrayList<>();
|
||||
byte[] frame = new byte[] {
|
||||
|
|
@ -102,7 +102,7 @@ public class ProtocolErrorTest {
|
|||
@Test
|
||||
public void testInvalidDirection() throws Exception
|
||||
{
|
||||
Frame.Decoder dec = new Frame.Decoder(null);
|
||||
Frame.Decoder dec = new Frame.Decoder(null, ProtocolVersionLimit.SERVER_DEFAULT);
|
||||
|
||||
List<Object> results = new ArrayList<>();
|
||||
// should generate a protocol exception for using a response frame with
|
||||
|
|
@ -133,7 +133,7 @@ public class ProtocolErrorTest {
|
|||
@Test
|
||||
public void testBodyLengthOverLimit() throws Exception
|
||||
{
|
||||
Frame.Decoder dec = new Frame.Decoder(null);
|
||||
Frame.Decoder dec = new Frame.Decoder(null, ProtocolVersionLimit.SERVER_DEFAULT);
|
||||
|
||||
List<Object> results = new ArrayList<>();
|
||||
byte[] frame = new byte[] {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,166 @@
|
|||
/*
|
||||
* 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.transport;
|
||||
|
||||
import java.net.InetAddress;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
|
||||
import com.datastax.driver.core.Cluster;
|
||||
import com.datastax.driver.core.ProtocolVersion;
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.cql3.CQLTester;
|
||||
|
||||
import static org.apache.cassandra.transport.ProtocolTestHelper.cleanupPeers;
|
||||
import static org.apache.cassandra.transport.ProtocolTestHelper.setStaticLimitInConfig;
|
||||
import static org.apache.cassandra.transport.ProtocolTestHelper.setupPeer;
|
||||
import static org.apache.cassandra.transport.ProtocolTestHelper.updatePeerInfo;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
public class ProtocolNegotiationTest extends CQLTester
|
||||
{
|
||||
// to avoid JMX naming clashes between cluster metrics
|
||||
private int clusterId = 0;
|
||||
|
||||
@BeforeClass
|
||||
public static void setup()
|
||||
{
|
||||
prepareNetwork();
|
||||
}
|
||||
|
||||
@Before
|
||||
public void clearConfig()
|
||||
{
|
||||
setStaticLimitInConfig(null);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void serverSupportsV3AndV4ByDefault() throws Throwable
|
||||
{
|
||||
reinitializeNetwork();
|
||||
// client can explicitly request either V3 or V4
|
||||
testConnection(ProtocolVersion.V3, ProtocolVersion.V3);
|
||||
testConnection(ProtocolVersion.V4, ProtocolVersion.V4);
|
||||
|
||||
// if not specified, V4 is the default
|
||||
testConnection(null, ProtocolVersion.V4);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testStaticLimit() throws Throwable
|
||||
{
|
||||
try
|
||||
{
|
||||
reinitializeNetwork();
|
||||
// No limit enforced to start
|
||||
assertEquals(Integer.MIN_VALUE, DatabaseDescriptor.getNativeProtocolMaxVersionOverride());
|
||||
testConnection(null, ProtocolVersion.V4);
|
||||
|
||||
// Update DatabaseDescriptor, then re-initialise the server to force it to read it
|
||||
setStaticLimitInConfig(ProtocolVersion.V3.toInt());
|
||||
reinitializeNetwork();
|
||||
assertEquals(3, DatabaseDescriptor.getNativeProtocolMaxVersionOverride());
|
||||
testConnection(ProtocolVersion.V4, ProtocolVersion.V3);
|
||||
testConnection(ProtocolVersion.V3, ProtocolVersion.V3);
|
||||
testConnection(null, ProtocolVersion.V3);
|
||||
} finally {
|
||||
setStaticLimitInConfig(null);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDynamicLimit() throws Throwable
|
||||
{
|
||||
InetAddress peer1 = setupPeer("127.1.0.1", "2.2.0");
|
||||
InetAddress peer2 = setupPeer("127.1.0.2", "2.2.0");
|
||||
InetAddress peer3 = setupPeer("127.1.0.3", "2.2.0");
|
||||
reinitializeNetwork();
|
||||
try
|
||||
{
|
||||
// legacy peers means max negotiable version is V3
|
||||
testConnection(ProtocolVersion.V4, ProtocolVersion.V3);
|
||||
testConnection(ProtocolVersion.V3, ProtocolVersion.V3);
|
||||
testConnection(null, ProtocolVersion.V3);
|
||||
|
||||
// receive notification that 2 peers have upgraded to a version that fully supports V4
|
||||
updatePeerInfo(peer1, "3.0.0");
|
||||
updatePeerInfo(peer2, "3.0.0");
|
||||
updateMaxNegotiableProtocolVersion();
|
||||
// version should still be capped
|
||||
testConnection(ProtocolVersion.V4, ProtocolVersion.V3);
|
||||
testConnection(ProtocolVersion.V3, ProtocolVersion.V3);
|
||||
testConnection(null, ProtocolVersion.V3);
|
||||
|
||||
// no legacy peers so V4 is negotiable
|
||||
// after the last peer upgrades, cap should be lifted
|
||||
updatePeerInfo(peer3, "3.0.0");
|
||||
updateMaxNegotiableProtocolVersion();
|
||||
testConnection(ProtocolVersion.V4, ProtocolVersion.V4);
|
||||
testConnection(ProtocolVersion.V3, ProtocolVersion.V3);
|
||||
testConnection(null, ProtocolVersion.V4);
|
||||
} finally {
|
||||
cleanupPeers(peer1, peer2, peer3);
|
||||
}
|
||||
}
|
||||
|
||||
private void testConnection(com.datastax.driver.core.ProtocolVersion requestedVersion,
|
||||
com.datastax.driver.core.ProtocolVersion expectedVersion)
|
||||
{
|
||||
long start = System.nanoTime();
|
||||
boolean expectError = requestedVersion != null && requestedVersion != expectedVersion;
|
||||
Cluster.Builder builder = Cluster.builder()
|
||||
.addContactPoints(nativeAddr)
|
||||
.withClusterName("Test Cluster" + clusterId++)
|
||||
.withPort(nativePort);
|
||||
|
||||
if (requestedVersion != null)
|
||||
builder = builder.withProtocolVersion(requestedVersion) ;
|
||||
|
||||
Cluster cluster = builder.build();
|
||||
logger.info("Setting up cluster took {}ms", TimeUnit.MILLISECONDS.convert(System.nanoTime() - start, TimeUnit.NANOSECONDS));
|
||||
start = System.nanoTime();
|
||||
try {
|
||||
cluster.connect();
|
||||
if (expectError)
|
||||
fail("Expected a protocol exception");
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
if (!expectError)
|
||||
{
|
||||
e.printStackTrace();
|
||||
fail("Did not expect any exception");
|
||||
}
|
||||
|
||||
assertTrue(e.getMessage().contains(String.format("Host does not support protocol version %s but %s", requestedVersion, expectedVersion)));
|
||||
} finally {
|
||||
logger.info("Testing connection took {}ms", TimeUnit.MILLISECONDS.convert(System.nanoTime() - start, TimeUnit.NANOSECONDS));
|
||||
start = System.nanoTime();
|
||||
cluster.closeAsync();
|
||||
logger.info("Tearing down cluster connection took {}ms", TimeUnit.MILLISECONDS.convert(System.nanoTime() - start, TimeUnit.NANOSECONDS));
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,95 @@
|
|||
/*
|
||||
* 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.transport;
|
||||
|
||||
import java.lang.reflect.Constructor;
|
||||
import java.lang.reflect.Field;
|
||||
import java.net.InetAddress;
|
||||
import java.net.UnknownHostException;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
|
||||
import com.google.common.util.concurrent.MoreExecutors;
|
||||
|
||||
import org.apache.cassandra.config.Config;
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.db.SystemKeyspace;
|
||||
import org.apache.cassandra.gms.VersionedValue;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
|
||||
public class ProtocolTestHelper
|
||||
{
|
||||
static ExecutorService executor = MoreExecutors.newDirectExecutorService();
|
||||
static InetAddress setupPeer(String address, String version) throws Throwable
|
||||
{
|
||||
InetAddress peer = peer(address);
|
||||
updatePeerInfo(peer, version);
|
||||
return peer;
|
||||
}
|
||||
|
||||
static void updatePeerInfo(InetAddress peer, String version) throws Throwable
|
||||
{
|
||||
SystemKeyspace.updatePeerInfo(peer, "release_version", version, executor);
|
||||
}
|
||||
|
||||
static InetAddress peer(String address)
|
||||
{
|
||||
try
|
||||
{
|
||||
return InetAddress.getByName(address);
|
||||
}
|
||||
catch (UnknownHostException e)
|
||||
{
|
||||
throw new RuntimeException("Error creating peer", e);
|
||||
}
|
||||
}
|
||||
|
||||
static void cleanupPeers(InetAddress...peers) throws Throwable
|
||||
{
|
||||
for (InetAddress peer : peers)
|
||||
if (peer != null)
|
||||
SystemKeyspace.removeEndpoint(peer);
|
||||
}
|
||||
|
||||
static void setStaticLimitInConfig(Integer version)
|
||||
{
|
||||
try
|
||||
{
|
||||
Field field = FBUtilities.getProtectedField(DatabaseDescriptor.class, "conf");
|
||||
((Config)field.get(null)).native_transport_max_negotiable_protocol_version = version == null ? Integer.MIN_VALUE : version;
|
||||
}
|
||||
catch (IllegalAccessException e)
|
||||
{
|
||||
throw new RuntimeException("Error setting native_transport_max_protocol_version on Config", e);
|
||||
}
|
||||
}
|
||||
|
||||
static VersionedValue releaseVersion(String versionString)
|
||||
{
|
||||
try
|
||||
{
|
||||
Constructor<VersionedValue> ctor = VersionedValue.class.getDeclaredConstructor(String.class);
|
||||
ctor.setAccessible(true);
|
||||
return ctor.newInstance(versionString);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
throw new RuntimeException("Error constructing VersionedValue for release version", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -27,13 +27,13 @@ public class ProtocolVersionTest
|
|||
public void testDecode()
|
||||
{
|
||||
for (ProtocolVersion version : ProtocolVersion.SUPPORTED)
|
||||
Assert.assertEquals(version, ProtocolVersion.decode(version.asInt()));
|
||||
Assert.assertEquals(version, ProtocolVersion.decode(version.asInt(), ProtocolVersionLimit.SERVER_DEFAULT));
|
||||
|
||||
for (ProtocolVersion version : ProtocolVersion.UNSUPPORTED)
|
||||
{ // unsupported old versions
|
||||
try
|
||||
{
|
||||
Assert.assertEquals(version, ProtocolVersion.decode(version.asInt()));
|
||||
Assert.assertEquals(version, ProtocolVersion.decode(version.asInt(), ProtocolVersionLimit.SERVER_DEFAULT));
|
||||
Assert.fail("Expected invalid protocol exception");
|
||||
}
|
||||
catch (ProtocolException ex)
|
||||
|
|
@ -45,7 +45,7 @@ public class ProtocolVersionTest
|
|||
|
||||
try
|
||||
{ // unsupported newer version
|
||||
Assert.assertEquals(null, ProtocolVersion.decode(63));
|
||||
Assert.assertEquals(null, ProtocolVersion.decode(63, ProtocolVersionLimit.SERVER_DEFAULT));
|
||||
Assert.fail("Expected invalid protocol exception");
|
||||
}
|
||||
catch (ProtocolException ex)
|
||||
|
|
|
|||
Loading…
Reference in New Issue