ChunkedOutput.close is not flushing all the pending output to client before it is actually closed

It is critical as it is affecting the stability of our server that run on Jersey 2.22.1 (2.23.1 tested but still not working)

Server running on ChunkedOutput is not returning anything to client end occasionally. Below is the step of the logic:
1) Client initiate a post request to server.
2) Server route the request to a subresource
3) Subresource, PositionValidatorSubResource, is handling this request as a chunkedOutput. It create a future task, and handle it to a java executor.
4) The task is writing several string from client to server as chunkedoutput.
5) Client received chunkedinput.
6) Issue: We found that for some of the chunkedoutput, there're no message received on client end (no log for http). We received a null from chunkedInput.read in such case. Additionally, we found that client connection is being released from the pool and closed
7) Another issue: ChunkedInput.close is not triggering any closing. Server can still write data to ChunkedOutput even after ChunkedInput.close is called.

Server:

```
@Path("/validate")
@Singleton
public class ValidationResource {
private static Log LOG = LogFactory.getLog(ValidationResource.class);
@Inject ValidationServiceResource valResource;

public ValidationResource() {
if (LOG.isDebugEnabled()) {
LOG.debug(String.format("Root Resource %s has been initialized", this.getClass()));
}
}

@Path("/position")
public Class validateOrder() {
return PositionValidatorSubResource.class;
}
}
```

SubResource:
The subresource below is constructing a AsyncValidationWorkflow, which spawn a AsyncValidationTask (doing a dummy string return),

```
@Singleton
public class PositionValidatorSubResource {
private static final Log LOG = LogFactory.getLog(PositionValidatorSubResource.class);
private final ValidationServiceResource valResource;
private final AtomicInteger requestCount;

@Inject
public PositionValidatorSubResource(ValidationServiceResource valResource) {
this.valResource = valResource;
requestCount = new AtomicInteger(0);
}

@POST
@Path("/async")
@Consumes({ MediaType.APPLICATION_XML, MediaType.APPLICATION_ATOM_XML, MediaType.APPLICATION_JSON })
public ChunkedOutput validatePositionsWorkflow(com.ml.elt.vs.request.obj.ValidationRequest valReq) {
final ChunkedOutput output = new ChunkedOutput<>(ValidationResponse.class, CommonConsts.CHUNKED_DELIMITER);
try {
AsyncValidationWorkflow workflowWorker = new AsyncValidationWorkflow<>(output, valResp, valReq.getReqID(), valReq, valCoreValReq, valResource, reqCounter);
valResource.getServiceExecutor().execute(workflowWorker);
}
catch (Exception e) {
LOG.error(e, e);
}
return output;
}
}

public class AsyncValidationWorkflow extends RunnableControllableValidationServiceTask {
private final ChunkedOutput output;
private com.ml.equity.firm.validation.request.ValidationRequest valReq;
private final int reqCount;
private String reqID;
private T response;
private ValidationServiceResource valResource;
private ValidationRequest vsValRequest;

public AsyncValidationWorkflow(ChunkedOutput output, T response, String reqID, ValidationRequest vsValRequest, com.ml.equity.firm.validation.request.ValidationRequest valReq, ValidationServiceResource valResource, int reqCount) {
this.output = output;
this.valReq = valReq;
this.reqCount = reqCount;
this.reqID = reqID;
this.response = response;
this.valResource = valResource;
this.vsValRequest = vsValRequest;
}

@Override
public Runnable getForceCompletionHandlingTask() {
return new AsyncDefaultReturnHandlerTask<>(output, response, reqCount);
}

@Override
public Runnable getExecutionTask() {
return new AsyncValidationTask<>(output, response, reqID, vsValRequest, valReq, valResource, reqCount);
}
}

public class AsyncValidationTask extends RejectableAsyncTask implements Runnable {
private static final Log LOG = LogFactory.getLog(AsyncValidationTask.class);
private final com.ml.equity.firm.validation.request.ValidationRequest valReq;
private final ValidationRequest vsValReq;
private final int reqCount;
private final String reqID;
private final ValidationServiceResource valResource;

public AsyncValidationTask(ChunkedOutput output, T response, String reqID, ValidationRequest vsValReq, com.ml.equity.firm.validation.request.ValidationRequest valReq, ValidationServiceResource valResource, int reqCount) {
super(output, response);
this.valReq = valReq;
this.reqCount = reqCount;
this.reqID = reqID;
this.vsValReq = vsValReq;
this.valResource = valResource;
}

public void run() {
try {
response = (T)new ValidationResponse();
response.setStatusCode("ABCDEEEERRRFFFF");
output.write(response);
response = (T)new ValidationResponse();
response.setStatusCode("ABCDEEEERRRFFFF_EEERTTRTRGGGS");
output.write(response);
response = (T)new ValidationResponse();
response.setRetMsgs(Collections.singletonList(ReturnMessage.newReturnMessage(ReturnCode.BOOK_MISSING_AGGUNIT_DESK, "Test is a stupid debug message")));
output.write(response);

}
catch (Exception e) {
e.printStackTrace();
}
finally {
try {
if (output.isClosed()) {
output.close();
}
}
catch (Exception e) {
e.printStackTrace();
}
}
}
}
```

Client Code:

