You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ja...@apache.org on 2018/08/17 12:55:18 UTC
cassandra git commit: Incomplete handling of exceptions when decoding
incoming messages
Repository: cassandra
Updated Branches:
refs/heads/trunk d8c451923 -> 298416a74
Incomplete handling of exceptions when decoding incoming messages
patch by jasobrown; reviewed by Dinesh Joshi for CASSANDRA-14574
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/298416a7
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/298416a7
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/298416a7
Branch: refs/heads/trunk
Commit: 298416a7445aa50874caebc779ca3094b32f3e31
Parents: d8c4519
Author: Jason Brown <ja...@gmail.com>
Authored: Wed Jul 18 13:47:22 2018 -0700
Committer: Jason Brown <ja...@gmail.com>
Committed: Fri Aug 17 05:54:37 2018 -0700
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../net/async/BaseMessageInHandler.java | 61 ++++++++-
.../cassandra/net/async/MessageInHandler.java | 121 +++++++---------
.../net/async/MessageInHandlerPre40.java | 137 +++++++++----------
.../test/microbench/MessageOutBench.java | 6 +-
.../net/async/MessageInHandlerTest.java | 65 ++++++++-
6 files changed, 238 insertions(+), 153 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/298416a7/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index d2970a4..0e671b0 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
4.0
+ * Incomplete handling of exceptions when decoding incoming messages (CASSANDRA-14574)
* Add diagnostic events for user audit logging (CASSANDRA-13668)
* Allow retrieving diagnostic events via JMX (CASSANDRA-14435)
* Add base classes for diagnostic events (CASSANDRA-13457)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/298416a7/src/java/org/apache/cassandra/net/async/BaseMessageInHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/async/BaseMessageInHandler.java b/src/java/org/apache/cassandra/net/async/BaseMessageInHandler.java
index 7314999..2f2a973 100644
--- a/src/java/org/apache/cassandra/net/async/BaseMessageInHandler.java
+++ b/src/java/org/apache/cassandra/net/async/BaseMessageInHandler.java
@@ -26,7 +26,6 @@ import java.util.Map;
import java.util.function.BiConsumer;
import com.google.common.annotations.VisibleForTesting;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,6 +39,14 @@ import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.ParameterType;
+/**
+ * Parses out individual messages from the incoming buffers. Each message, both header and payload, is incrementally built up
+ * from the available input data, then passed to the {@link #messageConsumer}.
+ *
+ * Note: this class derives from {@link ByteToMessageDecoder} to take advantage of the {@link ByteToMessageDecoder.Cumulator}
+ * behavior across {@link #decode(ChannelHandlerContext, ByteBuf, List)} invocations. That way we don't have to maintain
+ * the not-fully consumed {@link ByteBuf}s.
+ */
public abstract class BaseMessageInHandler extends ByteToMessageDecoder
{
public static final Logger logger = LoggerFactory.getLogger(BaseMessageInHandler.class);
@@ -52,7 +59,8 @@ public abstract class BaseMessageInHandler extends ByteToMessageDecoder
READ_PARAMETERS_SIZE,
READ_PARAMETERS_DATA,
READ_PAYLOAD_SIZE,
- READ_PAYLOAD
+ READ_PAYLOAD,
+ CLOSED
}
/**
@@ -77,6 +85,8 @@ public abstract class BaseMessageInHandler extends ByteToMessageDecoder
final InetAddressAndPort peer;
final int messagingVersion;
+ protected State state;
+
public BaseMessageInHandler(InetAddressAndPort peer, int messagingVersion, BiConsumer<MessageIn, Integer> messageConsumer)
{
this.peer = peer;
@@ -84,7 +94,36 @@ public abstract class BaseMessageInHandler extends ByteToMessageDecoder
this.messageConsumer = messageConsumer;
}
- public abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out);
+ // redeclared here to make the method public (for testing)
+ @VisibleForTesting
+ public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception
+ {
+ if (state == State.CLOSED)
+ {
+ in.skipBytes(in.readableBytes());
+ return;
+ }
+
+ try
+ {
+ handleDecode(ctx, in, out);
+ }
+ catch (Exception e)
+ {
+ // prevent any future attempts at reading messages from any inbound buffers, as we're already in a bad state
+ state = State.CLOSED;
+
+ // force the buffer to appear to be consumed, thereby exiting the ByteToMessageDecoder.callDecode() loop,
+ // and other paths in that class, more efficiently
+ in.skipBytes(in.readableBytes());
+
+ // throwing the exception up causes the ByteToMessageDecoder.callDecode() loop to exit. if we don't do that,
+ // we'll keep trying to process data out of the last received buffer (and it'll be really, really wrong)
+ throw e;
+ }
+ }
+
+ public abstract void handleDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
MessageHeader readFirstChunk(ByteBuf in) throws IOException
{
@@ -105,11 +144,13 @@ public abstract class BaseMessageInHandler extends ByteToMessageDecoder
if (cause instanceof EOFException)
logger.trace("eof reading from socket; closing", cause);
else if (cause instanceof UnknownTableException)
- logger.warn("Got message from unknown table while reading from socket; closing", cause);
+ logger.warn(" Got message from unknown table while reading from socket {}[{}]; closing",
+ ctx.channel().remoteAddress(), ctx.channel().id(), cause);
else if (cause instanceof IOException)
logger.trace("IOException reading from socket; closing", cause);
else
- logger.warn("Unexpected exception caught in inbound channel pipeline from " + ctx.channel().remoteAddress(), cause);
+ logger.warn("Unexpected exception caught in inbound channel pipeline from {}[{}]",
+ ctx.channel().remoteAddress(), ctx.channel().id(), cause);
ctx.close();
}
@@ -118,6 +159,7 @@ public abstract class BaseMessageInHandler extends ByteToMessageDecoder
public void channelInactive(ChannelHandlerContext ctx)
{
logger.trace("received channel closed message for peer {} on local addr {}", ctx.channel().remoteAddress(), ctx.channel().localAddress());
+ state = State.CLOSED;
ctx.fireChannelInactive();
}
@@ -128,7 +170,7 @@ public abstract class BaseMessageInHandler extends ByteToMessageDecoder
/**
* A simple struct to hold the message header data as it is being built up.
*/
- protected static class MessageHeader
+ static class MessageHeader
{
int messageId;
long constructionTime;
@@ -145,4 +187,11 @@ public abstract class BaseMessageInHandler extends ByteToMessageDecoder
*/
int parameterLength;
}
+
+ // for testing purposes only!!!
+ @VisibleForTesting
+ public State getState()
+ {
+ return state;
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/298416a7/src/java/org/apache/cassandra/net/async/MessageInHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/async/MessageInHandler.java b/src/java/org/apache/cassandra/net/async/MessageInHandler.java
index eb22e91..0a194d4 100644
--- a/src/java/org/apache/cassandra/net/async/MessageInHandler.java
+++ b/src/java/org/apache/cassandra/net/async/MessageInHandler.java
@@ -32,8 +32,6 @@ import org.slf4j.LoggerFactory;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.ByteToMessageDecoder;
-import org.apache.cassandra.db.monitoring.ApproximateTime;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.MessageIn;
@@ -42,18 +40,12 @@ import org.apache.cassandra.net.ParameterType;
import org.apache.cassandra.utils.vint.VIntCoding;
/**
- * Parses out individual messages from the incoming buffers. Each message, both header and payload, is incrementally built up
- * from the available input data, then passed to the {@link #messageConsumer}.
- *
- * Note: this class derives from {@link ByteToMessageDecoder} to take advantage of the {@link ByteToMessageDecoder.Cumulator}
- * behavior across {@link #decode(ChannelHandlerContext, ByteBuf, List)} invocations. That way we don't have to maintain
- * the not-fully consumed {@link ByteBuf}s.
+ * Parses incoming messages as per the 4.0 internode messaging protocol.
*/
public class MessageInHandler extends BaseMessageInHandler
{
public static final Logger logger = LoggerFactory.getLogger(MessageInHandler.class);
- private State state;
private MessageHeader messageHeader;
MessageInHandler(InetAddressAndPort peer, int messagingVersion)
@@ -76,77 +68,70 @@ public class MessageInHandler extends BaseMessageInHandler
* maintains a trivial state machine to remember progress across invocations.
*/
@SuppressWarnings("resource")
- public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
+ public void handleDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception
{
ByteBufDataInputPlus inputPlus = new ByteBufDataInputPlus(in);
- try
+ while (true)
{
- while (true)
+ switch (state)
{
- switch (state)
- {
- case READ_FIRST_CHUNK:
- MessageHeader header = readFirstChunk(in);
- if (header == null)
- return;
- header.from = peer;
- messageHeader = header;
- state = State.READ_VERB;
- // fall-through
- case READ_VERB:
- if (in.readableBytes() < VERB_LENGTH)
- return;
- messageHeader.verb = MessagingService.Verb.fromId(in.readInt());
- state = State.READ_PARAMETERS_SIZE;
- // fall-through
- case READ_PARAMETERS_SIZE:
- long length = VIntCoding.readUnsignedVInt(in);
- if (length < 0)
- return;
- messageHeader.parameterLength = (int) length;
- messageHeader.parameters = messageHeader.parameterLength == 0 ? Collections.emptyMap() : new EnumMap<>(ParameterType.class);
- state = State.READ_PARAMETERS_DATA;
- // fall-through
- case READ_PARAMETERS_DATA:
- if (messageHeader.parameterLength > 0)
- {
- if (in.readableBytes() < messageHeader.parameterLength)
- return;
- readParameters(in, inputPlus, messageHeader.parameterLength, messageHeader.parameters);
- }
- state = State.READ_PAYLOAD_SIZE;
- // fall-through
- case READ_PAYLOAD_SIZE:
- length = VIntCoding.readUnsignedVInt(in);
- if (length < 0)
- return;
- messageHeader.payloadSize = (int) length;
- state = State.READ_PAYLOAD;
- // fall-through
- case READ_PAYLOAD:
- if (in.readableBytes() < messageHeader.payloadSize)
+ case READ_FIRST_CHUNK:
+ MessageHeader header = readFirstChunk(in);
+ if (header == null)
+ return;
+ header.from = peer;
+ messageHeader = header;
+ state = State.READ_VERB;
+ // fall-through
+ case READ_VERB:
+ if (in.readableBytes() < VERB_LENGTH)
+ return;
+ messageHeader.verb = MessagingService.Verb.fromId(in.readInt());
+ state = State.READ_PARAMETERS_SIZE;
+ // fall-through
+ case READ_PARAMETERS_SIZE:
+ long length = VIntCoding.readUnsignedVInt(in);
+ if (length < 0)
+ return;
+ messageHeader.parameterLength = (int) length;
+ messageHeader.parameters = messageHeader.parameterLength == 0 ? Collections.emptyMap() : new EnumMap<>(ParameterType.class);
+ state = State.READ_PARAMETERS_DATA;
+ // fall-through
+ case READ_PARAMETERS_DATA:
+ if (messageHeader.parameterLength > 0)
+ {
+ if (in.readableBytes() < messageHeader.parameterLength)
return;
+ readParameters(in, inputPlus, messageHeader.parameterLength, messageHeader.parameters);
+ }
+ state = State.READ_PAYLOAD_SIZE;
+ // fall-through
+ case READ_PAYLOAD_SIZE:
+ length = VIntCoding.readUnsignedVInt(in);
+ if (length < 0)
+ return;
+ messageHeader.payloadSize = (int) length;
+ state = State.READ_PAYLOAD;
+ // fall-through
+ case READ_PAYLOAD:
+ if (in.readableBytes() < messageHeader.payloadSize)
+ return;
- // TODO consider deserializing the message not on the event loop
- MessageIn<Object> messageIn = MessageIn.read(inputPlus, messagingVersion,
+ // TODO consider deserializing the message not on the event loop
+ MessageIn<Object> messageIn = MessageIn.read(inputPlus, messagingVersion,
messageHeader.messageId, messageHeader.constructionTime, messageHeader.from,
messageHeader.payloadSize, messageHeader.verb, messageHeader.parameters);
- if (messageIn != null)
- messageConsumer.accept(messageIn, messageHeader.messageId);
+ if (messageIn != null)
+ messageConsumer.accept(messageIn, messageHeader.messageId);
- state = State.READ_FIRST_CHUNK;
- messageHeader = null;
- break;
- default:
- throw new IllegalStateException("unknown/unhandled state: " + state);
- }
+ state = State.READ_FIRST_CHUNK;
+ messageHeader = null;
+ break;
+ default:
+ throw new IllegalStateException("unknown/unhandled state: " + state);
}
}
- catch (Exception e)
- {
- exceptionCaught(ctx, e);
- }
}
private void readParameters(ByteBuf in, ByteBufDataInputPlus inputPlus, int parameterLength, Map<ParameterType, Object> parameters) throws IOException
http://git-wip-us.apache.org/repos/asf/cassandra/blob/298416a7/src/java/org/apache/cassandra/net/async/MessageInHandlerPre40.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/async/MessageInHandlerPre40.java b/src/java/org/apache/cassandra/net/async/MessageInHandlerPre40.java
index fb19b43..f5b6fc4 100644
--- a/src/java/org/apache/cassandra/net/async/MessageInHandlerPre40.java
+++ b/src/java/org/apache/cassandra/net/async/MessageInHandlerPre40.java
@@ -39,6 +39,9 @@ import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.ParameterType;
+/**
+ * Parses incoming messages as per the pre-4.0 internode messaging protocol.
+ */
public class MessageInHandlerPre40 extends BaseMessageInHandler
{
public static final Logger logger = LoggerFactory.getLogger(MessageInHandlerPre40.class);
@@ -47,7 +50,6 @@ public class MessageInHandlerPre40 extends BaseMessageInHandler
static final int PARAMETERS_VALUE_SIZE_LENGTH = Integer.BYTES;
static final int PAYLOAD_SIZE_LENGTH = Integer.BYTES;
- private State state;
private MessageHeader messageHeader;
MessageInHandlerPre40(InetAddressAndPort peer, int messagingVersion)
@@ -70,83 +72,76 @@ public class MessageInHandlerPre40 extends BaseMessageInHandler
* maintains a trivial state machine to remember progress across invocations.
*/
@SuppressWarnings("resource")
- public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
+ public void handleDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception
{
ByteBufDataInputPlus inputPlus = new ByteBufDataInputPlus(in);
- try
+ while (true)
{
- while (true)
+ switch (state)
{
- switch (state)
- {
- case READ_FIRST_CHUNK:
- MessageHeader header = readFirstChunk(in);
- if (header == null)
- return;
- messageHeader = header;
- state = State.READ_IP_ADDRESS;
- // fall-through
- case READ_IP_ADDRESS:
- // unfortunately, this assumes knowledge of how CompactEndpointSerializationHelper serializes data (the first byte is the size).
- // first, check that we can actually read the size byte, then check if we can read that number of bytes.
- // the "+ 1" is to make sure we have the size byte in addition to the serialized IP addr count of bytes in the buffer.
- int readableBytes = in.readableBytes();
- if (readableBytes < 1 || readableBytes < in.getByte(in.readerIndex()) + 1)
- return;
- messageHeader.from = CompactEndpointSerializationHelper.instance.deserialize(inputPlus, messagingVersion);
- state = State.READ_VERB;
- // fall-through
- case READ_VERB:
- if (in.readableBytes() < VERB_LENGTH)
- return;
- messageHeader.verb = MessagingService.Verb.fromId(in.readInt());
- state = State.READ_PARAMETERS_SIZE;
- // fall-through
- case READ_PARAMETERS_SIZE:
- if (in.readableBytes() < PARAMETERS_SIZE_LENGTH)
- return;
- messageHeader.parameterLength = in.readInt();
- messageHeader.parameters = messageHeader.parameterLength == 0 ? Collections.emptyMap() : new EnumMap<>(ParameterType.class);
- state = State.READ_PARAMETERS_DATA;
- // fall-through
- case READ_PARAMETERS_DATA:
- if (messageHeader.parameterLength > 0)
- {
- if (!readParameters(in, inputPlus, messageHeader.parameterLength, messageHeader.parameters))
- return;
- }
- state = State.READ_PAYLOAD_SIZE;
- // fall-through
- case READ_PAYLOAD_SIZE:
- if (in.readableBytes() < PAYLOAD_SIZE_LENGTH)
- return;
- messageHeader.payloadSize = in.readInt();
- state = State.READ_PAYLOAD;
- // fall-through
- case READ_PAYLOAD:
- if (in.readableBytes() < messageHeader.payloadSize)
+ case READ_FIRST_CHUNK:
+ MessageHeader header = readFirstChunk(in);
+ if (header == null)
+ return;
+ messageHeader = header;
+ state = State.READ_IP_ADDRESS;
+ // fall-through
+ case READ_IP_ADDRESS:
+ // unfortunately, this assumes knowledge of how CompactEndpointSerializationHelper serializes data (the first byte is the size).
+ // first, check that we can actually read the size byte, then check if we can read that number of bytes.
+ // the "+ 1" is to make sure we have the size byte in addition to the serialized IP addr count of bytes in the buffer.
+ int readableBytes = in.readableBytes();
+ if (readableBytes < 1 || readableBytes < in.getByte(in.readerIndex()) + 1)
+ return;
+ messageHeader.from = CompactEndpointSerializationHelper.instance.deserialize(inputPlus, messagingVersion);
+ state = State.READ_VERB;
+ // fall-through
+ case READ_VERB:
+ if (in.readableBytes() < VERB_LENGTH)
+ return;
+ messageHeader.verb = MessagingService.Verb.fromId(in.readInt());
+ state = State.READ_PARAMETERS_SIZE;
+ // fall-through
+ case READ_PARAMETERS_SIZE:
+ if (in.readableBytes() < PARAMETERS_SIZE_LENGTH)
+ return;
+ messageHeader.parameterLength = in.readInt();
+ messageHeader.parameters = messageHeader.parameterLength == 0 ? Collections.emptyMap() : new EnumMap<>(ParameterType.class);
+ state = State.READ_PARAMETERS_DATA;
+ // fall-through
+ case READ_PARAMETERS_DATA:
+ if (messageHeader.parameterLength > 0)
+ {
+ if (!readParameters(in, inputPlus, messageHeader.parameterLength, messageHeader.parameters))
return;
-
- // TODO consider deserailizing the messge not on the event loop
- MessageIn<Object> messageIn = MessageIn.read(inputPlus, messagingVersion,
- messageHeader.messageId, messageHeader.constructionTime, messageHeader.from,
- messageHeader.payloadSize, messageHeader.verb, messageHeader.parameters);
-
- if (messageIn != null)
- messageConsumer.accept(messageIn, messageHeader.messageId);
-
- state = State.READ_FIRST_CHUNK;
- messageHeader = null;
- break;
- default:
- throw new IllegalStateException("unknown/unhandled state: " + state);
- }
+ }
+ state = State.READ_PAYLOAD_SIZE;
+ // fall-through
+ case READ_PAYLOAD_SIZE:
+ if (in.readableBytes() < PAYLOAD_SIZE_LENGTH)
+ return;
+ messageHeader.payloadSize = in.readInt();
+ state = State.READ_PAYLOAD;
+ // fall-through
+ case READ_PAYLOAD:
+ if (in.readableBytes() < messageHeader.payloadSize)
+ return;
+
+ // TODO consider deserailizing the messge not on the event loop
+ MessageIn<Object> messageIn = MessageIn.read(inputPlus, messagingVersion,
+ messageHeader.messageId, messageHeader.constructionTime, messageHeader.from,
+ messageHeader.payloadSize, messageHeader.verb, messageHeader.parameters);
+
+ if (messageIn != null)
+ messageConsumer.accept(messageIn, messageHeader.messageId);
+
+ state = State.READ_FIRST_CHUNK;
+ messageHeader = null;
+ break;
+ default:
+ throw new IllegalStateException("unknown/unhandled state: " + state);
}
}
- catch (Exception e)
- {
- exceptionCaught(ctx, e);
- }
}
/**
http://git-wip-us.apache.org/repos/asf/cassandra/blob/298416a7/test/microbench/org/apache/cassandra/test/microbench/MessageOutBench.java
----------------------------------------------------------------------
diff --git a/test/microbench/org/apache/cassandra/test/microbench/MessageOutBench.java b/test/microbench/org/apache/cassandra/test/microbench/MessageOutBench.java
index 43b0c16..2aec668 100644
--- a/test/microbench/org/apache/cassandra/test/microbench/MessageOutBench.java
+++ b/test/microbench/org/apache/cassandra/test/microbench/MessageOutBench.java
@@ -99,12 +99,12 @@ public class MessageOutBench
}
@Benchmark
- public int serialize40() throws IOException
+ public int serialize40() throws Exception
{
return serialize(MessagingService.VERSION_40, handler40);
}
- private int serialize(int messagingVersion, BaseMessageInHandler handler) throws IOException
+ private int serialize(int messagingVersion, BaseMessageInHandler handler) throws Exception
{
buf.resetReaderIndex();
buf.resetWriterIndex();
@@ -118,7 +118,7 @@ public class MessageOutBench
}
@Benchmark
- public int serializePre40() throws IOException
+ public int serializePre40() throws Exception
{
return serialize(MessagingService.VERSION_30, handlerPre40);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/298416a7/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
index 8deb6dc..5997861 100644
--- a/test/unit/org/apache/cassandra/net/async/MessageInHandlerTest.java
+++ b/test/unit/org/apache/cassandra/net/async/MessageInHandlerTest.java
@@ -26,6 +26,7 @@ import java.util.Collections;
import java.util.EnumMap;
import java.util.List;
import java.util.Map;
+import java.util.Random;
import java.util.UUID;
import java.util.function.BiConsumer;
@@ -148,7 +149,7 @@ public class MessageInHandlerTest
MessageOut msgOut = new MessageOut<>(addr, MessagingService.Verb.ECHO, null, null, ImmutableList.of(), SMALL_MESSAGE);
for (Map.Entry<ParameterType, Object> param : parameters.entrySet())
msgOut = msgOut.withParameter(param.getKey(), param.getValue());
- serialize(msgOut);
+ serialize(msgOut, MSG_ID);
MessageInWrapper wrapper = new MessageInWrapper();
BaseMessageInHandler handler = getHandler(addr, messagingVersion, wrapper.messageConsumer);
@@ -164,11 +165,12 @@ public class MessageInHandlerTest
return wrapper;
}
- private void serialize(MessageOut msgOut) throws IOException
+ private void serialize(MessageOut msgOut, int id) throws IOException
{
- buf = Unpooled.buffer(1024, 1024); // 1k should be enough for everybody!
+ if (buf == null)
+ 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(id); // this is the id
buf.writeInt((int) NanoTimeToCurrentTimeMillis.convert(System.nanoTime()));
msgOut.serialize(new ByteBufDataOutputPlus(buf), messagingVersion);
@@ -181,7 +183,7 @@ public class MessageInHandlerTest
UUID uuid = UUIDGen.getTimeUUID();
msgOut = msgOut.withParameter(ParameterType.TRACE_SESSION, uuid);
- serialize(msgOut);
+ serialize(msgOut, MSG_ID);
// 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
@@ -270,6 +272,52 @@ public class MessageInHandlerTest
Assert.assertFalse(channel.isOpen());
}
+ /**
+ * this is for handling the bug uncovered by CASSANDRA-14574.
+ *
+ * TL;DR if we run into a problem processing a message out an incoming buffer (and we close the channel, etc),
+ * do not attempt to process anymore messages from the buffer (force the channel closed and
+ * reject any more read attempts from the buffer).
+ *
+ * The idea here is to put several messages into a ByteBuf, pass that to the channel/handler, and make sure that
+ * only the initial, correct messages in the buffer are processed. After one messages fails the rest of the buffer
+ * should be ignored.
+ */
+ @Test
+ public void exceptionHandled_14574() throws IOException
+ {
+ Map<ParameterType, Object> parameters = new EnumMap<>(ParameterType.class);
+ parameters.put(ParameterType.FAILURE_RESPONSE, MessagingService.ONE_BYTE);
+ parameters.put(ParameterType.FAILURE_REASON, Shorts.checkedCast(RequestFailureReason.READ_TOO_MANY_TOMBSTONES.code));
+ MessageOut msgOut = new MessageOut<>(addr, MessagingService.Verb.ECHO, null, null, ImmutableList.of(), SMALL_MESSAGE);
+ for (Map.Entry<ParameterType, Object> param : parameters.entrySet())
+ msgOut = msgOut.withParameter(param.getKey(), param.getValue());
+
+ // put one complete, correct message into the buffer
+ serialize(msgOut, 1);
+
+ // add a second message, but intentionally corrupt it by manipulating a byte in it's range
+ int startPosition = buf.writerIndex();
+ serialize(msgOut, 2);
+ int positionToHack = startPosition + 2;
+ buf.setByte(positionToHack, buf.getByte(positionToHack) - 1);
+
+ // add one more complete, correct message into the buffer
+ serialize(msgOut, 3);
+
+ MessageIdsWrapper wrapper = new MessageIdsWrapper();
+ BaseMessageInHandler handler = getHandler(addr, messagingVersion, wrapper.messageConsumer);
+ EmbeddedChannel channel = new EmbeddedChannel(handler);
+ Assert.assertTrue(channel.isOpen());
+ channel.writeOneInbound(buf);
+
+ Assert.assertFalse(buf.isReadable());
+ Assert.assertEquals(BaseMessageInHandler.State.CLOSED, handler.getState());
+ Assert.assertFalse(channel.isOpen());
+ Assert.assertEquals(1, wrapper.ids.size());
+ Assert.assertEquals(Integer.valueOf(1), wrapper.ids.get(0));
+ }
+
private static class MessageInWrapper
{
MessageIn messageIn;
@@ -281,4 +329,11 @@ public class MessageInHandlerTest
this.id = integer;
};
}
+
+ private static class MessageIdsWrapper
+ {
+ private final ArrayList<Integer> ids = new ArrayList<>();
+
+ final BiConsumer<MessageIn, Integer> messageConsumer = (messageIn, integer) -> ids.add(integer);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org