beobal commented on code in PR #3655:
URL: https://github.com/apache/cassandra/pull/3655#discussion_r1867544462
##########
src/java/org/apache/cassandra/transport/CQLMessageHandler.java:
##########
@@ -518,10 +533,23 @@ protected boolean
processFirstFrameOfLargeMessage(IntactFrame frame, Limit endpo
// max CQL message size defaults to 256mb, so should be safe to
downcast
int messageSize = Ints.checkedCast(header.bodySizeInBytes);
receivedBytes += buf.remaining();
+
+ if (authMessageTooBig(messageSize))
Review Comment:
Excellent point +1 to keeping it based on `ServerConnection.stage()`
##########
src/java/org/apache/cassandra/transport/CQLMessageHandler.java:
##########
@@ -747,18 +773,85 @@ private void markBackpressure(Overload backpressure)
this.backpressure = backpressure;
}
+ private void markTooBig()
+ {
+ this.tooBig = true;
+ }
+
+ protected void onIntactFrame(IntactFrame frame)
+ {
+ if (tooBig || overload != Overload.NONE)
+ // we do not want to add the frame to buffers (to not consume
a lot of memory and throw it away later
+ // we also do not want to release capacity because we haven't
accuired it
+ frame.consume();
+ else
+ super.onIntactFrame(frame);
+ }
+
+ protected void onCorruptFrame()
+ {
+ if (!isExpired && !isCorrupt && !tooBig)
+ {
+ releaseBuffers(); // release resources once we transition from
normal state to corrupt
+ if (overload != Overload.BYTES_IN_FLIGHT)
+ releaseCapacity(size);
+ }
+ isCorrupt = true;
+ isExpired |= approxTime.isAfter(expiresAtNanos);
+ }
+
+
protected void onComplete()
{
- if (overload != Overload.NONE)
-
handleErrorAndRelease(buildOverloadedException(endpointReserveCapacity,
globalReserveCapacity, overload), header);
+ if (tooBig)
+ // we haven't accuired a capacity for too big messages to
release it
+
handleError(buildOversizedCQLMessageException(header.bodySizeInBytes), header);
+ else if (overload != Overload.NONE)
+ if (overload == Overload.BYTES_IN_FLIGHT)
+ // we haven't accuired a capacity successfully to release
it
+
handleError(buildOverloadedException(endpointReserveCapacity,
globalReserveCapacity, overload), header);
+ else
+
handleErrorAndRelease(buildOverloadedException(endpointReserveCapacity,
globalReserveCapacity, overload), header);
else if (!isCorrupt)
processRequest(assembleFrame(), backpressure);
}
protected void abort()
{
- if (!isCorrupt)
- releaseBuffersAndCapacity(); // release resources if in normal
state when abort() is invoked
+ if (!isCorrupt && !tooBig && overload == Overload.NONE)
+ releaseBuffers();
+
+ if (overload == Overload.NONE || overload ==
Overload.BYTES_IN_FLIGHT)
+ releaseCapacity(size);
+ }
+
+ private OversizedCQLMessageException
buildOversizedCQLMessageException(long messageBodySize)
+ {
+ return new OversizedCQLMessageException("CQL Message of size " +
messageBodySize
+ + " bytes exceeds allowed
maximum of "
+ +
DatabaseDescriptor.getNativeTransportMaxMessageSizeInBytes() + " bytes");
+ }
+ }
+
+ static class OversizedAuthMessageException extends ProtocolException
+ {
+ OversizedAuthMessageException(String message)
+ {
+ super(message);
+ }
+ }
+
+ private boolean isItLargeMessageSentDuringAuth(Envelope.Header header)
Review Comment:
I pushed a commit here:
https://github.com/beobal/cassandra/commit/73b8895ae6c3312165330198cc4d9e720c471c98
which inlines this, seeing as it's now only used in a single place.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
