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

Fix CQLConnectionTest and SimpleClient 

There are a couple of fixes in this patch. As explained on the ticket, some of the flaky failures are a race of two events that cause the termination of the connection. First is a legitimate close as a result of expected failures. In this case, the server sends the error message and closes the connection. The test expects to receive that message. However, the test was sending more messages that couldn't be received because the server already closed the connection, and they were bounced by the OS, causing immediate connection shutdown on the client side, even before it could receive the error message.

The fix is to stop sending the messages after sending the message, which is expected to cause a failure. Some other accompanying modifications include using Awaitility to wait for specific events and consider the configured maximum number of consecutive failures.

Also added some more logging to help investigate failures in the future.

Patch by Jacek Lewandowski; reviewed by Sam Tunnicliffe for CASSANDRA-16949
This commit is contained in:
Jacek Lewandowski 2023-09-28 12:36:38 +02:00
parent 4eeaf6a6ee
commit 8486d678b0
2 changed files with 68 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/java/org/apache/cassandra/transport/SimpleClient.java
View File

@ -718,6 +718,7 @@ public class SimpleClient implements Closeable
int messageSize = envelopeSize(f.header);
if (bufferSize + messageSize >= MAX_FRAMED_PAYLOAD_SIZE)
{
logger.trace("Sending frame of size: {}", bufferSize);
combiner.add(flushBuffer(ctx, buffer, bufferSize));
buffer = new ArrayList<>();
bufferSize = 0;
@ -729,7 +730,10 @@ public class SimpleClient implements Closeable
}
if (pending)
{
logger.trace("Sending frame of size: {}", bufferSize);
combiner.add(flushBuffer(ctx, buffer, bufferSize));
}
combiner.finish(promise);
}
@ -742,6 +746,7 @@ public class SimpleClient implements Closeable
payload.finish();
ChannelPromise release = AsyncChannelPromise.withListener(ctx, future -> {
logger.trace("Sent frame of size: {}", bufferSize);
for (Envelope e : messages)
e.release();
});
@ -787,7 +792,15 @@ public class SimpleClient implements Closeable
f.body.readerIndex(f.body.readerIndex() + remaining);
payload.finish();
futures.add(ctx.writeAndFlush(payload, ctx.newPromise()));
ChannelPromise promise = ctx.newPromise();
logger.trace("Sending frame of large message: {}", remaining);
futures.add(ctx.writeAndFlush(payload, promise));
promise.addListener(result -> {
if (!result.isSuccess())
logger.warn("Failed to send frame of large message, size: " + remaining, result.cause());
else
logger.trace("Sent frame of large message, size: {}", remaining);
});
}
f.release();
return futures.toArray(EMPTY_FUTURES_ARRAY);

84
test/unit/org/apache/cassandra/transport/CQLConnectionTest.java
View File

