Refactored ChunkedInputClosedOnErrorTest to give us more info - The test randomly failed and it still does, however now it checks the exception, runs more iterations (100 or more instead of 10), and is more readable. Signed-off-by: David Matějček <david.matejcek@omnifish.ee>
diff --git a/connectors/netty-connector/src/test/java/org/glassfish/jersey/netty/connector/ChunkedInputClosedOnErrorTest.java b/connectors/netty-connector/src/test/java/org/glassfish/jersey/netty/connector/ChunkedInputClosedOnErrorTest.java index 32cd089..f48f999 100644 --- a/connectors/netty-connector/src/test/java/org/glassfish/jersey/netty/connector/ChunkedInputClosedOnErrorTest.java +++ b/connectors/netty-connector/src/test/java/org/glassfish/jersey/netty/connector/ChunkedInputClosedOnErrorTest.java
@@ -1,4 +1,5 @@ /* + * Copyright (c) 2026 Contributors to the Eclipse Foundation. * Copyright (c) 2025 Oracle and/or its affiliates. All rights reserved. * * This program and the accompanying materials are made available under the @@ -16,18 +17,9 @@ package org.glassfish.jersey.netty.connector; +import io.netty.buffer.ByteBuf; import io.netty.channel.Channel; import io.netty.handler.stream.ChunkedInput; -import org.glassfish.jersey.client.ClientConfig; -import org.glassfish.jersey.client.ClientProperties; -import org.glassfish.jersey.client.ClientRequest; -import org.glassfish.jersey.client.spi.Connector; -import org.glassfish.jersey.client.spi.ConnectorProvider; -import org.glassfish.jersey.netty.connector.internal.NettyEntityWriter; -import org.glassfish.jersey.server.ResourceConfig; -import org.glassfish.jersey.test.JerseyTest; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Test; import javax.ws.rs.ProcessingException; import javax.ws.rs.WebApplicationException; @@ -38,8 +30,8 @@ import javax.ws.rs.core.Configuration; import javax.ws.rs.core.MediaType; import javax.ws.rs.core.MultivaluedMap; -import javax.ws.rs.core.Response; import javax.ws.rs.ext.MessageBodyWriter; + import java.io.IOException; import java.io.OutputStream; import java.lang.annotation.Annotation; @@ -47,118 +39,136 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicReference; +import org.glassfish.jersey.client.ClientConfig; +import org.glassfish.jersey.client.ClientProperties; +import org.glassfish.jersey.client.ClientRequest; +import org.glassfish.jersey.client.spi.Connector; +import org.glassfish.jersey.client.spi.ConnectorProvider; +import org.glassfish.jersey.netty.connector.internal.NettyEntityWriter; +import org.glassfish.jersey.server.ResourceConfig; +import org.glassfish.jersey.test.JerseyTest; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertAll; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + /** * Bug 5837 reproducer */ public class ChunkedInputClosedOnErrorTest extends JerseyTest { - private static Client initClient(ConnectorProvider provider) { - ClientConfig defaultConfig = new ClientConfig(); - defaultConfig.property(ClientProperties.CONNECT_TIMEOUT, 10 * 1000); - defaultConfig.property(ClientProperties.READ_TIMEOUT, 10 * 1000); - defaultConfig.connectorProvider(provider); - Client client = ClientBuilder.newBuilder() - .withConfig(defaultConfig) - .build(); - return client; - } - @Override protected Application configure() { return new ResourceConfig(); } @Test - public void testChunkedInputNotStuckedTimes() throws InterruptedException { - for (int i = 0; i != 10; i++) { - boolean ret = testChunkedInputNotStucked(); - Assertions.assertTrue(ret, "JerseyChunkedInput was not closed on error"); + public void testChunkedInputNotStuckedTimes() throws Exception { + long timeout = System.currentTimeMillis() + 100; + int counter = 0; + while (System.currentTimeMillis() < timeout || counter < 300) { + counter++; + TestNettyConnectorProvider provider = new TestNettyConnectorProvider(); + Client client = provider.initClient(); + ProcessingException processingException = assertThrows(ProcessingException.class, + () -> client.register(new MultipartWriter()).target(target().getUri()).request() + .post(Entity.entity(new MultipartWriter(), MediaType.MULTIPART_FORM_DATA_TYPE))); + provider.waitForWriterSet(); + assertAll("Iteration: " + counter, + () -> assertInstanceOf(IllegalArgumentException.class, processingException.getCause()), + () -> assertEquals("TestException", processingException.getCause().getMessage()), + () -> assertTrue(provider.isEndOfInput(), "JerseyChunkedInput was not closed on error") + ); } } - public boolean testChunkedInputNotStucked() throws InterruptedException { - final AtomicReference<NettyEntityWriter> writer = new AtomicReference<>(); - final CountDownLatch writerSetLatch = new CountDownLatch(1); - final CountDownLatch flushLatch = new CountDownLatch(1); - ConnectorProvider provider = new ConnectorProvider() { - @Override - public Connector getConnector(Client client, Configuration runtimeConfig) { - return new NettyConnector(client, NettyConnectorProvider.config().rw()) { - @Override - NettyEntityWriter nettyEntityWriter(ClientRequest clientRequest, Channel channel, - NettyConnectorProvider.Config.RW config) { - writer.set(super.nettyEntityWriter(clientRequest, channel, config)); - writerSetLatch.countDown(); - return new NettyEntityWriter() { - private boolean slept = false; + private static class TestNettyConnectorProvider implements ConnectorProvider { - @Override - public void write(Object object) { - writer.get().write(object); - } + private final AtomicReference<NettyEntityWriter> writer = new AtomicReference<>(); + private final CountDownLatch writerSetLatch = new CountDownLatch(1); + private final CountDownLatch flushLatch = new CountDownLatch(1); - @Override - public void writeAndFlush(Object object) { - writer.get().writeAndFlush(object); - } + Client initClient() { + ClientConfig defaultConfig = new ClientConfig(); + defaultConfig.property(ClientProperties.CONNECT_TIMEOUT, 1_000); + defaultConfig.property(ClientProperties.READ_TIMEOUT, 10_000); + defaultConfig.connectorProvider(this); + return ClientBuilder.newBuilder().withConfig(defaultConfig).build(); + } - @Override - public void flush() throws IOException { - writer.get().flush(); - flushLatch.countDown(); - } + void waitForWriterSet() throws InterruptedException { + writerSetLatch.await(); + } - @Override - public ChunkedInput getChunkedInput() { - for (StackTraceElement element : Thread.currentThread().getStackTrace()) { - // caught from catch block in executorService.execute(new Runnable() { - // "sleep" to simulate race condition - if (element.getClassName().contains("NettyConnector") - && element.getMethodName().equals("run")) { - try { - flushLatch.await(); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } + boolean isEndOfInput() throws Exception { + return writer.get().getChunkedInput().isEndOfInput(); + } + + @Override + public Connector getConnector(Client client, Configuration runtimeConfig) { + return new NettyConnector(client, NettyConnectorProvider.config().rw()) { + + @Override + NettyEntityWriter nettyEntityWriter(ClientRequest clientRequest, Channel channel, + NettyConnectorProvider.Config.RW config) { + writer.set(super.nettyEntityWriter(clientRequest, channel, config)); + writerSetLatch.countDown(); + return new NettyEntityWriter() { + + @Override + public void write(Object object) { + writer.get().write(object); + } + + @Override + public void writeAndFlush(Object object) { + writer.get().writeAndFlush(object); + } + + @Override + public void flush() throws IOException { + writer.get().flush(); + flushLatch.countDown(); + } + + @Override + public ChunkedInput<ByteBuf> getChunkedInput() { + for (StackTraceElement element : Thread.currentThread().getStackTrace()) { + // caught from catch block in executorService.execute(new Runnable() + // { + // "sleep" to simulate race condition + if (element.getClassName().contains("NettyConnector") + && element.getMethodName().equals("run")) { + try { + flushLatch.await(); + } catch (InterruptedException e) { + throw new RuntimeException(e); } } - return writer.get().getChunkedInput(); } + return writer.get().getChunkedInput(); + } - @Override - public OutputStream getOutputStream() { - return writer.get().getOutputStream(); - } + @Override + public OutputStream getOutputStream() { + return writer.get().getOutputStream(); + } - @Override - public long getLength() { - return writer.get().getLength(); - } + @Override + public long getLength() { + return writer.get().getLength(); + } - @Override - public Type getType() { - return writer.get().getType(); - } - }; - } - }; - } - }; - - Client client = initClient(provider); - try { - Response r = client - .register(new MultipartWriter()) - .target(target().getUri()).request() - .post(Entity.entity(new MultipartWriter(), MediaType.MULTIPART_FORM_DATA_TYPE)); - } catch (ProcessingException expected) { - - } - writerSetLatch.await(); - try { - return writer.get().getChunkedInput().isEndOfInput(); - } catch (Exception e) { - throw new RuntimeException(e); + @Override + public Type getType() { + return writer.get().getType(); + } + }; + } + }; } } @@ -170,11 +180,10 @@ } @Override - public void writeTo(Object object, Class<?> type, Type genericType, Annotation[] annotations, MediaType mediaType, - MultivaluedMap<String, Object> httpHeaders, OutputStream entityStream) throws IOException, - WebApplicationException { + public void writeTo(Object object, Class<?> type, Type genericType, Annotation[] annotations, + MediaType mediaType, MultivaluedMap<String, Object> httpHeaders, OutputStream entityStream) + throws IOException, WebApplicationException { throw new IllegalArgumentException("TestException"); } } - }