mirror of https://github.com/apache/cassandra
Merge e380ef1f8d into 10557d7ffe
This commit is contained in:
vivekkoya
GitHub
commit
0e91255687
|
|
@ -1085,7 +1085,6 @@ public class OutboundConnection
|
|||
*
|
||||
* Most of the actual work is performed by OutboundConnectionInitiator, this method just manages
|
||||
* our book keeping on either success or failure.
|
||||
*
|
||||
* This method is only to be invoked by the eventLoop, and the inner class' methods should only be evaluated by the eventtLoop
|
||||
*/
|
||||
Future<?> initiate()
|
||||
|
|
@ -1137,42 +1136,45 @@ public class OutboundConnection
|
|||
// it is expected that close, if successful, has already cancelled us; so we do not need to worry about leaking connections
|
||||
assert !state.isClosed();
|
||||
|
||||
MessagingSuccess success = result.success();
|
||||
debug.onConnect(success.messagingVersion, settings);
|
||||
state.disconnected().maintenance.cancel(false);
|
||||
if (result.success() instanceof MessagingSuccess success)
|
||||
{
|
||||
debug.onConnect(success.messagingVersion, settings);
|
||||
state.disconnected().maintenance.cancel(false);
|
||||
|
||||
FrameEncoder.PayloadAllocator payloadAllocator = success.allocator;
|
||||
Channel channel = success.channel;
|
||||
Established established = new Established(success.messagingVersion, channel, payloadAllocator, settings);
|
||||
state = established;
|
||||
channel.pipeline().addLast("handleExceptionalStates", new ChannelInboundHandlerAdapter() {
|
||||
@Override
|
||||
public void channelInactive(ChannelHandlerContext ctx)
|
||||
FrameEncoder.PayloadAllocator payloadAllocator = success.allocator;
|
||||
Channel channel = success.channel;
|
||||
Established established = new Established(success.messagingVersion, channel, payloadAllocator, settings);
|
||||
state = established;
|
||||
channel.pipeline().addLast("handleExceptionalStates", new ChannelInboundHandlerAdapter()
|
||||
{
|
||||
disconnectNow(established);
|
||||
ctx.fireChannelInactive();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
|
||||
{
|
||||
try
|
||||
@Override
|
||||
public void channelInactive(ChannelHandlerContext ctx)
|
||||
{
|
||||
invalidateChannel(established, cause);
|
||||
disconnectNow(established);
|
||||
ctx.fireChannelInactive();
|
||||
}
|
||||
catch (Throwable t)
|
||||
{
|
||||
logger.error("Unexpected exception in {}.exceptionCaught", this.getClass().getSimpleName(), t);
|
||||
}
|
||||
}
|
||||
});
|
||||
++successfulConnections;
|
||||
|
||||
logger.info("{} successfully connected, version = {}, framing = {}, encryption = {}",
|
||||
id(true),
|
||||
success.messagingVersion,
|
||||
settings.framing,
|
||||
encryptionConnectionSummary(channel));
|
||||
@Override
|
||||
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
|
||||
{
|
||||
try
|
||||
{
|
||||
invalidateChannel(established, cause);
|
||||
}
|
||||
catch (Throwable t)
|
||||
{
|
||||
logger.error("Unexpected exception in {}.exceptionCaught", this.getClass().getSimpleName(), t);
|
||||
}
|
||||
}
|
||||
});
|
||||
++successfulConnections;
|
||||
|
||||
logger.info("{} successfully connected, version = {}, framing = {}, encryption = {}",
|
||||
id(true),
|
||||
success.messagingVersion,
|
||||
settings.framing,
|
||||
encryptionConnectionSummary(channel));
|
||||
}
|
||||
break;
|
||||
|
||||
case RETRY:
|
||||
|
|
|
|||
|
|
@ -528,14 +528,29 @@ public class OutboundConnectionInitiator<SuccessType extends OutboundConnectionI
|
|||
}
|
||||
|
||||
boolean isSuccess() { return outcome == Outcome.SUCCESS; }
|
||||
public SuccessType success() { return (SuccessType) this; }
|
||||
public Success success()
|
||||
{
|
||||
if (this instanceof Success success)
|
||||
return success;
|
||||
return null;
|
||||
}
|
||||
static MessagingSuccess messagingSuccess(Channel channel, int messagingVersion, FrameEncoder.PayloadAllocator allocator) { return new MessagingSuccess(channel, messagingVersion, allocator); }
|
||||
static StreamingSuccess streamingSuccess(Channel channel, int messagingVersion) { return new StreamingSuccess(channel, messagingVersion); }
|
||||
|
||||
public Retry retry() { return (Retry) this; }
|
||||
public Retry retry()
|
||||
{
|
||||
if (this instanceof Retry retry)
|
||||
return retry;
|
||||
return null;
|
||||
}
|
||||
static <SuccessType extends Success> Result<SuccessType> retry(int withMessagingVersion) { return new Retry<>(withMessagingVersion); }
|
||||
|
||||
public Incompatible incompatible() { return (Incompatible) this; }
|
||||
public Incompatible incompatible()
|
||||
{
|
||||
if (this instanceof Incompatible incompatible)
|
||||
return incompatible;
|
||||
return null;
|
||||
}
|
||||
static <SuccessType extends Success> Result<SuccessType> incompatible(int closestSupportedVersion, int maxMessagingVersion) { return new Incompatible(closestSupportedVersion, maxMessagingVersion); }
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,201 @@
|
|||
/*
|
||||
* 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.net;
|
||||
|
||||
import java.nio.channels.ClosedChannelException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import com.google.common.net.InetAddresses;
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Before;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
|
||||
import io.netty.channel.EventLoop;
|
||||
import io.netty.util.concurrent.Future;
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions;
|
||||
import org.apache.cassandra.config.ParameterizedClass;
|
||||
import org.apache.cassandra.db.commitlog.CommitLog;
|
||||
import org.apache.cassandra.gms.GossipDigestSyn;
|
||||
import org.apache.cassandra.locator.InetAddressAndPort;
|
||||
import org.apache.cassandra.security.DefaultSslContextFactory;
|
||||
import static org.apache.cassandra.net.OutboundConnectionInitiator.Result;
|
||||
import static org.apache.cassandra.net.OutboundConnectionInitiator.SslFallbackConnectionType;
|
||||
import static org.apache.cassandra.net.OutboundConnectionInitiator.initiateStreaming;
|
||||
import static org.apache.cassandra.net.MessagingService.current_version;
|
||||
import static org.apache.cassandra.net.MessagingService.minimum_version;
|
||||
|
||||
public class StreamingTest
|
||||
{
|
||||
private static final SocketFactory factory = new SocketFactory();
|
||||
static final InetAddressAndPort TO_ADDR = InetAddressAndPort.getByAddressOverrideDefaults(InetAddresses.forString("127.0.0.2"), 7012);
|
||||
static final InetAddressAndPort FROM_ADDR = InetAddressAndPort.getByAddressOverrideDefaults(InetAddresses.forString("127.0.0.1"), 7012);
|
||||
private volatile Throwable handshakeEx;
|
||||
@BeforeClass
|
||||
public static void startup()
|
||||
{
|
||||
DatabaseDescriptor.daemonInitialization();
|
||||
CommitLog.instance.start();
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
public static void cleanup() throws InterruptedException
|
||||
{
|
||||
factory.shutdownNow();
|
||||
}
|
||||
|
||||
@Before
|
||||
public void setup()
|
||||
{
|
||||
handshakeEx = null;
|
||||
}
|
||||
|
||||
private Future<Result<Result.StreamingSuccess>> streamingConnect(AcceptVersions acceptOutbound, AcceptVersions acceptInbound) throws ExecutionException, InterruptedException
|
||||
{
|
||||
InboundSockets inbound = new InboundSockets(new InboundConnectionSettings().withAcceptMessaging(acceptInbound));
|
||||
try
|
||||
{
|
||||
inbound.open();
|
||||
InetAddressAndPort endpoint = inbound.sockets().stream().map(s -> s.settings.bindAddress).findFirst().get();
|
||||
EventLoop eventLoop = factory.defaultGroup().next();
|
||||
Future<Result<Result.StreamingSuccess>> result = initiateStreaming(eventLoop,
|
||||
new OutboundConnectionSettings(endpoint)
|
||||
.withAcceptVersions(acceptOutbound)
|
||||
.withDefaults(ConnectionCategory.STREAMING),
|
||||
SslFallbackConnectionType.SERVER_CONFIG
|
||||
);
|
||||
result.awaitUninterruptibly();
|
||||
return result;
|
||||
}
|
||||
finally
|
||||
{
|
||||
inbound.close().await(1L, TimeUnit.SECONDS);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testIncompatibleVersion() throws InterruptedException, ExecutionException
|
||||
{
|
||||
Future<Result<Result.StreamingSuccess>> result = streamingConnect(new AcceptVersions(current_version + 1, current_version + 1), new AcceptVersions(minimum_version + 2, current_version + 3));
|
||||
if (result.isSuccess()) {
|
||||
Result<Result.StreamingSuccess> nowResult = result.getNow();
|
||||
Assert.assertNull(nowResult.success());
|
||||
Assert.assertEquals(Result.Outcome.INCOMPATIBLE, nowResult.outcome);
|
||||
Assert.assertEquals(current_version, nowResult.incompatible().closestSupportedVersion);
|
||||
Assert.assertEquals(current_version, nowResult.incompatible().maxMessagingVersion);
|
||||
|
||||
} else {
|
||||
Assert.assertTrue(false);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCompatibleVersion() throws InterruptedException, ExecutionException
|
||||
{
|
||||
Future<Result<Result.StreamingSuccess>> result = streamingConnect(new AcceptVersions(MessagingService.minimum_version, current_version + 1), new AcceptVersions(minimum_version + 2, current_version + 3));
|
||||
if (result.isSuccess()) {
|
||||
Result<Result.StreamingSuccess> nowResult = result.getNow();
|
||||
Assert.assertNotNull(nowResult.success().channel);
|
||||
Assert.assertEquals(Result.Outcome.SUCCESS, nowResult.outcome);
|
||||
Assert.assertEquals(current_version, nowResult.success().messagingVersion);
|
||||
} else {
|
||||
Assert.assertTrue(false);
|
||||
}
|
||||
}
|
||||
|
||||
private ServerEncryptionOptions getServerEncryptionOptions(SslFallbackConnectionType sslConnectionType, boolean optional)
|
||||
{
|
||||
ServerEncryptionOptions serverEncryptionOptions = new ServerEncryptionOptions().withOptional(optional)
|
||||
.withKeyStore("test/conf/cassandra_ssl_test.keystore")
|
||||
.withKeyStorePassword("cassandra")
|
||||
.withOutboundKeystore("test/conf/cassandra_ssl_test_outbound.keystore")
|
||||
.withOutboundKeystorePassword("cassandra")
|
||||
.withTrustStore("test/conf/cassandra_ssl_test.truststore")
|
||||
.withTrustStorePassword("cassandra")
|
||||
.withSslContextFactory((new ParameterizedClass(DefaultSslContextFactory.class.getName(),
|
||||
new HashMap<>())));
|
||||
if (sslConnectionType == SslFallbackConnectionType.MTLS)
|
||||
{
|
||||
serverEncryptionOptions = serverEncryptionOptions.withInternodeEncryption(ServerEncryptionOptions.InternodeEncryption.all)
|
||||
.withRequireClientAuth(true);
|
||||
}
|
||||
else if (sslConnectionType == SslFallbackConnectionType.SSL)
|
||||
{
|
||||
serverEncryptionOptions = serverEncryptionOptions.withInternodeEncryption(ServerEncryptionOptions.InternodeEncryption.all)
|
||||
.withRequireClientAuth(false);
|
||||
}
|
||||
return serverEncryptionOptions;
|
||||
}
|
||||
|
||||
private OutboundConnection initiateOutbound(InetAddressAndPort endpoint, SslFallbackConnectionType connectionType, boolean optional) throws ClosedChannelException
|
||||
{
|
||||
final OutboundConnectionSettings settings = new OutboundConnectionSettings(endpoint)
|
||||
.withAcceptVersions(new AcceptVersions(minimum_version, current_version))
|
||||
.withDefaults(ConnectionCategory.MESSAGING)
|
||||
.withEncryption(getServerEncryptionOptions(connectionType, optional))
|
||||
.withDebugCallbacks(new HandshakeAcknowledgeChecker(t -> handshakeEx = t))
|
||||
.withFrom(FROM_ADDR);
|
||||
OutboundConnections outboundConnections = OutboundConnections.tryRegister(new ConcurrentHashMap<>(), TO_ADDR, settings);
|
||||
GossipDigestSyn syn = new GossipDigestSyn("cluster", "partitioner", new ArrayList<>(0));
|
||||
Message<GossipDigestSyn> message = Message.out(Verb.GOSSIP_DIGEST_SYN, syn);
|
||||
OutboundConnection outboundConnection = outboundConnections.connectionFor(message);
|
||||
outboundConnection.enqueue(message);
|
||||
return outboundConnection;
|
||||
}
|
||||
private static class HandshakeAcknowledgeChecker implements OutboundDebugCallbacks
|
||||
{
|
||||
private final AtomicInteger acks = new AtomicInteger(0);
|
||||
private final Consumer<Throwable> fail;
|
||||
|
||||
private HandshakeAcknowledgeChecker(Consumer<Throwable> fail)
|
||||
{
|
||||
this.fail = fail;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onSendSmallFrame(int messageCount, int payloadSizeInBytes)
|
||||
{
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onSentSmallFrame(int messageCount, int payloadSizeInBytes)
|
||||
{
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onFailedSmallFrame(int messageCount, int payloadSizeInBytes)
|
||||
{
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onConnect(int messagingVersion, OutboundConnectionSettings settings)
|
||||
{
|
||||
if (acks.incrementAndGet() > 1)
|
||||
fail.accept(new AssertionError("Handshake was acknowledged more than once"));
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue