autoRead fix

Author: mhl-b

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpContentSizeHandler.java

@@ -24,7 +24,6 @@
  * TODO: move to {@link org.elasticsearch.rest.RestController}
  */
 public class Netty4HttpContentSizeHandler extends HttpObjectAggregator {
-    private boolean ignoreContentAfterContinueResponse = false;
     private boolean ignoreFollowingContent = false;
     private int contentLength = 0;
     private HttpRequest currentRequest;

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpHeaderValidator.java

@@ -50,7 +50,7 @@ State getState() {
     }
 
     boolean canAutoRead() {
-        return state != WAITING_TO_START && state != QUEUEING_DATA;
+        return state != QUEUEING_DATA && state != DROPPING_DATA_PERMANENTLY;
     }
 
     @SuppressWarnings("fallthrough")

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStream.java

@@ -106,6 +106,7 @@ public void next() {
                     send();
                 } catch (Throwable t) {
                     // must catch everything
+                    doClose();
                     channel.pipeline().fireExceptionCaught(t);
                 } finally {
                     activityTracker.stopActivity();

server/src/main/java/org/elasticsearch/rest/RestContentAggregator.java

@@ -167,13 +167,14 @@ static class NoContent implements ChunkHandler {
 
         @Override
         public void onNext(final ReleasableBytesReference lastEmptyChunk, boolean isLast) throws Exception {
+            lastEmptyChunk.close();
             if (lastEmptyChunk.length() == 0 && isLast) {
                 var aggReq = new RestRequest(request, new AggregatedHttpRequest(request.getHttpRequest(), HttpBody.empty()));
                 var aggChan = new AggregatedRestRequestChannel(channel, aggReq);
                 result.accept(aggReq, aggChan);
             } else {
                 request.contentStream().close();
-                throw new IllegalStateException("unexpected HTTP content");
+                channel.sendResponse(new RestResponse(RestStatus.BAD_REQUEST, "unexpected HTTP content"));
             }
         }
     }