diff --git a/src/java/org/apache/cassandra/transport/SimpleClient.java b/src/java/org/apache/cassandra/transport/SimpleClient.java index ae89e93e73..32acee2dbb 100644 --- a/src/java/org/apache/cassandra/transport/SimpleClient.java +++ b/src/java/org/apache/cassandra/transport/SimpleClient.java @@ -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); diff --git a/test/unit/org/apache/cassandra/transport/CQLConnectionTest.java b/test/unit/org/apache/cassandra/transport/CQLConnectionTest.java index fef25fb3e2..5c98245b81 100644 --- a/test/unit/org/apache/cassandra/transport/CQLConnectionTest.java +++ b/test/unit/org/apache/cassandra/transport/CQLConnectionTest.java @@ -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 envelopeProvider = i -> randomEnvelope(i, Message.Type.OPTIONS, totalBytesPerEnvelope, totalBytesPerEnvelope); Predicate 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 envelopeProvider = mutatedEnvelopeProvider(firstTen, b -> b.put(4, (byte)99)); + IntFunction envelopeProvider = mutatedEnvelopeProvider(ignored -> true, b -> b.put(4, (byte)99)); Predicate 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 envelopeProvider = mutatedEnvelopeProvider(shouldError, b -> b.putInt(5, -10)); Predicate 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 envelopeProvider = (i) -> randomEnvelope(i, Message.Type.OPTIONS); - Message.Decoder decoder = new FixedDecoder(firstTen, protocolError); + FixedDecoder decoder = new FixedDecoder(ignored -> true, protocolError); Function frameTransform = Function.identity(); Predicate 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 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 envelopeProvider, Function transform, long expectedBytesAllocated, - Message.Decoder requestDecoder, + FixedDecoder requestDecoder, Predicate 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()