```
public class ValidationClient {

private static final Log LOG = LogFactory.getLog(ValidationClient.class);

private final String urlBase;
private final Client client;
private final WebTarget baseTarget;
private final LoadingCache cachedWebTargets = CacheBuilder.newBuilder().build(new CacheLoader() {
public WebTarget load(String path) {
return baseTarget.path(path);
}
});
private final ValidationClientAsyncExecutorProvider asyncExecutorProvider;

protected ValidationClient() {
this.urlBase = "";

final PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
connectionManager.setMaxTotal(1);
connectionManager.setDefaultMaxPerRoute(1);
connectionManager.setValidateAfterInactivity(2 * 1000);

defaultResponseTimeout = 20 * 1000;
defaultChunkedResponseTimeout = 86400 * 1000;

final ClientConfig clientConfig = new ClientConfig();
clientConfig.register(MOXyJsonProvider.class);
clientConfig.property(ApacheClientProperties.CONNECTION_MANAGER, connectionManager);
clientConfig.property(ApacheClientProperties.REQUEST_CONFIG,
RequestConfig.custom().setSocketTimeout(defaultResponseTimeout).setConnectTimeout(20 * 1000).build());
clientConfig.property(ClientProperties.ASYNC_THREADPOOL_SIZE, 1);
clientConfig.connectorProvider(new ApacheConnectorProvider());
asyncExecutorProvider = new ValidationClientAsyncExecutorProvider(1);
asyncExecutorProvider.setAsyncExecutorTerminationTimeoutInSecs(20 * 200);
clientConfig.register(asyncExecutorProvider);

this.mediaType = MediaType.APPLICATION_XML;
this.client = ClientBuilder.newClient(clientConfig);

this.baseTarget = client.target(urlBase);

LOG.info("ValidationClient constructed successfully");
}

private void asyncChunkedResponsePostRequest(String path, final I request, final InvocationCallback> callback, Integer chunkedResponseTimeout)
throws ValidationClientException {
try {
cachedWebTargets.get(path).request().accept(mediaType).property(ClientProperties.READ_TIMEOUT, defaultChunkedResponseTimeout) // Overridden timeout value for this // request, null will use default to // defaultChunkedResponseTimeout .async().post(Entity.entity(request, mediaType), callback);
}
catch (Exception e) {
throw handleException(e);
}
}

public void asyncChunkedResponseValidationRequest(final ValidationRequest request, final ValidationClientAsyncChunkedResponseCallback callback)
throws ValidationClientException {
asyncChunkedResponsePostRequest("/validate/position/async", request, new InvocationCallback>() {

@Override
public void failed(Throwable throwable) {
callback.failed(handleException(throwable));
}

@Override
public void completed(ChunkedInput response) {
completedAsyncChunkedResponse(request, response, callback);
}
}, chunkedResponseTimeout);
}

private void completedAsyncChunkedResponse(I request, ChunkedInput response, ValidationClientAsyncChunkedResponseCallback callback) {
response.setParser(ChunkedInput.createParser(CommonConsts.CHUNKED_DELIMITER));
O chunk;
while (!response.isClosed() && ((chunk = response.read()) != null)) { // Close by server side, or server return with a null callback.response(request, chunk);
if (callback.shouldClose(request, chunk)) { // client side reaches the stop condition LOG.info("Closing Connection from client......");
response.close();
break;
}
}

if (!response.isClosed()) {
response.close(); // close in case server return null but not sitting within should close logic }
LOG.info("Return...");
}
}
```

Testing Client:

```
@Ignore
public class ManualTestClient {

private static final Log LOG = LogFactory.getLog(ManualTestClient.class);
static AtomicInteger atm = new AtomicInteger(0);

private static ValidationClient client = ValidationClientBuilder.newClient(new ValidationClientConfigurations());

private static class TestCallback implements ValidationClientAsyncChunkedResponseCallback {
private int val;

public TestCallback(int val) {
this.val = val;

}

public void response(ValidationRequest request, ValidationResponse response) {
if (response == null) {
LOG.debug("Received : " + val + " has null response.");
}
else {
if (response.getRetMsgs() != null && !response.getRetMsgs().isEmpty()) {
LOG.debug(val + "," + "NotEmpty");
try {
Thread.sleep(10);
}
catch (Exception e) {
e.printStackTrace();
}
}
}
}

@Override
public void failed(ValidationClientException validationClientException) {
LOG.error("[received async response] failed !" + " " + val, validationClientException);
}

@Override
public boolean shouldClose(ValidationRequest request, ValidationResponse response) {
if (response != null) {
return response.getIsServiceCompleted();
}
else {
return false;
}
}

}

private static ValidationRequest getValidationRequest(int i) {
ValidationRequest req = new ValidationRequest();

return req;
}

private static void testAsyncChunked() throws InterruptedException {

for (int i = 1; i < 301; i++) {
try {
LOG.info("Request sent " + i);
client.asyncChunkedResponseValidationRequest(getValidationRequest(i), new TestCallback(i));
}
catch (Exception e) {
LOG.error("Error in test async chunked response", e);
}
}
}

public static void main(String[] args) throws InterruptedException {
BasicConfigurator.configure();

testAsyncChunked();
}

}
```
#### Affected Versions
[2.22.1, 2.22.2, 2.23.1]