#7281 check that at least one byte of raw content is consumed by the…

… interceptor and clarify its javadoc

Signed-off-by: Ludovic Orban <lorban@bitronix.be>

jetty-server/src/main/java/org/eclipse/jetty/server/AsyncContentProducer.java

@@ -432,16 +432,32 @@ private HttpInput.Content intercept()
     {
         try
         {
+            int remainingBeforeInterception = _rawContent.remaining();
             HttpInput.Content content = _interceptor.readFrom(_rawContent);
             if (content != null && content.isSpecial() && !_rawContent.isSpecial())
             {
                 // Set the _error flag to mark the content as definitive, i.e.:
                 // do not try to produce new raw content to get a fresher error
                 // when the special content was generated by the interceptor.
                 _error = true;
+                Throwable error = content.getError();
+                if (error != null && _httpChannel.getResponse().isCommitted())
+                    _httpChannel.abort(error);
                 if (LOG.isDebugEnabled())
                     LOG.debug("interceptor generated special content {}", this);
             }
+            else if (content != _rawContent && !_rawContent.isSpecial() && _rawContent.remaining() == remainingBeforeInterception)
+            {
+                IOException failure = new IOException("Interceptor " + _interceptor + " did not consume any of the " + _rawContent.remaining() + " remaining byte(s) of content");
+                failCurrentContent(failure);
+                // Set the _error flag to mark the content as definitive, i.e.:
+                // do not try to produce new raw content to get a fresher error
+                // when the special content was caused by the interceptor not
+                // consuming the raw content.
+                _error = true;
+                content = _transformedContent;
+            }
+
             if (LOG.isDebugEnabled())
                 LOG.debug("intercepted raw content {}", this);
             return content;

jetty-server/src/main/java/org/eclipse/jetty/server/HttpInput.java

@@ -453,8 +453,6 @@ public String toString()
      *     occur only after the contained byte buffer is empty (see above) or at any time if the returned content was special.</li>
      *     <li>Once {@link #readFrom(Content)} returned a special content, subsequent calls to {@link #readFrom(Content)} must
      *     always return the same special content.</li>
-     *     <li>When {@link #readFrom(Content)} gets passed a non-special content, it must either return the content it was
-     *     passed or fully consume the contained byte buffer.</li>
      *     <li>Implementations implementing both this interface and {@link Destroyable} will have their
      *     {@link Destroyable#destroy()} method called when {@link #recycle()} is called.</li>
      * </ul>
@@ -464,8 +462,9 @@ public interface Interceptor
     {
         /**
          * @param content The content to be intercepted.
-         * The content will be modified with any data the interceptor consumes, but there is no requirement
-         * that all the data is consumed by the interceptor.
+         * The content will be modified with any data the interceptor consumes. There is no requirement
+         * that all the data is consumed by the interceptor but at least one byte must be consumed
+         * unless the returned content is the passed content instance.
          * @return The intercepted content or null if interception is completed for that content.
          */
         Content readFrom(Content content);

jetty-server/src/test/java/org/eclipse/jetty/server/AsyncContentProducerTest.java

@@ -40,6 +40,7 @@
 import org.junit.jupiter.api.Test;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.endsWith;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
@@ -279,6 +280,35 @@ public void failed(Throwable x)
         assertThat(contentFailedCount.get(), is(1));
     }
 
+    @Test
+    public void testAsyncContentProducerInterceptorDoesNotConsume()
+    {
+        AtomicInteger contentFailedCount = new AtomicInteger();
+        ContentProducer contentProducer = new AsyncContentProducer(new StaticContentHttpChannel(new HttpInput.Content(ByteBuffer.allocate(1))
+        {
+            @Override
+            public void failed(Throwable x)
+            {
+                contentFailedCount.incrementAndGet();
+            }
+        }));
+        try (AutoLock lock = contentProducer.lock())
+        {
+            contentProducer.setInterceptor(content -> null);
+
+            assertThat(contentProducer.isReady(), is(true));
+
+            HttpInput.Content content1 = contentProducer.nextContent();
+            assertThat(content1.isSpecial(), is(true));
+            assertThat(content1.getError().getMessage(), endsWith("did not consume any of the 1 remaining byte(s) of content"));
+
+            HttpInput.Content content2 = contentProducer.nextContent();
+            assertThat(content2.isSpecial(), is(true));
+            assertThat(content1.getError().getMessage(), endsWith("did not consume any of the 1 remaining byte(s) of content"));
+        }
+        assertThat(contentFailedCount.get(), is(1));
+    }
+
     @Test
     public void testAsyncContentProducerGzipInterceptor() throws Exception
     {

jetty-server/src/test/java/org/eclipse/jetty/server/BlockingContentProducerTest.java

@@ -31,12 +31,12 @@
 import org.eclipse.jetty.util.component.Destroyable;
 import org.eclipse.jetty.util.compression.InflaterPool;
 import org.eclipse.jetty.util.thread.AutoLock;
-import org.hamcrest.Matchers;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.endsWith;
 import static org.hamcrest.Matchers.nullValue;
 import static org.hamcrest.core.Is.is;
 
@@ -273,6 +273,33 @@ public void testBlockingContentProducerErrorContentIsPassedToInterceptor() throw
         assertThat(interceptor.contents.get(3).getError().getMessage(), is("testBlockingContentProducerErrorContentIsPassedToInterceptor error"));
     }
 
+    @Test
+    public void testBlockingContentProducerInterceptorDoesNotConsume()
+    {
+        AtomicInteger contentFailedCount = new AtomicInteger();
+        ContentProducer contentProducer = new BlockingContentProducer(new AsyncContentProducer(new StaticContentHttpChannel(new HttpInput.Content(ByteBuffer.allocate(1))
+        {
+            @Override
+            public void failed(Throwable x)
+            {
+                contentFailedCount.incrementAndGet();
+            }
+        })));
+        try (AutoLock lock = contentProducer.lock())
+        {
+            contentProducer.setInterceptor(content -> null);
+
+            HttpInput.Content content1 = contentProducer.nextContent();
+            assertThat(content1.isSpecial(), is(true));
+            assertThat(content1.getError().getMessage(), endsWith("did not consume any of the 1 remaining byte(s) of content"));
+
+            HttpInput.Content content2 = contentProducer.nextContent();
+            assertThat(content2.isSpecial(), is(true));
+            assertThat(content1.getError().getMessage(), endsWith("did not consume any of the 1 remaining byte(s) of content"));
+        }
+        assertThat(contentFailedCount.get(), is(1));
+    }
+
     @Test
     public void testBlockingContentProducerGzipInterceptor()
     {