@ -30,6 +30,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.*;
import com.google.common.util.concurrent.Uninterruptibles;
import org.awaitility.Awaitility;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
@ -155,11 +157,8 @@ public class CQLConnectionTest
error -> error.error.getMessage().contains("unrecoverable CRC mismatch detected in frame body");
// expected allocated bytes are 0 as the errors happen before allocation
testFrameCorruption(10, Codec.crc(alloc), envelopeProvider, corruptor, 0, errorCheck);
testFrameCorruption(10, Codec.lz4(alloc), envelopeProvider, corruptor, 0, errorCheck);
testFrameCorruption(100, Codec.crc(alloc), envelopeProvider, corruptor, 0, errorCheck);
testFrameCorruption(100, Codec.lz4(alloc), envelopeProvider, corruptor, 0, errorCheck);
testFrameCorruption(1, Codec.crc(alloc), envelopeProvider, corruptor, 0, errorCheck);
testFrameCorruption(1, Codec.lz4(alloc), envelopeProvider, corruptor, 0, errorCheck);
// we don't do more rounds with higher message count as the connection
// will be closed when the first corrupt frame is encountered
@ -184,21 +183,19 @@ public class CQLConnectionTest
// a byte which will be in the second frame of the large message .
int seenBytes = 0;
int corruptedByte = 0;
public ByteBuf apply(ByteBuf msg)
{
// If we've already injected some corruption, pass through
if (corruptedByte > 0)
return msg;
// Will the current buffer size take us into the second frame? If so, corrupt it
if (seenBytes + msg.readableBytes() > MAX_FRAMED_PAYLOAD_SIZE + 100)
if (corruptedByte == 0 && seenBytes + msg.readableBytes() > MAX_FRAMED_PAYLOAD_SIZE * 3 / 2)
{
int frameBoundary = MAX_FRAMED_PAYLOAD_SIZE - seenBytes;
corruptedByte = msg.readerIndex() + frameBoundary + 100;
logger.info("Corrupting");
corruptedByte = msg.readerIndex() + 100;
flipBit(msg, corruptedByte);
}
else
{
logger.info("Skipping");
seenBytes += msg.readableBytes();
}
@ -206,12 +203,12 @@ public class CQLConnectionTest
}
};
int totalBytesPerEnvelope = MAX_FRAMED_PAYLOAD_SIZE * 2;
int totalBytesPerEnvelope = MAX_FRAMED_PAYLOAD_SIZE * 3 / 2; // make sure we send 2 frame and no more
IntFunction<Envelope> envelopeProvider = i -> randomEnvelope(i, Message.Type.OPTIONS, totalBytesPerEnvelope, totalBytesPerEnvelope);
Predicate<ErrorMessage> errorCheck =
error -> error.error.getMessage().contains("unrecoverable CRC mismatch detected in frame body");
testFrameCorruption(2, Codec.crc(alloc), envelopeProvider, corruptor, totalBytesPerEnvelope, errorCheck);
testFrameCorruption(1, Codec.crc(alloc), envelopeProvider, corruptor, totalBytesPerEnvelope, errorCheck);
}
@Test
@ -299,13 +296,13 @@ public class CQLConnectionTest
// badly behaved client could also include garbage which may render
// any following bytes in the Frame unusable, even though the Frame
// level CRC32 is valid for the payload.
final IntPredicate firstTen = i -> i < 10;
int maxErrorCount = DatabaseDescriptor.getConsecutiveMessageErrorsThreshold();
// mutate the request from the erroring streams to have an invalid opcode (99)
IntFunction<Envelope> envelopeProvider = mutatedEnvelopeProvider(firstTen, b -> b.put(4, (byte)99));
IntFunction<Envelope> envelopeProvider = mutatedEnvelopeProvider(ignored -> true, b -> b.put(4, (byte)99));
Predicate<ErrorMessage> errorCheck = error -> error.error.getMessage().contains("Unknown opcode 99");
testFrameCorruption(100, Codec.crc(alloc), envelopeProvider, Function.identity(), 0, errorCheck);
testFrameCorruption(maxErrorCount + 1, Codec.crc(alloc), envelopeProvider, Function.identity(), 0, errorCheck);
}
@Test
@ -314,13 +311,13 @@ public class CQLConnectionTest
// A negative value for the body length of an envelope is essentially a
// fatal exception as the stream of bytes is unrecoverable
// every other message should error while extracting the Envelope header
IntPredicate shouldError = i -> i % 2 == 0;
// the first message should error while extracting the Envelope header
IntPredicate shouldError = i -> i == 0;
// set the bodyLength byte to a negative value
IntFunction<Envelope> envelopeProvider = mutatedEnvelopeProvider(shouldError, b -> b.putInt(5, -10));
Predicate<ErrorMessage> errorCheck = error ->
error.error.getMessage().contains("Invalid value for envelope header body length field: -10");
testFrameCorruption(100, Codec.crc(alloc), envelopeProvider, Function.identity(), 0, errorCheck);
testFrameCorruption(1, Codec.crc(alloc), envelopeProvider, Function.identity(), 0, errorCheck);
}
@Test
@ -380,13 +377,13 @@ public class CQLConnectionTest
// client could also send garbage which may render any following
// bytes in the Frame unusable, even though the Frame level CRC32
// is valid for the payload.
final IntPredicate firstTen = i -> i < 10;
int maxErrorCount = DatabaseDescriptor.getConsecutiveMessageErrorsThreshold();
final ProtocolException protocolError = new ProtocolException("Unknown opcode 99");
IntFunction<Envelope> envelopeProvider = (i) -> randomEnvelope(i, Message.Type.OPTIONS);
Message.Decoder<Message.Request> decoder = new FixedDecoder(firstTen, protocolError);
FixedDecoder decoder = new FixedDecoder(ignored -> true, protocolError);
Function<ByteBuf, ByteBuf> frameTransform = Function.identity();
Predicate<ErrorMessage> errorCheck = error -> error.error.getMessage().contains(protocolError.getMessage());
testFrameCorruption(100, Codec.crc(alloc), envelopeProvider, frameTransform, 0, decoder, errorCheck);
testFrameCorruption(maxErrorCount + 1, Codec.crc(alloc), envelopeProvider, frameTransform, 0, decoder, errorCheck);
}
private void runTest(ServerConfigurator configurator,
@ -406,6 +403,7 @@ public class CQLConnectionTest
for (int i = 0; i < messageCount; i++)
client.send(envelopeProvider.apply(i));
client.awaitFlushed();
long totalBytes = client.sendSize;
@ -440,7 +438,7 @@ public class CQLConnectionTest
long expectedBytesAllocated,
Predicate<ErrorMessage> errorPredicate)
{
testFrameCorruption(messageCount, codec, envelopeProvider, transform, expectedBytesAllocated, null, errorPredicate);
testFrameCorruption(messageCount, codec, envelopeProvider, transform, expectedBytesAllocated, new FixedDecoder(), errorPredicate);
}
private void testFrameCorruption(int messageCount,
@ -448,15 +446,12 @@ public class CQLConnectionTest
IntFunction<Envelope> envelopeProvider,
Function<ByteBuf, ByteBuf> transform,
long expectedBytesAllocated,
Message.Decoder<Message.Request> requestDecoder,
FixedDecoder requestDecoder,
Predicate<ErrorMessage> errorPredicate)
{
AllocationObserver observer = new AllocationObserver(false);
InboundProxyHandler.Controller controller = new InboundProxyHandler.Controller();
if (requestDecoder == null)
requestDecoder = new FixedDecoder();
ServerConfigurator configurator = ServerConfigurator.builder()
.withAllocationObserver(observer)
.withProxyController(controller)
@ -475,10 +470,11 @@ public class CQLConnectionTest
for (int i = 0; i < messageCount; i++)
client.send(envelopeProvider.apply(i));
client.awaitFlushed();
client.awaitResponses();
// Client has disconnected
assertFalse(client.isConnected());
client.awaitState(false);
// But before it did, it sent an error response
Envelope received = client.inboundMessages.poll();
assertNotNull(received);
@ -1091,7 +1087,35 @@ public class CQLConnectionTest
private void awaitResponses() throws InterruptedException
{
responsesReceived.await(1, TimeUnit.SECONDS);
assertTrue(responsesReceived.await(10, TimeUnit.SECONDS));
}
private void awaitState(boolean connected)
{
Awaitility.await()
.atMost(10, TimeUnit.SECONDS)
.until(() -> this.connected == connected);
}
private void awaitFlushed()
{
int lastSize = flusher.outbound.size();
long lastUpdate = System.currentTimeMillis();
while (!flusher.outbound.isEmpty())
{
int newSize = flusher.outbound.size();
if (newSize < lastSize)
{
lastSize = newSize;
lastUpdate = System.currentTimeMillis();
}
else if (System.currentTimeMillis() - lastUpdate > 30000)
{
throw new RuntimeException("Timeout");
}
logger.info("Waiting for flush to complete - outbound queue size: {}", flusher.outbound.size());
Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
}
}
private boolean isConnected()