Use standard HTTP mechanism to surface truncated query responses

sql/src/main/java/org/apache/druid/sql/http/ArrayLinesWriter.java

@@ -41,6 +41,8 @@ public ArrayLinesWriter(final OutputStream outputStream, final ObjectMapper json
     this.serializers = jsonMapper.getSerializerProviderInstance();
     this.outputStream = outputStream;
     this.jsonGenerator = jsonMapper.writer().getFactory().createGenerator(outputStream);
+    // Disable auto closing of target stream since that is not managed by this writer.
+    jsonGenerator.configure(JsonGenerator.Feature.AUTO_CLOSE_TARGET, false);
     jsonGenerator.setRootValueSeparator(new SerializedString("\n"));
   }
 

sql/src/main/java/org/apache/druid/sql/http/ArrayWriter.java

@@ -44,6 +44,8 @@ public ArrayWriter(final OutputStream outputStream, final ObjectMapper jsonMappe
     this.jsonGenerator = jsonMapper.getFactory().createGenerator(outputStream);
     this.outputStream = outputStream;
 
+    // Disable auto closing of target stream since that is not managed by this writer.
+    jsonGenerator.configure(JsonGenerator.Feature.AUTO_CLOSE_TARGET, false);
     // Disable automatic JSON termination, so clients can detect truncated responses.
     jsonGenerator.configure(JsonGenerator.Feature.AUTO_CLOSE_JSON_CONTENT, false);
   }

sql/src/main/java/org/apache/druid/sql/http/CsvWriter.java

@@ -126,6 +126,7 @@ public void writeRowEnd()
   @Override
   public void close() throws IOException
   {
-    writer.close();
+    // Just flush the writer but do not close it, since we do not want to close the target output stream
+    writer.flush();
   }
 }

sql/src/main/java/org/apache/druid/sql/http/ObjectLinesWriter.java

@@ -41,6 +41,8 @@ public ObjectLinesWriter(final OutputStream outputStream, final ObjectMapper jso
     this.serializers = jsonMapper.getSerializerProviderInstance();
     this.outputStream = outputStream;
     this.jsonGenerator = jsonMapper.writer().getFactory().createGenerator(outputStream);
+    // Disable auto closing of target stream since that is not managed by this writer.
+    jsonGenerator.configure(JsonGenerator.Feature.AUTO_CLOSE_TARGET, false);
     jsonGenerator.setRootValueSeparator(new SerializedString("\n"));
   }
 

sql/src/main/java/org/apache/druid/sql/http/ObjectWriter.java

@@ -47,6 +47,8 @@ public ObjectWriter(final OutputStream outputStream, final ObjectMapper jsonMapp
     this.jsonGenerator = jsonMapper.getFactory().createGenerator(outputStream);
     this.outputStream = outputStream;
 
+    // Disable auto closing of target stream since that is not managed by this writer.
+    jsonGenerator.configure(JsonGenerator.Feature.AUTO_CLOSE_TARGET, false);
     // Disable automatic JSON termination, so clients can detect truncated responses.
     jsonGenerator.configure(JsonGenerator.Feature.AUTO_CLOSE_JSON_CONTENT, false);
   }

sql/src/main/java/org/apache/druid/sql/http/ResultFormat.java

@@ -146,9 +146,18 @@ public interface Writer extends Closeable
     void writeRowEnd() throws IOException;
 
     /**
-     * End of the response. Must allow the user to know that they have read all data successfully.
+     * End of the response. Some implementations might be writing a trailer at the end for clients to confirm
+     * that response is not truncated. However, that is no longer needed now since http client can rely on
+     * standard chunked encoding mechanics to detect a truncated response.
      */
     void writeResponseEnd() throws IOException;
+
+    /**
+     * Implementations must not close the target output stream that they are writing to.
+     * @throws IOException
+     */
+    @Override
+    void close() throws IOException;
   }
 
   @JsonCreator