[BUG] Serialization bugs can cause node drops #1885
Conversation
|
Can one of the admins verify this patch? |
|
❌ Gradle Check failure 95da4fd94f8a46ec9e79e14fc48a334fa1a5d69f |
|
❌ Gradle Check failure b12d93f48c3558b923e50f12a2977268df3507f0 |
|
✅ Gradle Check success 5bdeb781ed5c1508c838408f989c7e5a334b3224 |
|
❌ Gradle Check failure 18ba5adcd5d00f6be0b26a6a260055c8b9bbb0a9 |
Signed-off-by: Andriy Redko <andriy.redko@aiven.io>
| } | ||
| } catch (EOFException e) { | ||
| // Another favor of (de)serialization issues is when stream contains less bytes than | ||
| // the request handler needs to deserialize the payload. |
There was a problem hiding this comment.
Explicitly handle EOFException as a favour of (de)serialization failure.
| @@ -297,6 +300,23 @@ private static void sendErrorResponse(String actionName, TransportChannel transp | |||
| try { | |||
| response = handler.read(stream); | |||
| response.remoteAddress(new TransportAddress(remoteAddress)); | |||
|
|
|||
| if (stream != EMPTY_STREAM_INPUT) { | |||
There was a problem hiding this comment.
Treat unconsumed stream as (de)serialization failure and handle it as TransportSerializationException
| // Check the entire message has been read | ||
| final int nextByte = stream.read(); | ||
| // calling read() is useful to make sure the message is fully read, even if there is an EOS marker | ||
| if (nextByte != -1) { |
There was a problem hiding this comment.
Treat unconsumed stream as (de)serialization failure and handle it as TransportSerializationException
There was a problem hiding this comment.
I think there are redundant code blocks for
- handlerResponseError
- handleResponse
Can we simply this further
There was a problem hiding this comment.
Those two are slightly different (fe logging, remote address setting), I am not sure the simplification would make sense here (unless we alter the behaviour slightly)
| @@ -211,7 +224,7 @@ public TestResponse read(StreamInput in) throws IOException { | |||
|
|
|||
| BytesReference fullResponseBytes = channel.getMessageCaptor().get(); | |||
| BytesReference responseContent = fullResponseBytes.slice(headerSize, fullResponseBytes.length() - headerSize); | |||
| Header responseHeader = new Header(fullRequestBytes.length() - 6, requestId, responseStatus, version); | |||
| Header responseHeader = new Header(fullResponseBytes.length() - 6, requestId, responseStatus, version); | |||
There was a problem hiding this comment.
Seems to be an issues with copy/paste :-)
| if (nextByte != -1) { | ||
|
|
||
| try { | ||
| final T request = reg.newRequest(stream); |
There was a problem hiding this comment.
The EOFException will come from this line, right? I'd prefer to keep the EOFException try/catch more tightly bounded in order to make things easier to follow with less indentation. What do you think?
There was a problem hiding this comment.
The EOFException will come from this line, right?
Correct
I'd prefer to keep the EOFException try/catch more tightly bounded in order to make things easier to follow with less indentation. What do you think?
I thought about that as well, the issue is that request is used at the end of the code block as well
...
if (ThreadPool.Names.SAME.equals(executor)) {
try {
reg.processMessageReceived(request, transportChannel);
} catch (Exception e) {
sendErrorResponse(reg.getAction(), transportChannel, e);
}
} else {
threadPool.executor(executor).execute(new RequestHandler<>(reg, request, transportChannel));
}
So the alternative would introduce nullable request instance, no sure it is better (but readability would suffer I think). With the IllegalStateException check wrapped into dedicated function it may look better actually
There was a problem hiding this comment.
Since the catch block always throws you can do this:
final T request;
try {
request = reg.newRequest(stream);
} catch (EOFException e) {
// Another favor of (de)serialization issues is when stream contains less bytes than
// the request handler needs to deserialize the payload.
throw new IllegalStateException(
"Message fully read (request) but more data is expected for requestId ["
+ requestId
+ "], action ["
+ action
+ "]; resetting",
e
);
}
request.remoteAddress(new TransportAddress(channel.getRemoteAddress()));
checkStreamIsFullyConsumed(requestId, action, stream);
final String executor = reg.getExecutor();
if (ThreadPool.Names.SAME.equals(executor)) {
try {
reg.processMessageReceived(request, transportChannel);
} catch (Exception e) {
sendErrorResponse(reg.getAction(), transportChannel, e);
}
} else {
threadPool.executor(executor).execute(new RequestHandler<>(reg, request, transportChannel));
}There was a problem hiding this comment.
That the part I don't like to be fair:
final T request;
try {
request = reg.newRequest(stream);
}
I would try with the function instead, just a moment
There was a problem hiding this comment.
@andrross done, thank you!
server/src/main/java/org/opensearch/transport/InboundHandler.java
Outdated
Show resolved
Hide resolved
Signed-off-by: Andriy Redko <andriy.redko@aiven.io>
|
Hehe .... bad luck ... |
|
❌ Gradle Check failure ba3b9fb2f2c1568520ab68bef77cfd1867f8ca3b |
|
✅ Gradle Check success 7d5eb12c9c57cc9fca4a7c42a7344516b97bd472 |
|
❌ Gradle Check failure e4d2a618b8d2c90f303d5096a38c0ed07f5b4ce8 |
Signed-off-by: Andriy Redko <andriy.redko@aiven.io>
|
❌ Gradle Check failure 8889dcf8ad187d6660a088a67e66f1e1c47907d8 |
| * @param error "true" if response represents error, "false" otherwise | ||
| * @throws IOException IOException | ||
| */ | ||
| private void checkStreamIsFullyConsumed( |
There was a problem hiding this comment.
Are we also closing the stream on exception?
There was a problem hiding this comment.
Yes but not in the handler, afaik the streams for InboundMessage are closed within InboundPipeline
This commit restructures InboundHandler to ensure all data is consumed over the wire. Signed-off-by: Andriy Redko <andriy.redko@aiven.io>
Signed-off-by: Andriy Redko andriy.redko@aiven.io
Description
Most commonly backward incompatible changes to APIs like stats can cause ser/de over the wire to fail. A bit more details regarding the (de)serialization handling inside
InboundHandler:Interestingly, it is only raised when input stream has unconsumed bytes, the case when stream has less bytes than needed to reconstruct the payload is not classified as serialization issue (resulting in
java.io.EOFExceptions). More importantly, there is a difference if the message is request or response:requestthan all exceptions are caught insidehandleRequest, the exceptions are not escalated toTcpTransport#handleExceptionand the error response is returnedresponsethan there are two outcomes:IllegalStateExceptionis thrown, escalated up toTcpTransport#handleExceptionwhich ends up closing the specific TCP channel that caused the serialization issue, example is belowjava.io.EOFException(or any other issue manifests during the deserialization), it is wrapped intoTransportSerializationExceptionand the error response is sent, not reaching theTcpTransport#handleExceptionin any wayThe suggested changes are intended to report more details regarding the cause of the (de)serialization failure and provide uniformity in handling (de)serialization issues:
java.io.EOFExceptionas (de)serialization issue, wrapping it into theIllegalStateException("Message fully read (request) but more data is expected for requestId ..."): it does not lead to any changes in current behaviour, only more informative message is going to be returned to the client.IllegalStateExceptionand wrap it intoTransportSerializationException, it changes the current behaviour: the exception is handled in a regular way, without escalating toTcpTransport#handleExceptionand closing the channel (desired outcome) (the same wayjava.io.EOFExceptionare handled). The example is below (to compare with the previous stack trace):Issues Resolved
Closes #624
Check List
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.