cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jasobr...@apache.org
Subject [06/11] cassandra git commit: switch internode messaging to netty
Date Tue, 22 Aug 2017 20:55:13 GMT
http://git-wip-us.apache.org/repos/asf/cassandra/blob/356dc3c2/test/unit/org/apache/cassandra/net/async/ChannelWriterTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/net/async/ChannelWriterTest.java b/test/unit/org/apache/cassandra/net/async/ChannelWriterTest.java
new file mode 100644
index 0000000..128fe4b
--- /dev/null
+++ b/test/unit/org/apache/cassandra/net/async/ChannelWriterTest.java
@@ -0,0 +1,312 @@
+/*
+ * 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.async;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Optional;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.WriteBufferWaterMark;
+import io.netty.channel.embedded.EmbeddedChannel;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.net.MessageOut;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.async.ChannelWriter.CoalescingChannelWriter;
+import org.apache.cassandra.utils.CoalescingStrategies;
+import org.apache.cassandra.utils.CoalescingStrategies.CoalescingStrategy;
+
+import static org.apache.cassandra.net.MessagingService.Verb.ECHO;
+
+/**
+ * with the write_Coalescing_* methods, if there's data in the channel.unsafe().outboundBuffer()
+ * it means that there's something in the channel that hasn't yet been flushed to the transport (socket).
+ * once a flush occurs, there will be an entry in EmbeddedChannel's outboundQueue. those two facts are leveraged in these tests.
+ */
+public class ChannelWriterTest
+{
+    private static final int COALESCE_WINDOW_MS = 10;
+
+    private EmbeddedChannel channel;
+    private ChannelWriter channelWriter;
+    private NonSendingOutboundMessagingConnection omc;
+    private Optional<CoalescingStrategy> coalescingStrategy;
+
+    @BeforeClass
+    public static void before()
+    {
+        DatabaseDescriptor.daemonInitialization();
+    }
+
+    @Before
+    public void setup()
+    {
+        OutboundConnectionIdentifier id = OutboundConnectionIdentifier.small(new InetSocketAddress("127.0.0.1", 0),
+                                                                             new InetSocketAddress("127.0.0.2", 0));
+        channel = new EmbeddedChannel();
+        omc = new NonSendingOutboundMessagingConnection(id, null, Optional.empty());
+        channelWriter = ChannelWriter.create(channel, omc::handleMessageResult, Optional.empty());
+        channel.pipeline().addFirst(new MessageOutHandler(id, MessagingService.current_version, channelWriter, () -> null));
+        coalescingStrategy = CoalescingStrategies.newCoalescingStrategy(CoalescingStrategies.Strategy.FIXED.name(), COALESCE_WINDOW_MS, null, "test");
+    }
+
+    @Test
+    public void create_nonCoalescing()
+    {
+        Assert.assertSame(ChannelWriter.SimpleChannelWriter.class, ChannelWriter.create(channel, omc::handleMessageResult, Optional.empty()).getClass());
+    }
+
+    @Test
+    public void create_Coalescing()
+    {
+        Assert.assertSame(CoalescingChannelWriter.class, ChannelWriter.create(channel, omc::handleMessageResult, coalescingStrategy).getClass());
+    }
+
+    @Test
+    public void write_IsWritable()
+    {
+        Assert.assertTrue(channel.isWritable());
+        Assert.assertTrue(channelWriter.write(new QueuedMessage(new MessageOut<>(ECHO), 42), true));
+        Assert.assertTrue(channel.isWritable());
+        Assert.assertTrue(channel.releaseOutbound());
+    }
+
+    @Test
+    public void write_NotWritable()
+    {
+        channel.config().setOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(1, 2));
+
+        // send one message through, which will trigger the writability check (and turn it off)
+        Assert.assertTrue(channel.isWritable());
+        ByteBuf buf = channel.alloc().buffer(8, 8);
+        channel.unsafe().outboundBuffer().addMessage(buf, buf.capacity(), channel.newPromise());
+        Assert.assertFalse(channel.isWritable());
+        Assert.assertFalse(channelWriter.write(new QueuedMessage(new MessageOut<>(ECHO), 42), true));
+        Assert.assertFalse(channel.isWritable());
+        Assert.assertFalse(channel.releaseOutbound());
+        buf.release();
+    }
+
+    @Test
+    public void write_NotWritableButWriteAnyway()
+    {
+        channel.config().setOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(1, 2));
+
+        // send one message through, which will trigger the writability check (and turn it off)
+        Assert.assertTrue(channel.isWritable());
+        ByteBuf buf = channel.alloc().buffer(8, 8);
+        channel.unsafe().outboundBuffer().addMessage(buf, buf.capacity(), channel.newPromise());
+        Assert.assertFalse(channel.isWritable());
+        Assert.assertTrue(channelWriter.write(new QueuedMessage(new MessageOut<>(ECHO), 42), false));
+        Assert.assertTrue(channel.isWritable());
+        Assert.assertTrue(channel.releaseOutbound());
+    }
+
+    @Test
+    public void write_Coalescing_LostRaceForFlushTask()
+    {
+        CoalescingChannelWriter channelWriter = resetEnvForCoalescing(DatabaseDescriptor.getOtcCoalescingEnoughCoalescedMessages());
+        channelWriter.scheduledFlush.set(true);
+        Assert.assertTrue(channel.unsafe().outboundBuffer().totalPendingWriteBytes() == 0);
+        Assert.assertTrue(channelWriter.write(new QueuedMessage(new MessageOut<>(ECHO), 42), true));
+        Assert.assertTrue(channel.unsafe().outboundBuffer().totalPendingWriteBytes() > 0);
+        Assert.assertFalse(channel.releaseOutbound());
+        Assert.assertTrue(channelWriter.scheduledFlush.get());
+    }
+
+    @Test
+    public void write_Coalescing_HitMinMessageCountForImmediateCoalesce()
+    {
+        CoalescingChannelWriter channelWriter = resetEnvForCoalescing(1);
+
+        Assert.assertTrue(channel.unsafe().outboundBuffer().totalPendingWriteBytes() == 0);
+        Assert.assertFalse(channelWriter.scheduledFlush.get());
+        Assert.assertTrue(channelWriter.write(new QueuedMessage(new MessageOut<>(ECHO), 42), true));
+
+        Assert.assertTrue(channel.unsafe().outboundBuffer().totalPendingWriteBytes() == 0);
+        Assert.assertTrue(channel.releaseOutbound());
+        Assert.assertFalse(channelWriter.scheduledFlush.get());
+    }
+
+    @Test
+    public void write_Coalescing_ScheduleFlushTask()
+    {
+        CoalescingChannelWriter channelWriter = resetEnvForCoalescing(DatabaseDescriptor.getOtcCoalescingEnoughCoalescedMessages());
+
+        Assert.assertTrue(channel.unsafe().outboundBuffer().totalPendingWriteBytes() == 0);
+        Assert.assertFalse(channelWriter.scheduledFlush.get());
+        Assert.assertTrue(channelWriter.write(new QueuedMessage(new MessageOut<>(ECHO), 42), true));
+
+        Assert.assertTrue(channelWriter.scheduledFlush.get());
+        Assert.assertTrue(channel.unsafe().outboundBuffer().totalPendingWriteBytes() > 0);
+        Assert.assertTrue(channelWriter.scheduledFlush.get());
+
+        // this unfortunately know a little too much about how the sausage is made in CoalescingChannelWriter :-/
+        channel.runScheduledPendingTasks();
+        channel.runPendingTasks();
+        Assert.assertTrue(channel.unsafe().outboundBuffer().totalPendingWriteBytes() == 0);
+        Assert.assertFalse(channelWriter.scheduledFlush.get());
+        Assert.assertTrue(channel.releaseOutbound());
+    }
+
+    private CoalescingChannelWriter resetEnvForCoalescing(int minMessagesForCoalesce)
+    {
+        channel = new EmbeddedChannel();
+        CoalescingChannelWriter cw = new CoalescingChannelWriter(channel, omc::handleMessageResult, coalescingStrategy.get(), minMessagesForCoalesce);
+        channel.pipeline().addFirst(new ChannelOutboundHandlerAdapter()
+        {
+            public void flush(ChannelHandlerContext ctx) throws Exception
+            {
+                cw.onTriggeredFlush(ctx);
+            }
+        });
+        omc.setChannelWriter(cw);
+        return cw;
+    }
+
+    @Test
+    public void writeBacklog_Empty()
+    {
+        BlockingQueue<QueuedMessage> queue = new LinkedBlockingQueue<>();
+        Assert.assertEquals(0, channelWriter.writeBacklog(queue, false));
+        Assert.assertFalse(channel.releaseOutbound());
+    }
+
+    @Test
+    public void writeBacklog_ChannelNotWritable()
+    {
+        Assert.assertTrue(channel.isWritable());
+        // force the channel to be non writable
+        channel.config().setOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(1, 2));
+        ByteBuf buf = channel.alloc().buffer(8, 8);
+        channel.unsafe().outboundBuffer().addMessage(buf, buf.capacity(), channel.newPromise());
+        Assert.assertFalse(channel.isWritable());
+
+        Assert.assertEquals(0, channelWriter.writeBacklog(new LinkedBlockingQueue<>(), false));
+        Assert.assertFalse(channel.releaseOutbound());
+        Assert.assertFalse(channel.isWritable());
+        buf.release();
+    }
+
+    @Test
+    public void writeBacklog_NotEmpty()
+    {
+        BlockingQueue<QueuedMessage> queue = new LinkedBlockingQueue<>();
+        int count = 12;
+        for (int i = 0; i < count; i++)
+            queue.offer(new QueuedMessage(new MessageOut<>(ECHO), i));
+        Assert.assertEquals(count, channelWriter.writeBacklog(queue, false));
+        Assert.assertTrue(channel.releaseOutbound());
+    }
+
+    @Test
+    public void close()
+    {
+        Assert.assertFalse(channelWriter.isClosed());
+        Assert.assertTrue(channel.isOpen());
+        channelWriter.close();
+        Assert.assertFalse(channel.isOpen());
+        Assert.assertTrue(channelWriter.isClosed());
+    }
+
+    @Test
+    public void softClose()
+    {
+        Assert.assertFalse(channelWriter.isClosed());
+        Assert.assertTrue(channel.isOpen());
+        channelWriter.softClose();
+        Assert.assertFalse(channel.isOpen());
+        Assert.assertTrue(channelWriter.isClosed());
+    }
+
+    @Test
+    public void handleMessagePromise_FutureIsCancelled()
+    {
+        ChannelPromise promise = channel.newPromise();
+        promise.cancel(false);
+        channelWriter.handleMessageFuture(promise, new QueuedMessage(new MessageOut<>(ECHO), 1), true);
+        Assert.assertTrue(channel.isActive());
+        Assert.assertEquals(1, omc.getCompletedMessages().longValue());
+        Assert.assertEquals(0, omc.getDroppedMessages().longValue());
+    }
+
+    @Test
+    public void handleMessagePromise_ExpiredException_DoNotRetryMsg()
+    {
+        ChannelPromise promise = channel.newPromise();
+        promise.setFailure(new ExpiredException());
+
+        channelWriter.handleMessageFuture(promise, new QueuedMessage(new MessageOut<>(ECHO), 1), true);
+        Assert.assertTrue(channel.isActive());
+        Assert.assertEquals(1, omc.getCompletedMessages().longValue());
+        Assert.assertEquals(1, omc.getDroppedMessages().longValue());
+        Assert.assertFalse(omc.sendMessageInvoked);
+    }
+
+    @Test
+    public void handleMessagePromise_NonIOException()
+    {
+        ChannelPromise promise = channel.newPromise();
+        promise.setFailure(new NullPointerException("this is a test"));
+        channelWriter.handleMessageFuture(promise, new QueuedMessage(new MessageOut<>(ECHO), 1), true);
+        Assert.assertTrue(channel.isActive());
+        Assert.assertEquals(1, omc.getCompletedMessages().longValue());
+        Assert.assertEquals(0, omc.getDroppedMessages().longValue());
+        Assert.assertFalse(omc.sendMessageInvoked);
+    }
+
+    @Test
+    public void handleMessagePromise_IOException_ChannelNotClosed_RetryMsg()
+    {
+        ChannelPromise promise = channel.newPromise();
+        promise.setFailure(new IOException("this is a test"));
+        Assert.assertTrue(channel.isActive());
+        channelWriter.handleMessageFuture(promise, new QueuedMessage(new MessageOut<>(ECHO), 1, 0, true, true), true);
+
+        Assert.assertFalse(channel.isActive());
+        Assert.assertEquals(1, omc.getCompletedMessages().longValue());
+        Assert.assertEquals(0, omc.getDroppedMessages().longValue());
+        Assert.assertTrue(omc.sendMessageInvoked);
+    }
+
+    @Test
+    public void handleMessagePromise_Cancelled()
+    {
+        ChannelPromise promise = channel.newPromise();
+        promise.cancel(false);
+        Assert.assertTrue(channel.isActive());
+        channelWriter.handleMessageFuture(promise, new QueuedMessage(new MessageOut<>(ECHO), 1, 0, true, true), true);
+
+        Assert.assertTrue(channel.isActive());
+        Assert.assertEquals(1, omc.getCompletedMessages().longValue());
+        Assert.assertEquals(0, omc.getDroppedMessages().longValue());
+        Assert.assertFalse(omc.sendMessageInvoked);
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/356dc3c2/test/unit/org/apache/cassandra/net/async/HandshakeHandlersTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/net/async/HandshakeHandlersTest.java b/test/unit/org/apache/cassandra/net/async/HandshakeHandlersTest.java
new file mode 100644
index 0000000..fa6e2b5
--- /dev/null
+++ b/test/unit/org/apache/cassandra/net/async/HandshakeHandlersTest.java
@@ -0,0 +1,204 @@
+/*
+ * 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.async;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.util.Optional;
+
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import io.netty.channel.embedded.EmbeddedChannel;
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.auth.AllowAllInternodeAuthenticator;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.RowUpdateBuilder;
+import org.apache.cassandra.db.compaction.CompactionManager;
+import org.apache.cassandra.db.marshal.AsciiType;
+import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.net.MessageOut;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.schema.KeyspaceParams;
+
+import static org.apache.cassandra.net.async.InboundHandshakeHandler.State.MESSAGING_HANDSHAKE_COMPLETE;
+import static org.apache.cassandra.net.async.OutboundMessagingConnection.State.READY;
+
+public class HandshakeHandlersTest
+{
+    private static final String KEYSPACE1 = "NettyPipilineTest";
+    private static final String STANDARD1 = "Standard1";
+
+    private static final InetSocketAddress LOCAL_ADDR = new InetSocketAddress("127.0.0.1", 9999);
+    private static final InetSocketAddress REMOTE_ADDR = new InetSocketAddress("127.0.0.2", 9999);
+    private static final int MESSAGING_VERSION = MessagingService.current_version;
+    private static final OutboundConnectionIdentifier connectionId = OutboundConnectionIdentifier.small(LOCAL_ADDR, REMOTE_ADDR);
+
+    @BeforeClass
+    public static void beforeClass() throws ConfigurationException
+    {
+        SchemaLoader.prepareServer();
+        SchemaLoader.createKeyspace(KEYSPACE1,
+                                    KeyspaceParams.simple(1),
+                                    SchemaLoader.standardCFMD(KEYSPACE1, STANDARD1, 0, AsciiType.instance, BytesType.instance));
+        CompactionManager.instance.disableAutoCompaction();
+    }
+
+    @Test
+    public void handshake_HappyPath()
+    {
+        // beacuse both CHH & SHH are ChannelInboundHandlers, we can't use the same EmbeddedChannel to handle them
+        InboundHandshakeHandler inboundHandshakeHandler = new InboundHandshakeHandler(new TestAuthenticator(true));
+        EmbeddedChannel inboundChannel = new EmbeddedChannel(inboundHandshakeHandler);
+
+        OutboundMessagingConnection imc = new OutboundMessagingConnection(connectionId, null, Optional.empty(), new AllowAllInternodeAuthenticator());
+        OutboundConnectionParams params = OutboundConnectionParams.builder()
+                                                                  .connectionId(connectionId)
+                                                                  .callback(imc::finishHandshake)
+                                                                  .mode(NettyFactory.Mode.MESSAGING)
+                                                                  .protocolVersion(MessagingService.current_version)
+                                                                  .coalescingStrategy(Optional.empty())
+                                                                  .build();
+        OutboundHandshakeHandler outboundHandshakeHandler = new OutboundHandshakeHandler(params);
+        EmbeddedChannel outboundChannel = new EmbeddedChannel(outboundHandshakeHandler);
+        Assert.assertEquals(1, outboundChannel.outboundMessages().size());
+
+        // move internode protocol Msg1 to the server's channel
+        Object o;
+        while ((o = outboundChannel.readOutbound()) != null)
+            inboundChannel.writeInbound(o);
+            Assert.assertEquals(1, inboundChannel.outboundMessages().size());
+
+        // move internode protocol Msg2 to the client's channel
+        while ((o = inboundChannel.readOutbound()) != null)
+            outboundChannel.writeInbound(o);
+        Assert.assertEquals(1, outboundChannel.outboundMessages().size());
+
+        // move internode protocol Msg3 to the server's channel
+        while ((o = outboundChannel.readOutbound()) != null)
+            inboundChannel.writeInbound(o);
+
+        Assert.assertEquals(READY, imc.getState());
+        Assert.assertEquals(MESSAGING_HANDSHAKE_COMPLETE, inboundHandshakeHandler.getState());
+    }
+
+    @Test
+    public void lotsOfMutations_NoCompression() throws IOException
+    {
+        lotsOfMutations(false);
+    }
+
+    @Test
+    public void lotsOfMutations_WithCompression() throws IOException
+    {
+        lotsOfMutations(true);
+    }
+
+    private void lotsOfMutations(boolean compress)
+    {
+        TestChannels channels = buildChannels(compress);
+        EmbeddedChannel outboundChannel = channels.outboundChannel;
+        EmbeddedChannel inboundChannel = channels.inboundChannel;
+
+        // now the actual test!
+        ByteBuffer buf = ByteBuffer.allocate(1 << 10);
+        byte[] bytes = "ThisIsA16CharStr".getBytes();
+        while (buf.remaining() > 0)
+            buf.put(bytes);
+
+        // write a bunch of messages to the channel
+        ColumnFamilyStore cfs1 = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1);
+        int count = 1024;
+        for (int i = 0; i < count; i++)
+        {
+            if (i % 2 == 0)
+            {
+                Mutation mutation = new RowUpdateBuilder(cfs1.metadata.get(), 0, "k")
+                                    .clustering("bytes")
+                                    .add("val", buf)
+                                    .build();
+
+                QueuedMessage msg = new QueuedMessage(mutation.createMessage(), i);
+                outboundChannel.writeAndFlush(msg);
+            }
+            else
+            {
+                outboundChannel.writeAndFlush(new QueuedMessage(new MessageOut<>(MessagingService.Verb.ECHO), i));
+            }
+        }
+        outboundChannel.flush();
+
+        // move the messages to the other channel
+        Object o;
+        while ((o = outboundChannel.readOutbound()) != null)
+            inboundChannel.writeInbound(o);
+
+        Assert.assertTrue(outboundChannel.outboundMessages().isEmpty());
+        Assert.assertFalse(inboundChannel.finishAndReleaseAll());
+    }
+
+    private TestChannels buildChannels(boolean compress)
+    {
+        OutboundConnectionParams params = OutboundConnectionParams.builder()
+                                                                  .connectionId(connectionId)
+                                                                  .callback(this::nop)
+                                                                  .mode(NettyFactory.Mode.MESSAGING)
+                                                                  .compress(compress)
+                                                                  .coalescingStrategy(Optional.empty())
+                                                                  .protocolVersion(MessagingService.current_version)
+                                                                  .build();
+        OutboundHandshakeHandler outboundHandshakeHandler = new OutboundHandshakeHandler(params);
+        EmbeddedChannel outboundChannel = new EmbeddedChannel(outboundHandshakeHandler);
+        OutboundMessagingConnection omc = new OutboundMessagingConnection(connectionId, null, Optional.empty(), new AllowAllInternodeAuthenticator());
+        omc.setTargetVersion(MESSAGING_VERSION);
+        outboundHandshakeHandler.setupPipeline(outboundChannel, MESSAGING_VERSION);
+
+        // remove the outbound handshake message from the outbound messages
+        outboundChannel.outboundMessages().clear();
+
+        InboundHandshakeHandler handler = new InboundHandshakeHandler(new TestAuthenticator(true));
+        EmbeddedChannel inboundChannel = new EmbeddedChannel(handler);
+        handler.setupMessagingPipeline(inboundChannel.pipeline(), REMOTE_ADDR.getAddress(), compress, MESSAGING_VERSION);
+
+        return new TestChannels(outboundChannel, inboundChannel);
+    }
+
+    private static class TestChannels
+    {
+        final EmbeddedChannel outboundChannel;
+        final EmbeddedChannel inboundChannel;
+
+        TestChannels(EmbeddedChannel outboundChannel, EmbeddedChannel inboundChannel)
+        {
+            this.outboundChannel = outboundChannel;
+            this.inboundChannel = inboundChannel;
+        }
+    }
+
+    private Void nop(OutboundHandshakeHandler.HandshakeResult handshakeResult)
+    {
+        // do nothing, really
+        return null;
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/356dc3c2/test/unit/org/apache/cassandra/net/async/HandshakeProtocolTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/net/async/HandshakeProtocolTest.java b/test/unit/org/apache/cassandra/net/async/HandshakeProtocolTest.java
new file mode 100644
index 0000000..a3d646d
--- /dev/null
+++ b/test/unit/org/apache/cassandra/net/async/HandshakeProtocolTest.java
@@ -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.net.async;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.async.HandshakeProtocol.FirstHandshakeMessage;
+import org.apache.cassandra.net.async.HandshakeProtocol.SecondHandshakeMessage;
+import org.apache.cassandra.net.async.HandshakeProtocol.ThirdHandshakeMessage;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.junit.Assert.assertEquals;
+
+public class HandshakeProtocolTest
+{
+    private ByteBuf buf;
+
+    @BeforeClass
+    public static void before()
+    {
+        // Kind of stupid, but the test trigger the initialization of the MessagingService class and that require
+        // DatabaseDescriptor to be configured ...
+        DatabaseDescriptor.daemonInitialization();
+    }
+
+    @After
+    public void tearDown()
+    {
+        if (buf != null && buf.refCnt() > 0)
+            buf.release();
+    }
+
+    @Test
+    public void firstMessageTest() throws Exception
+    {
+        firstMessageTest(NettyFactory.Mode.MESSAGING, false);
+        firstMessageTest(NettyFactory.Mode.MESSAGING, true);
+        firstMessageTest(NettyFactory.Mode.STREAMING, false);
+        firstMessageTest(NettyFactory.Mode.STREAMING, true);
+    }
+
+    private void firstMessageTest(NettyFactory.Mode mode, boolean compression) throws Exception
+    {
+        FirstHandshakeMessage before = new FirstHandshakeMessage(MessagingService.current_version, mode, compression);
+        buf = before.encode(PooledByteBufAllocator.DEFAULT);
+        FirstHandshakeMessage after = FirstHandshakeMessage.maybeDecode(buf);
+        assertEquals(before, after);
+        assertEquals(before.hashCode(), after.hashCode());
+        Assert.assertFalse(before.equals(null));
+    }
+
+    @Test
+    public void secondMessageTest() throws Exception
+    {
+        SecondHandshakeMessage before = new SecondHandshakeMessage(MessagingService.current_version);
+        buf = before.encode(PooledByteBufAllocator.DEFAULT);
+        SecondHandshakeMessage after = SecondHandshakeMessage.maybeDecode(buf);
+        assertEquals(before, after);
+        assertEquals(before.hashCode(), after.hashCode());
+        Assert.assertFalse(before.equals(null));
+    }
+
+    @Test
+    public void thirdMessageTest() throws Exception
+    {
+        ThirdHandshakeMessage before = new ThirdHandshakeMessage(MessagingService.current_version, FBUtilities.getBroadcastAddress());
+        buf = before.encode(PooledByteBufAllocator.DEFAULT);
+        ThirdHandshakeMessage after = ThirdHandshakeMessage.maybeDecode(buf);
+        assertEquals(before, after);
+        assertEquals(before.hashCode(), after.hashCode());
+        Assert.assertFalse(before.equals(null));
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/cassandra/blob/356dc3c2/test/unit/org/apache/cassandra/net/async/InboundHandshakeHandlerTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/net/async/InboundHandshakeHandlerTest.java b/test/unit/org/apache/cassandra/net/async/InboundHandshakeHandlerTest.java
new file mode 100644
index 0000000..44dc469
--- /dev/null
+++ b/test/unit/org/apache/cassandra/net/async/InboundHandshakeHandlerTest.java
@@ -0,0 +1,289 @@
+/*
+ * 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.async;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.ArrayList;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufOutputStream;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.compression.Lz4FrameDecoder;
+import io.netty.handler.codec.compression.Lz4FrameEncoder;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.net.CompactEndpointSerializationHelper;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.async.HandshakeProtocol.FirstHandshakeMessage;
+import org.apache.cassandra.net.async.HandshakeProtocol.ThirdHandshakeMessage;
+import org.apache.cassandra.net.async.InboundHandshakeHandler.State;
+
+import static org.apache.cassandra.net.async.NettyFactory.Mode.MESSAGING;
+
+public class InboundHandshakeHandlerTest
+{
+    private static final InetSocketAddress addr = new InetSocketAddress("127.0.0.1", 0);
+    private static final int MESSAGING_VERSION = MessagingService.current_version;
+    private static final int VERSION_30 = MessagingService.VERSION_30;
+
+    private InboundHandshakeHandler handler;
+    private EmbeddedChannel channel;
+    private ByteBuf buf;
+
+    @BeforeClass
+    public static void beforeClass()
+    {
+        DatabaseDescriptor.daemonInitialization();
+    }
+
+    @Before
+    public void setUp()
+    {
+        TestAuthenticator authenticator = new TestAuthenticator(false);
+        handler = new InboundHandshakeHandler(authenticator);
+        channel = new EmbeddedChannel(handler);
+    }
+
+    @After
+    public void tearDown()
+    {
+        if (buf != null)
+            buf.release();
+        channel.finishAndReleaseAll();
+    }
+
+    @Test
+    public void handleAuthenticate_Good()
+    {
+        handler = new InboundHandshakeHandler(new TestAuthenticator(true));
+        channel = new EmbeddedChannel(handler);
+        boolean result = handler.handleAuthenticate(addr, channel.pipeline().firstContext());
+        Assert.assertTrue(result);
+        Assert.assertTrue(channel.isOpen());
+    }
+
+    @Test
+    public void handleAuthenticate_Bad()
+    {
+        boolean result = handler.handleAuthenticate(addr, channel.pipeline().firstContext());
+        Assert.assertFalse(result);
+        Assert.assertFalse(channel.isOpen());
+        Assert.assertFalse(channel.isActive());
+    }
+
+    @Test
+    public void handleAuthenticate_BadSocketAddr()
+    {
+        boolean result = handler.handleAuthenticate(new FakeSocketAddress(), channel.pipeline().firstContext());
+        Assert.assertFalse(result);
+        Assert.assertFalse(channel.isOpen());
+        Assert.assertFalse(channel.isActive());
+    }
+
+    private static class FakeSocketAddress extends SocketAddress
+    {   }
+
+    @Test
+    public void decode_AlreadyFailed()
+    {
+        handler.setState(State.HANDSHAKE_FAIL);
+        buf = new FirstHandshakeMessage(MESSAGING_VERSION, MESSAGING, true).encode(PooledByteBufAllocator.DEFAULT);
+        handler.decode(channel.pipeline().firstContext(), buf, new ArrayList<>());
+        Assert.assertFalse(channel.isOpen());
+        Assert.assertFalse(channel.isActive());
+        Assert.assertSame(State.HANDSHAKE_FAIL, handler.getState());
+    }
+
+    @Test
+    public void handleStart_NotEnoughInputBytes() throws IOException
+    {
+        ByteBuf buf = Unpooled.EMPTY_BUFFER;
+        State state = handler.handleStart(channel.pipeline().firstContext(), buf);
+        Assert.assertEquals(State.START, state);
+        Assert.assertTrue(channel.isOpen());
+        Assert.assertTrue(channel.isActive());
+    }
+
+    @Test (expected = IOException.class)
+    public void handleStart_BadMagic() throws IOException
+    {
+        InboundHandshakeHandler handler = new InboundHandshakeHandler(new TestAuthenticator(false));
+        EmbeddedChannel channel = new EmbeddedChannel(handler);
+        buf = Unpooled.buffer(32, 32);
+
+        FirstHandshakeMessage first = new FirstHandshakeMessage(MESSAGING_VERSION,
+                                                                MESSAGING,
+                                                                true);
+
+        buf.writeInt(MessagingService.PROTOCOL_MAGIC << 2);
+        buf.writeInt(first.encodeFlags());
+        handler.handleStart(channel.pipeline().firstContext(), buf);
+    }
+
+    @Test
+    public void handleStart_VersionTooHigh() throws IOException
+    {
+        channel.eventLoop();
+        buf = new FirstHandshakeMessage(MESSAGING_VERSION + 1, MESSAGING, true).encode(PooledByteBufAllocator.DEFAULT);
+        State state = handler.handleStart(channel.pipeline().firstContext(), buf);
+        Assert.assertEquals(State.HANDSHAKE_FAIL, state);
+        Assert.assertFalse(channel.isOpen());
+        Assert.assertFalse(channel.isActive());
+    }
+
+    @Test
+    public void handleStart_VersionLessThan3_0() throws IOException
+    {
+        buf = new FirstHandshakeMessage(VERSION_30 - 1, MESSAGING, true).encode(PooledByteBufAllocator.DEFAULT);
+        State state = handler.handleStart(channel.pipeline().firstContext(), buf);
+        Assert.assertEquals(State.HANDSHAKE_FAIL, state);
+
+        Assert.assertFalse(channel.isOpen());
+        Assert.assertFalse(channel.isActive());
+    }
+
+    @Test
+    public void handleStart_HappyPath_Messaging() throws IOException
+    {
+        buf = new FirstHandshakeMessage(MESSAGING_VERSION, MESSAGING, true).encode(PooledByteBufAllocator.DEFAULT);
+        State state = handler.handleStart(channel.pipeline().firstContext(), buf);
+        Assert.assertEquals(State.AWAIT_MESSAGING_START_RESPONSE, state);
+        if (buf.refCnt() > 0)
+            buf.release();
+
+        buf = new ThirdHandshakeMessage(MESSAGING_VERSION, addr.getAddress()).encode(PooledByteBufAllocator.DEFAULT);
+        state = handler.handleMessagingStartResponse(channel.pipeline().firstContext(), buf);
+
+        Assert.assertEquals(State.MESSAGING_HANDSHAKE_COMPLETE, state);
+        Assert.assertTrue(channel.isOpen());
+        Assert.assertTrue(channel.isActive());
+        Assert.assertFalse(channel.outboundMessages().isEmpty());
+        channel.releaseOutbound();
+    }
+
+    @Test
+    public void handleMessagingStartResponse_NotEnoughInputBytes() throws IOException
+    {
+        ByteBuf buf = Unpooled.EMPTY_BUFFER;
+        State state = handler.handleMessagingStartResponse(channel.pipeline().firstContext(), buf);
+        Assert.assertEquals(State.AWAIT_MESSAGING_START_RESPONSE, state);
+        Assert.assertTrue(channel.isOpen());
+        Assert.assertTrue(channel.isActive());
+    }
+
+    @Test
+    public void handleMessagingStartResponse_BadMaxVersion() throws IOException
+    {
+        buf = Unpooled.buffer(32, 32);
+        buf.writeInt(MESSAGING_VERSION + 1);
+        CompactEndpointSerializationHelper.serialize(addr.getAddress(), new ByteBufOutputStream(buf));
+        State state = handler.handleMessagingStartResponse(channel.pipeline().firstContext(), buf);
+        Assert.assertEquals(State.HANDSHAKE_FAIL, state);
+        Assert.assertFalse(channel.isOpen());
+        Assert.assertFalse(channel.isActive());
+    }
+
+    @Test
+    public void handleMessagingStartResponse_HappyPath() throws IOException
+    {
+        buf = Unpooled.buffer(32, 32);
+        buf.writeInt(MESSAGING_VERSION);
+        CompactEndpointSerializationHelper.serialize(addr.getAddress(), new ByteBufOutputStream(buf));
+        State state = handler.handleMessagingStartResponse(channel.pipeline().firstContext(), buf);
+        Assert.assertEquals(State.MESSAGING_HANDSHAKE_COMPLETE, state);
+        Assert.assertTrue(channel.isOpen());
+        Assert.assertTrue(channel.isActive());
+    }
+
+    @Test
+    public void setupPipeline_NoCompression()
+    {
+        ChannelPipeline pipeline = channel.pipeline();
+        Assert.assertNotNull(pipeline.get(InboundHandshakeHandler.class));
+
+        handler.setupMessagingPipeline(pipeline, addr.getAddress(), false, MESSAGING_VERSION);
+        Assert.assertNotNull(pipeline.get(MessageInHandler.class));
+        Assert.assertNull(pipeline.get(Lz4FrameDecoder.class));
+        Assert.assertNull(pipeline.get(Lz4FrameEncoder.class));
+        Assert.assertNull(pipeline.get(InboundHandshakeHandler.class));
+    }
+
+    @Test
+    public void setupPipeline_WithCompression()
+    {
+        ChannelPipeline pipeline = channel.pipeline();
+        Assert.assertNotNull(pipeline.get(InboundHandshakeHandler.class));
+
+        handler.setupMessagingPipeline(pipeline, addr.getAddress(), true, MESSAGING_VERSION);
+        Assert.assertNotNull(pipeline.get(MessageInHandler.class));
+        Assert.assertNotNull(pipeline.get(Lz4FrameDecoder.class));
+        Assert.assertNull(pipeline.get(Lz4FrameEncoder.class));
+        Assert.assertNull(pipeline.get(InboundHandshakeHandler.class));
+    }
+
+    @Test
+    public void failHandshake()
+    {
+        ChannelPromise future = channel.newPromise();
+        handler.setHandshakeTimeout(future);
+        Assert.assertFalse(future.isCancelled());
+        Assert.assertTrue(channel.isOpen());
+        handler.failHandshake(channel.pipeline().firstContext());
+        Assert.assertSame(State.HANDSHAKE_FAIL, handler.getState());
+        Assert.assertTrue(future.isCancelled());
+        Assert.assertFalse(channel.isOpen());
+    }
+
+    @Test
+    public void failHandshake_AlreadyConnected()
+    {
+        ChannelPromise future = channel.newPromise();
+        handler.setHandshakeTimeout(future);
+        Assert.assertFalse(future.isCancelled());
+        Assert.assertTrue(channel.isOpen());
+        handler.setState(State.MESSAGING_HANDSHAKE_COMPLETE);
+        handler.failHandshake(channel.pipeline().firstContext());
+        Assert.assertSame(State.MESSAGING_HANDSHAKE_COMPLETE, handler.getState());
+        Assert.assertTrue(channel.isOpen());
+    }
+
+    @Test
+    public void failHandshake_TaskIsCancelled()
+    {
+        ChannelPromise future = channel.newPromise();
+        future.cancel(false);
+        handler.setHandshakeTimeout(future);
+        handler.setState(State.AWAIT_MESSAGING_START_RESPONSE);
+        Assert.assertTrue(channel.isOpen());
+        handler.failHandshake(channel.pipeline().firstContext());
+        Assert.assertSame(State.AWAIT_MESSAGING_START_RESPONSE, handler.getState());
+        Assert.assertTrue(channel.isOpen());
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/356dc3c2/test/unit/org/apache/cassandra/net/async/MessageInHandlerTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/net/async/MessageInHandlerTest.java b/test/unit/org/apache/cassandra/net/async/MessageInHandlerTest.java
new file mode 100644
index 0000000..bb82d2c
--- /dev/null
+++ b/test/unit/org/apache/cassandra/net/async/MessageInHandlerTest.java
@@ -0,0 +1,242 @@
+/*
+ * 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.async;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiConsumer;
+
+import com.google.common.base.Charsets;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufOutputStream;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.util.concurrent.Future;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.net.MessageIn;
+import org.apache.cassandra.net.MessageOut;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.async.MessageInHandler.MessageHeader;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.NanoTimeToCurrentTimeMillis;
+
+public class MessageInHandlerTest
+{
+    private static final InetSocketAddress addr = new InetSocketAddress("127.0.0.1", 0);
+    private static final int MSG_VERSION = MessagingService.current_version;
+
+    private static final int MSG_ID = 42;
+
+    private ByteBuf buf;
+
+    @BeforeClass
+    public static void before()
+    {
+        DatabaseDescriptor.daemonInitialization();
+    }
+
+    @After
+    public void tearDown()
+    {
+        if (buf != null && buf.refCnt() > 0)
+            buf.release();
+    }
+
+    @Test
+    public void decode_BadMagic() throws Exception
+    {
+        int len = MessageInHandler.FIRST_SECTION_BYTE_COUNT;
+        buf = Unpooled.buffer(len, len);
+        buf.writeInt(-1);
+        buf.writerIndex(len);
+
+        MessageInHandler handler = new MessageInHandler(addr.getAddress(), MSG_VERSION, null);
+        EmbeddedChannel channel = new EmbeddedChannel(handler);
+        Assert.assertTrue(channel.isOpen());
+        channel.writeInbound(buf);
+        Assert.assertFalse(channel.isOpen());
+    }
+
+    @Test
+    public void decode_HappyPath_NoParameters() throws Exception
+    {
+        MessageInWrapper result = decode_HappyPath(Collections.emptyMap());
+        Assert.assertTrue(result.messageIn.parameters.isEmpty());
+    }
+
+    @Test
+    public void decode_HappyPath_WithParameters() throws Exception
+    {
+        Map<String, byte[]> parameters = new HashMap<>();
+        parameters.put("p1", "val1".getBytes(Charsets.UTF_8));
+        parameters.put("p2", "val2".getBytes(Charsets.UTF_8));
+        MessageInWrapper result = decode_HappyPath(parameters);
+        Assert.assertEquals(2, result.messageIn.parameters.size());
+    }
+
+    private MessageInWrapper decode_HappyPath(Map<String, byte[]> parameters) throws Exception
+    {
+        MessageOut msgOut = new MessageOut(MessagingService.Verb.ECHO);
+        for (Map.Entry<String, byte[]> param : parameters.entrySet())
+            msgOut = msgOut.withParameter(param.getKey(), param.getValue());
+        serialize(msgOut);
+
+        MessageInWrapper wrapper = new MessageInWrapper();
+        MessageInHandler handler = new MessageInHandler(addr.getAddress(), MSG_VERSION, wrapper.messageConsumer);
+        List<Object> out = new ArrayList<>();
+        handler.decode(null, buf, out);
+
+        Assert.assertNotNull(wrapper.messageIn);
+        Assert.assertEquals(MSG_ID, wrapper.id);
+        Assert.assertEquals(msgOut.from, wrapper.messageIn.from);
+        Assert.assertEquals(msgOut.verb, wrapper.messageIn.verb);
+        Assert.assertTrue(out.isEmpty());
+
+        return wrapper;
+    }
+
+    private void serialize(MessageOut msgOut) throws IOException
+    {
+        buf = Unpooled.buffer(1024, 1024); // 1k should be enough for everybody!
+        buf.writeInt(MessagingService.PROTOCOL_MAGIC);
+        buf.writeInt(MSG_ID); // this is the id
+        buf.writeInt((int) NanoTimeToCurrentTimeMillis.convert(System.nanoTime()));
+
+        msgOut.serialize(new ByteBufDataOutputPlus(buf), MSG_VERSION);
+    }
+
+    @Test
+    public void decode_WithHalfReceivedParameters() throws Exception
+    {
+        MessageOut msgOut = new MessageOut(MessagingService.Verb.ECHO);
+        msgOut = msgOut.withParameter("p3", "val1".getBytes(Charsets.UTF_8));
+
+        serialize(msgOut);
+
+        // move the write index pointer back a few bytes to simulate like the full bytes are not present.
+        // yeah, it's lame, but it tests the basics of what is happening during the deserialiization
+        int originalWriterIndex = buf.writerIndex();
+        buf.writerIndex(originalWriterIndex - 6);
+
+        MessageInWrapper wrapper = new MessageInWrapper();
+        MessageInHandler handler = new MessageInHandler(addr.getAddress(), MSG_VERSION, wrapper.messageConsumer);
+        List<Object> out = new ArrayList<>();
+        handler.decode(null, buf, out);
+
+        Assert.assertNull(wrapper.messageIn);
+
+        MessageHeader header = handler.getMessageHeader();
+        Assert.assertEquals(MSG_ID, header.messageId);
+        Assert.assertEquals(msgOut.verb, header.verb);
+        Assert.assertEquals(msgOut.from, header.from);
+        Assert.assertTrue(out.isEmpty());
+
+        // now, set the writer index back to the original value to pretend that we actually got more bytes in
+        buf.writerIndex(originalWriterIndex);
+        handler.decode(null, buf, out);
+        Assert.assertNotNull(wrapper.messageIn);
+        Assert.assertTrue(out.isEmpty());
+    }
+
+    @Test
+    public void canReadNextParam_HappyPath() throws IOException
+    {
+        buildParamBuf(13);
+        Assert.assertTrue(MessageInHandler.canReadNextParam(buf));
+    }
+
+    @Test
+    public void canReadNextParam_OnlyFirstByte() throws IOException
+    {
+        buildParamBuf(13);
+        buf.writerIndex(1);
+        Assert.assertFalse(MessageInHandler.canReadNextParam(buf));
+    }
+
+    @Test
+    public void canReadNextParam_PartialUTF() throws IOException
+    {
+        buildParamBuf(13);
+        buf.writerIndex(5);
+        Assert.assertFalse(MessageInHandler.canReadNextParam(buf));
+    }
+
+    @Test
+    public void canReadNextParam_TruncatedValueLength() throws IOException
+    {
+        buildParamBuf(13);
+        buf.writerIndex(buf.writerIndex() - 13 - 2);
+        Assert.assertFalse(MessageInHandler.canReadNextParam(buf));
+    }
+
+    @Test
+    public void canReadNextParam_MissingLastBytes() throws IOException
+    {
+        buildParamBuf(13);
+        buf.writerIndex(buf.writerIndex() - 2);
+        Assert.assertFalse(MessageInHandler.canReadNextParam(buf));
+    }
+
+    private void buildParamBuf(int valueLength) throws IOException
+    {
+        buf = Unpooled.buffer(1024, 1024); // 1k should be enough for everybody!
+        ByteBufDataOutputPlus output = new ByteBufDataOutputPlus(buf);
+        output.writeUTF("name");
+        byte[] array = new byte[valueLength];
+        output.writeInt(array.length);
+        output.write(array);
+    }
+
+    @Test
+    public void exceptionHandled()
+    {
+        MessageInHandler handler = new MessageInHandler(addr.getAddress(), MSG_VERSION, null);
+        EmbeddedChannel channel = new EmbeddedChannel(handler);
+        Assert.assertTrue(channel.isOpen());
+        handler.exceptionCaught(channel.pipeline().firstContext(), new EOFException());
+        Assert.assertFalse(channel.isOpen());
+    }
+
+    private static class MessageInWrapper
+    {
+        MessageIn messageIn;
+        int id;
+
+        final BiConsumer<MessageIn, Integer> messageConsumer = (messageIn, integer) ->
+        {
+            this.messageIn = messageIn;
+            this.id = integer;
+        };
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/356dc3c2/test/unit/org/apache/cassandra/net/async/MessageOutHandlerTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/net/async/MessageOutHandlerTest.java b/test/unit/org/apache/cassandra/net/async/MessageOutHandlerTest.java
new file mode 100644
index 0000000..566dfdb
--- /dev/null
+++ b/test/unit/org/apache/cassandra/net/async/MessageOutHandlerTest.java
@@ -0,0 +1,289 @@
+/*
+ * 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.async;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.HashMap;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.sun.org.apache.bcel.internal.generic.DDIV;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.DefaultChannelPromise;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.UnsupportedMessageTypeException;
+import io.netty.handler.timeout.IdleStateEvent;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.net.MessageOut;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.tracing.Tracing;
+import org.apache.cassandra.utils.UUIDGen;
+
+public class MessageOutHandlerTest
+{
+    private static final int MESSAGING_VERSION = MessagingService.current_version;
+
+    private ChannelWriter channelWriter;
+    private EmbeddedChannel channel;
+    private MessageOutHandler handler;
+
+    @BeforeClass
+    public static void before()
+    {
+        DatabaseDescriptor.daemonInitialization();
+        DatabaseDescriptor.createAllDirectories();
+    }
+
+    @Before
+    public void setup()
+    {
+        setup(MessageOutHandler.AUTO_FLUSH_THRESHOLD);
+    }
+
+    private void setup(int flushThreshold)
+    {
+        OutboundConnectionIdentifier connectionId = OutboundConnectionIdentifier.small(new InetSocketAddress("127.0.0.1", 0),
+                                                                                       new InetSocketAddress("127.0.0.2", 0));
+        OutboundMessagingConnection omc = new NonSendingOutboundMessagingConnection(connectionId, null, Optional.empty());
+        channel = new EmbeddedChannel();
+        channelWriter = ChannelWriter.create(channel, omc::handleMessageResult, Optional.empty());
+        handler = new MessageOutHandler(connectionId, MESSAGING_VERSION, channelWriter, () -> null, flushThreshold);
+        channel.pipeline().addLast(handler);
+    }
+
+    @Test
+    public void write_NoFlush() throws ExecutionException, InterruptedException, TimeoutException
+    {
+        MessageOut message = new MessageOut(MessagingService.Verb.ECHO);
+        ChannelFuture future = channel.write(new QueuedMessage(message, 42));
+        Assert.assertTrue(!future.isDone());
+        Assert.assertFalse(channel.releaseOutbound());
+    }
+
+    @Test
+    public void write_WithFlush() throws ExecutionException, InterruptedException, TimeoutException
+    {
+        setup(1);
+        MessageOut message = new MessageOut(MessagingService.Verb.ECHO);
+        ChannelFuture future = channel.write(new QueuedMessage(message, 42));
+        Assert.assertTrue(future.isSuccess());
+        Assert.assertTrue(channel.releaseOutbound());
+    }
+
+    @Test
+    public void serializeMessage() throws IOException
+    {
+        channelWriter.pendingMessageCount.set(1);
+        QueuedMessage msg = new QueuedMessage(new MessageOut(MessagingService.Verb.INTERNAL_RESPONSE), 1);
+        ChannelFuture future = channel.writeAndFlush(msg);
+
+        Assert.assertTrue(future.isSuccess());
+        Assert.assertTrue(1 <= channel.outboundMessages().size());
+        Assert.assertTrue(channel.releaseOutbound());
+    }
+
+    @Test
+    public void wrongMessageType()
+    {
+        ChannelPromise promise = new DefaultChannelPromise(channel);
+        Assert.assertFalse(handler.isMessageValid("this is the wrong message type", promise));
+
+        Assert.assertFalse(promise.isSuccess());
+        Assert.assertNotNull(promise.cause());
+        Assert.assertSame(UnsupportedMessageTypeException.class, promise.cause().getClass());
+    }
+
+    @Test
+    public void unexpiredMessage()
+    {
+        QueuedMessage msg = new QueuedMessage(new MessageOut(MessagingService.Verb.INTERNAL_RESPONSE), 1);
+        ChannelPromise promise = new DefaultChannelPromise(channel);
+        Assert.assertTrue(handler.isMessageValid(msg, promise));
+
+        // we won't know if it was successful yet, but we'll know if it's a failure because cause will be set
+        Assert.assertNull(promise.cause());
+    }
+
+    @Test
+    public void expiredMessage()
+    {
+        QueuedMessage msg = new QueuedMessage(new MessageOut(MessagingService.Verb.INTERNAL_RESPONSE), 1, 0, true, true);
+        ChannelPromise promise = new DefaultChannelPromise(channel);
+        Assert.assertFalse(handler.isMessageValid(msg, promise));
+
+        Assert.assertFalse(promise.isSuccess());
+        Assert.assertNotNull(promise.cause());
+        Assert.assertSame(ExpiredException.class, promise.cause().getClass());
+        Assert.assertTrue(channel.outboundMessages().isEmpty());
+    }
+
+    @Test
+    public void write_MessageTooLarge()
+    {
+        write_BadMessageSize(Integer.MAX_VALUE + 1);
+    }
+
+    @Test
+    public void write_MessageSizeIsBananas()
+    {
+        write_BadMessageSize(Integer.MIN_VALUE + 10000);
+    }
+
+    private void write_BadMessageSize(long size)
+    {
+        IVersionedSerializer<Object> serializer = new IVersionedSerializer<Object>()
+        {
+            public void serialize(Object o, DataOutputPlus out, int version)
+            {   }
+
+            public Object deserialize(DataInputPlus in, int version)
+            {
+                return null;
+            }
+
+            public long serializedSize(Object o, int version)
+            {
+                return size;
+            }
+        };
+        MessageOut message = new MessageOut(MessagingService.Verb.UNUSED_5, "payload", serializer);
+        ChannelFuture future = channel.write(new QueuedMessage(message, 42));
+        Throwable t = future.cause();
+        Assert.assertNotNull(t);
+        Assert.assertSame(IllegalStateException.class, t.getClass());
+        Assert.assertTrue(channel.isOpen());
+        Assert.assertFalse(channel.releaseOutbound());
+    }
+
+    @Test
+    public void writeForceExceptionPath()
+    {
+        IVersionedSerializer<Object> serializer = new IVersionedSerializer<Object>()
+        {
+            public void serialize(Object o, DataOutputPlus out, int version)
+            {
+                throw new RuntimeException("this exception is part of the test - DON'T PANIC");
+            }
+
+            public Object deserialize(DataInputPlus in, int version)
+            {
+                return null;
+            }
+
+            public long serializedSize(Object o, int version)
+            {
+                return 42;
+            }
+        };
+        MessageOut message = new MessageOut(MessagingService.Verb.UNUSED_5, "payload", serializer);
+        ChannelFuture future = channel.write(new QueuedMessage(message, 42));
+        Throwable t = future.cause();
+        Assert.assertNotNull(t);
+        Assert.assertFalse(channel.isOpen());
+        Assert.assertFalse(channel.releaseOutbound());
+    }
+
+    @Test
+    public void captureTracingInfo_ForceException()
+    {
+        MessageOut message = new MessageOut(MessagingService.Verb.INTERNAL_RESPONSE)
+                             .withParameter(Tracing.TRACE_HEADER, new byte[9]);
+        handler.captureTracingInfo(new QueuedMessage(message, 42));
+    }
+
+    @Test
+    public void captureTracingInfo_UnknownSession()
+    {
+        UUID uuid = UUID.randomUUID();
+        MessageOut message = new MessageOut(MessagingService.Verb.INTERNAL_RESPONSE)
+                             .withParameter(Tracing.TRACE_HEADER, UUIDGen.decompose(uuid));
+        handler.captureTracingInfo(new QueuedMessage(message, 42));
+    }
+
+    @Test
+    public void captureTracingInfo_KnownSession()
+    {
+        Tracing.instance.newSession(new HashMap<>());
+        MessageOut message = new MessageOut(MessagingService.Verb.REQUEST_RESPONSE);
+        handler.captureTracingInfo(new QueuedMessage(message, 42));
+    }
+
+    @Test
+    public void userEventTriggered_RandomObject()
+    {
+        Assert.assertTrue(channel.isOpen());
+        ChannelUserEventSender sender = new ChannelUserEventSender();
+        channel.pipeline().addFirst(sender);
+        sender.sendEvent("ThisIsAFakeEvent");
+        Assert.assertTrue(channel.isOpen());
+    }
+
+    @Test
+    public void userEventTriggered_Idle_NoPendingBytes()
+    {
+        Assert.assertTrue(channel.isOpen());
+        ChannelUserEventSender sender = new ChannelUserEventSender();
+        channel.pipeline().addFirst(sender);
+        sender.sendEvent(IdleStateEvent.WRITER_IDLE_STATE_EVENT);
+        Assert.assertTrue(channel.isOpen());
+    }
+
+    @Test
+    public void userEventTriggered_Idle_WithPendingBytes()
+    {
+        Assert.assertTrue(channel.isOpen());
+        ChannelUserEventSender sender = new ChannelUserEventSender();
+        channel.pipeline().addFirst(sender);
+
+        MessageOut message = new MessageOut(MessagingService.Verb.INTERNAL_RESPONSE);
+        channel.writeOutbound(new QueuedMessage(message, 42));
+        sender.sendEvent(IdleStateEvent.WRITER_IDLE_STATE_EVENT);
+        Assert.assertFalse(channel.isOpen());
+    }
+
+    private static class ChannelUserEventSender extends ChannelOutboundHandlerAdapter
+    {
+        private ChannelHandlerContext ctx;
+
+        @Override
+        public void handlerAdded(final ChannelHandlerContext ctx) throws Exception
+        {
+            this.ctx = ctx;
+        }
+
+        private void sendEvent(Object event)
+        {
+            ctx.fireUserEventTriggered(event);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/356dc3c2/test/unit/org/apache/cassandra/net/async/NettyFactoryTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/net/async/NettyFactoryTest.java b/test/unit/org/apache/cassandra/net/async/NettyFactoryTest.java
new file mode 100644
index 0000000..c4cc7e6
--- /dev/null
+++ b/test/unit/org/apache/cassandra/net/async/NettyFactoryTest.java
@@ -0,0 +1,300 @@
+/*
+ * 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.async;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.Optional;
+
+import com.google.common.net.InetAddresses;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.DefaultEventLoop;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.channel.group.ChannelGroup;
+import io.netty.channel.group.DefaultChannelGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.GlobalEventExecutor;
+import org.apache.cassandra.auth.AllowAllInternodeAuthenticator;
+import org.apache.cassandra.auth.IInternodeAuthenticator;
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions;
+import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions.InternodeEncryption;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.async.NettyFactory.InboundInitializer;
+import org.apache.cassandra.net.async.NettyFactory.OutboundInitializer;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.NativeLibrary;
+
+public class NettyFactoryTest
+{
+    private static final InetSocketAddress LOCAL_ADDR = new InetSocketAddress("127.0.0.1", 9876);
+    private static final InetSocketAddress REMOTE_ADDR = new InetSocketAddress("127.0.0.2", 9876);
+    private static final int receiveBufferSize = 1 << 16;
+    private static final IInternodeAuthenticator AUTHENTICATOR = new AllowAllInternodeAuthenticator();
+
+    private ChannelGroup channelGroup;
+    private NettyFactory factory;
+
+    @BeforeClass
+    public static void before()
+    {
+        DatabaseDescriptor.daemonInitialization();
+    }
+
+    @Before
+    public void setUp()
+    {
+        channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
+    }
+
+    @After
+    public void tearDown()
+    {
+        if (factory != null)
+            factory.close();
+    }
+
+    @Test
+    public void createServerChannel_Epoll()
+    {
+        Channel inboundChannel = createServerChannel(true);
+        if (inboundChannel == null)
+            return;
+        Assert.assertEquals(EpollServerSocketChannel.class, inboundChannel.getClass());
+        inboundChannel.close();
+    }
+
+    private Channel createServerChannel(boolean useEpoll)
+    {
+        InboundInitializer inboundInitializer = new InboundInitializer(AUTHENTICATOR, null, channelGroup);
+        factory = new NettyFactory(useEpoll);
+
+        try
+        {
+            return factory.createInboundChannel(LOCAL_ADDR, inboundInitializer, receiveBufferSize);
+        }
+        catch (Exception e)
+        {
+            if (NativeLibrary.osType == NativeLibrary.OSType.LINUX)
+                throw e;
+
+            return null;
+        }
+    }
+
+    @Test
+    public void createServerChannel_Nio()
+    {
+        Channel inboundChannel = createServerChannel(false);
+        Assert.assertNotNull("we should always be able to get a NIO channel", inboundChannel);
+        Assert.assertEquals(NioServerSocketChannel.class, inboundChannel.getClass());
+        inboundChannel.close();
+    }
+
+    @Test(expected = ConfigurationException.class)
+    public void createServerChannel_SecondAttemptToBind()
+    {
+        Channel inboundChannel = null;
+        try
+        {
+            InetSocketAddress addr = new InetSocketAddress("127.0.0.1", 9876);
+            InboundInitializer inboundInitializer = new InboundInitializer(AUTHENTICATOR, null, channelGroup);
+            inboundChannel = NettyFactory.instance.createInboundChannel(LOCAL_ADDR, inboundInitializer, receiveBufferSize);
+            NettyFactory.instance.createInboundChannel(LOCAL_ADDR, inboundInitializer, receiveBufferSize);
+        }
+        finally
+        {
+            if (inboundChannel != null)
+                inboundChannel.close();
+        }
+    }
+
+    @Test(expected = ConfigurationException.class)
+    public void createServerChannel_UnbindableAddress()
+    {
+        InetSocketAddress addr = new InetSocketAddress("1.1.1.1", 9876);
+        InboundInitializer inboundInitializer = new InboundInitializer(AUTHENTICATOR, null, channelGroup);
+        NettyFactory.instance.createInboundChannel(addr, inboundInitializer, receiveBufferSize);
+    }
+
+    @Test
+    public void deterineAcceptGroupSize()
+    {
+        Assert.assertEquals(1, NettyFactory.determineAcceptGroupSize(InternodeEncryption.none));
+        Assert.assertEquals(1, NettyFactory.determineAcceptGroupSize(InternodeEncryption.all));
+        Assert.assertEquals(2, NettyFactory.determineAcceptGroupSize(InternodeEncryption.rack));
+        Assert.assertEquals(2, NettyFactory.determineAcceptGroupSize(InternodeEncryption.dc));
+
+        InetAddress originalBroadcastAddr = FBUtilities.getBroadcastAddress();
+        try
+        {
+            FBUtilities.setBroadcastInetAddress(InetAddresses.increment(FBUtilities.getLocalAddress()));
+            DatabaseDescriptor.setListenOnBroadcastAddress(true);
+
+            Assert.assertEquals(2, NettyFactory.determineAcceptGroupSize(InternodeEncryption.none));
+            Assert.assertEquals(2, NettyFactory.determineAcceptGroupSize(InternodeEncryption.all));
+            Assert.assertEquals(4, NettyFactory.determineAcceptGroupSize(InternodeEncryption.rack));
+            Assert.assertEquals(4, NettyFactory.determineAcceptGroupSize(InternodeEncryption.dc));
+        }
+        finally
+        {
+            FBUtilities.setBroadcastInetAddress(originalBroadcastAddr);
+            DatabaseDescriptor.setListenOnBroadcastAddress(false);
+        }
+    }
+
+    @Test
+    public void getEventLoopGroup_EpollWithIoRatioBoost()
+    {
+        getEventLoopGroup_Epoll(true);
+    }
+
+    private EpollEventLoopGroup getEventLoopGroup_Epoll(boolean ioBoost)
+    {
+        EventLoopGroup eventLoopGroup;
+        try
+        {
+            eventLoopGroup = NettyFactory.getEventLoopGroup(true, 1, "testEventLoopGroup", ioBoost);
+        }
+        catch (Exception e)
+        {
+            if (NativeLibrary.osType == NativeLibrary.OSType.LINUX)
+                throw e;
+
+            // ignore as epoll is only available on linux platforms, so don't fail the test on other OSes
+            return null;
+        }
+
+        Assert.assertTrue(eventLoopGroup instanceof EpollEventLoopGroup);
+        return (EpollEventLoopGroup) eventLoopGroup;
+    }
+
+    @Test
+    public void getEventLoopGroup_EpollWithoutIoRatioBoost()
+    {
+        getEventLoopGroup_Epoll(false);
+    }
+
+    @Test
+    public void getEventLoopGroup_NioWithoutIoRatioBoost()
+    {
+        getEventLoopGroup_Nio(true);
+    }
+
+    private NioEventLoopGroup getEventLoopGroup_Nio(boolean ioBoost)
+    {
+        EventLoopGroup eventLoopGroup = NettyFactory.getEventLoopGroup(false, 1, "testEventLoopGroup", ioBoost);
+        Assert.assertTrue(eventLoopGroup instanceof NioEventLoopGroup);
+        return (NioEventLoopGroup) eventLoopGroup;
+    }
+
+    @Test
+    public void getEventLoopGroup_NioWithIoRatioBoost()
+    {
+        getEventLoopGroup_Nio(true);
+    }
+
+    @Test
+    public void createOutboundBootstrap_Epoll()
+    {
+        Bootstrap bootstrap = createOutboundBootstrap(true);
+        Assert.assertEquals(EpollEventLoopGroup.class, bootstrap.config().group().getClass());
+    }
+
+    private Bootstrap createOutboundBootstrap(boolean useEpoll)
+    {
+        factory = new NettyFactory(useEpoll);
+        OutboundConnectionIdentifier id = OutboundConnectionIdentifier.gossip(LOCAL_ADDR, REMOTE_ADDR);
+        OutboundConnectionParams params = OutboundConnectionParams.builder()
+                                                                  .connectionId(id)
+                                                                  .coalescingStrategy(Optional.empty())
+                                                                  .protocolVersion(MessagingService.current_version)
+                                                                  .build();
+        return factory.createOutboundBootstrap(params);
+    }
+
+    @Test
+    public void createOutboundBootstrap_Nio()
+    {
+        Bootstrap bootstrap = createOutboundBootstrap(false);
+        Assert.assertEquals(NioEventLoopGroup.class, bootstrap.config().group().getClass());
+    }
+
+    @Test
+    public void createInboundInitializer_WithoutSsl() throws Exception
+    {
+        InboundInitializer initializer = new InboundInitializer(AUTHENTICATOR, null, channelGroup);
+        NioSocketChannel channel = new NioSocketChannel();
+        initializer.initChannel(channel);
+        Assert.assertNull(channel.pipeline().get(SslHandler.class));
+    }
+
+    private ServerEncryptionOptions encOptions()
+    {
+        ServerEncryptionOptions encryptionOptions;
+        encryptionOptions = new ServerEncryptionOptions();
+        encryptionOptions.keystore = "test/conf/cassandra_ssl_test.keystore";
+        encryptionOptions.keystore_password = "cassandra";
+        encryptionOptions.truststore = "test/conf/cassandra_ssl_test.truststore";
+        encryptionOptions.truststore_password = "cassandra";
+        encryptionOptions.require_client_auth = false;
+        encryptionOptions.cipher_suites = new String[] {"TLS_RSA_WITH_AES_128_CBC_SHA"};
+        return encryptionOptions;
+    }
+    @Test
+    public void createInboundInitializer_WithSsl() throws Exception
+    {
+        ServerEncryptionOptions encryptionOptions = encOptions();
+        InboundInitializer initializer = new InboundInitializer(AUTHENTICATOR, encryptionOptions, channelGroup);
+        NioSocketChannel channel = new NioSocketChannel();
+        Assert.assertNull(channel.pipeline().get(SslHandler.class));
+        initializer.initChannel(channel);
+        Assert.assertNotNull(channel.pipeline().get(SslHandler.class));
+    }
+
+    @Test
+    public void createOutboundInitializer_WithSsl() throws Exception
+    {
+        OutboundConnectionIdentifier id = OutboundConnectionIdentifier.gossip(LOCAL_ADDR, REMOTE_ADDR);
+        OutboundConnectionParams params = OutboundConnectionParams.builder()
+                                                                  .connectionId(id)
+                                                                  .encryptionOptions(encOptions())
+                                                                  .protocolVersion(MessagingService.current_version)
+                                                                  .build();
+        OutboundInitializer outboundInitializer = new OutboundInitializer(params);
+        NioSocketChannel channel = new NioSocketChannel();
+        Assert.assertNull(channel.pipeline().get(SslHandler.class));
+        outboundInitializer.initChannel(channel);
+        Assert.assertNotNull(channel.pipeline().get(SslHandler.class));
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/356dc3c2/test/unit/org/apache/cassandra/net/async/NonSendingOutboundMessagingConnection.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/net/async/NonSendingOutboundMessagingConnection.java b/test/unit/org/apache/cassandra/net/async/NonSendingOutboundMessagingConnection.java
new file mode 100644
index 0000000..b0b15b8
--- /dev/null
+++ b/test/unit/org/apache/cassandra/net/async/NonSendingOutboundMessagingConnection.java
@@ -0,0 +1,42 @@
+/*
+ * 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.async;
+
+import java.util.Optional;
+
+import org.apache.cassandra.auth.AllowAllInternodeAuthenticator;
+import org.apache.cassandra.config.EncryptionOptions;
+import org.apache.cassandra.utils.CoalescingStrategies;
+
+class NonSendingOutboundMessagingConnection extends OutboundMessagingConnection
+{
+    boolean sendMessageInvoked;
+
+    NonSendingOutboundMessagingConnection(OutboundConnectionIdentifier connectionId, EncryptionOptions.ServerEncryptionOptions encryptionOptions, Optional<CoalescingStrategies.CoalescingStrategy> coalescingStrategy)
+    {
+        super(connectionId, encryptionOptions, coalescingStrategy, new AllowAllInternodeAuthenticator());
+    }
+
+    @Override
+    boolean sendMessage(QueuedMessage queuedMessage)
+    {
+        sendMessageInvoked = true;
+        return true;
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/356dc3c2/test/unit/org/apache/cassandra/net/async/OutboundConnectionParamsTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/net/async/OutboundConnectionParamsTest.java b/test/unit/org/apache/cassandra/net/async/OutboundConnectionParamsTest.java
new file mode 100644
index 0000000..0ce4968
--- /dev/null
+++ b/test/unit/org/apache/cassandra/net/async/OutboundConnectionParamsTest.java
@@ -0,0 +1,36 @@
+/*
+ * 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.async;
+
+import org.junit.Test;
+
+public class OutboundConnectionParamsTest
+{
+    @Test (expected = IllegalArgumentException.class)
+    public void build_SendSizeLessThanZero()
+    {
+        OutboundConnectionParams.builder().sendBufferSize(-1).build();
+    }
+
+    @Test (expected = IllegalArgumentException.class)
+    public void build_SendSizeHuge()
+    {
+        OutboundConnectionParams.builder().sendBufferSize(1 << 30).build();
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/356dc3c2/test/unit/org/apache/cassandra/net/async/OutboundHandshakeHandlerTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/net/async/OutboundHandshakeHandlerTest.java b/test/unit/org/apache/cassandra/net/async/OutboundHandshakeHandlerTest.java
new file mode 100644
index 0000000..f8bfab1
--- /dev/null
+++ b/test/unit/org/apache/cassandra/net/async/OutboundHandshakeHandlerTest.java
@@ -0,0 +1,209 @@
+/*
+ * 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.async;
+
+import java.net.InetSocketAddress;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Optional;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.compression.Lz4FrameDecoder;
+import io.netty.handler.codec.compression.Lz4FrameEncoder;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.async.HandshakeProtocol.SecondHandshakeMessage;
+import org.apache.cassandra.net.async.OutboundHandshakeHandler.HandshakeResult;
+
+import static org.apache.cassandra.net.async.OutboundHandshakeHandler.HandshakeResult.UNKNOWN_PROTOCOL_VERSION;
+
+public class OutboundHandshakeHandlerTest
+{
+    private static final int MESSAGING_VERSION = MessagingService.current_version;
+    private static final InetSocketAddress localAddr = new InetSocketAddress("127.0.0.1", 0);
+    private static final InetSocketAddress remoteAddr = new InetSocketAddress("127.0.0.2", 0);
+    private static final String HANDLER_NAME = "clientHandshakeHandler";
+
+    private EmbeddedChannel channel;
+    private OutboundHandshakeHandler handler;
+    private OutboundConnectionIdentifier connectionId;
+    private OutboundConnectionParams params;
+    private CallbackHandler callbackHandler;
+    private ByteBuf buf;
+
+    @BeforeClass
+    public static void before()
+    {
+        DatabaseDescriptor.daemonInitialization();
+    }
+
+    @Before
+    public void setup()
+    {
+        channel = new EmbeddedChannel(new ChannelOutboundHandlerAdapter());
+        connectionId = OutboundConnectionIdentifier.small(localAddr, remoteAddr);
+        callbackHandler = new CallbackHandler();
+        params = OutboundConnectionParams.builder()
+                                         .connectionId(connectionId)
+                                         .callback(handshakeResult -> callbackHandler.receive(handshakeResult))
+                                         .mode(NettyFactory.Mode.MESSAGING)
+                                         .protocolVersion(MessagingService.current_version)
+                                         .coalescingStrategy(Optional.empty())
+                                         .build();
+        handler = new OutboundHandshakeHandler(params);
+        channel.pipeline().addFirst(HANDLER_NAME, handler);
+    }
+
+    @After
+    public void tearDown()
+    {
+        if (buf != null && buf.refCnt() > 0)
+            buf.release();
+        Assert.assertFalse(channel.finishAndReleaseAll());
+    }
+
+    @Test
+    public void decode_SmallInput() throws Exception
+    {
+        buf = Unpooled.buffer(2, 2);
+        List<Object> out = new LinkedList<>();
+        handler.decode(channel.pipeline().firstContext(), buf, out);
+        Assert.assertEquals(0, buf.readerIndex());
+        Assert.assertTrue(out.isEmpty());
+    }
+
+    @Test
+    public void decode_HappyPath() throws Exception
+    {
+        buf = new SecondHandshakeMessage(MESSAGING_VERSION).encode(PooledByteBufAllocator.DEFAULT);
+        channel.writeInbound(buf);
+        Assert.assertEquals(1, channel.outboundMessages().size());
+        Assert.assertTrue(channel.isOpen());
+        Assert.assertTrue(channel.releaseOutbound()); // throw away any responses from decode()
+
+        Assert.assertEquals(MESSAGING_VERSION, callbackHandler.result.negotiatedMessagingVersion);
+        Assert.assertEquals(HandshakeResult.Outcome.SUCCESS, callbackHandler.result.outcome);
+    }
+
+    @Test
+    public void decode_HappyPathThrowsException() throws Exception
+    {
+        callbackHandler.failOnCallback = true;
+        buf = new SecondHandshakeMessage(MESSAGING_VERSION).encode(PooledByteBufAllocator.DEFAULT);
+        channel.writeInbound(buf);
+        Assert.assertFalse(channel.isOpen());
+        Assert.assertEquals(1, channel.outboundMessages().size());
+        Assert.assertTrue(channel.releaseOutbound()); // throw away any responses from decode()
+
+        Assert.assertEquals(UNKNOWN_PROTOCOL_VERSION, callbackHandler.result.negotiatedMessagingVersion);
+        Assert.assertEquals(HandshakeResult.Outcome.NEGOTIATION_FAILURE, callbackHandler.result.outcome);
+    }
+
+    @Test
+    public void decode_ReceivedLowerMsgVersion() throws Exception
+    {
+        int msgVersion = MESSAGING_VERSION - 1;
+        buf = new SecondHandshakeMessage(msgVersion).encode(PooledByteBufAllocator.DEFAULT);
+        channel.writeInbound(buf);
+        Assert.assertTrue(channel.inboundMessages().isEmpty());
+
+        Assert.assertEquals(msgVersion, callbackHandler.result.negotiatedMessagingVersion);
+        Assert.assertEquals(HandshakeResult.Outcome.DISCONNECT, callbackHandler.result.outcome);
+        Assert.assertFalse(channel.isOpen());
+        Assert.assertTrue(channel.outboundMessages().isEmpty());
+    }
+
+    @Test
+    public void decode_ReceivedHigherMsgVersion() throws Exception
+    {
+        int msgVersion = MESSAGING_VERSION - 1;
+        channel.pipeline().remove(HANDLER_NAME);
+        params = OutboundConnectionParams.builder()
+                                         .connectionId(connectionId)
+                                         .callback(handshakeResult -> callbackHandler.receive(handshakeResult))
+                                         .mode(NettyFactory.Mode.MESSAGING)
+                                         .protocolVersion(msgVersion)
+                                         .coalescingStrategy(Optional.empty())
+                                         .build();
+        handler = new OutboundHandshakeHandler(params);
+        channel.pipeline().addFirst(HANDLER_NAME, handler);
+        buf = new SecondHandshakeMessage(MESSAGING_VERSION).encode(PooledByteBufAllocator.DEFAULT);
+        channel.writeInbound(buf);
+
+        Assert.assertEquals(MESSAGING_VERSION, callbackHandler.result.negotiatedMessagingVersion);
+        Assert.assertEquals(HandshakeResult.Outcome.DISCONNECT, callbackHandler.result.outcome);
+    }
+
+    @Test
+    public void setupPipeline_WithCompression()
+    {
+        EmbeddedChannel chan = new EmbeddedChannel(new ChannelOutboundHandlerAdapter());
+        ChannelPipeline pipeline =  chan.pipeline();
+        params = OutboundConnectionParams.builder(params).compress(true).protocolVersion(MessagingService.current_version).build();
+        handler = new OutboundHandshakeHandler(params);
+        pipeline.addFirst(handler);
+        handler.setupPipeline(chan, MESSAGING_VERSION);
+        Assert.assertNotNull(pipeline.get(Lz4FrameEncoder.class));
+        Assert.assertNull(pipeline.get(Lz4FrameDecoder.class));
+        Assert.assertNotNull(pipeline.get(MessageOutHandler.class));
+    }
+
+    @Test
+    public void setupPipeline_NoCompression()
+    {
+        EmbeddedChannel chan = new EmbeddedChannel(new ChannelOutboundHandlerAdapter());
+        ChannelPipeline pipeline =  chan.pipeline();
+        params = OutboundConnectionParams.builder(params).compress(false).protocolVersion(MessagingService.current_version).build();
+        handler = new OutboundHandshakeHandler(params);
+        pipeline.addFirst(handler);
+        handler.setupPipeline(chan, MESSAGING_VERSION);
+        Assert.assertNull(pipeline.get(Lz4FrameEncoder.class));
+        Assert.assertNull(pipeline.get(Lz4FrameDecoder.class));
+        Assert.assertNotNull(pipeline.get(MessageOutHandler.class));
+    }
+
+    private static class CallbackHandler
+    {
+        boolean failOnCallback;
+        HandshakeResult result;
+
+        Void receive(HandshakeResult handshakeResult)
+        {
+            if (failOnCallback)
+            {
+                // only fail the first callback
+                failOnCallback = false;
+                throw new RuntimeException("this exception is expected in the test - DON'T PANIC");
+            }
+            result = handshakeResult;
+            return null;
+        }
+    }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org


Mime
View raw message