sklochkov2 commented on code in PR #19754:
URL: https://github.com/apache/druid/pull/19754#discussion_r3668972308
##########
processing/src/main/java/org/apache/druid/java/util/http/client/netty/HttpClientPipelineFactory.java:
##########
@@ -19,24 +19,22 @@
package org.apache.druid.java.util.http.client.netty;
-import org.jboss.netty.channel.ChannelPipeline;
-import org.jboss.netty.channel.ChannelPipelineFactory;
-import org.jboss.netty.channel.DefaultChannelPipeline;
-import org.jboss.netty.handler.codec.http.HttpClientCodec;
-import org.jboss.netty.handler.codec.http.HttpContentDecompressor;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.http.HttpClientCodec;
+import io.netty.handler.codec.http.HttpContentDecompressor;
/**
*/
-public class HttpClientPipelineFactory implements ChannelPipelineFactory
+public class HttpClientPipelineFactory extends
ChannelInitializer<SocketChannel>
{
@Override
- public ChannelPipeline getPipeline()
+ protected void initChannel(SocketChannel ch)
{
- ChannelPipeline pipeline = new DefaultChannelPipeline();
+ ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("codec", new HttpClientCodec());
- pipeline.addLast("inflater", new HttpContentDecompressor());
-
- return pipeline;
+ pipeline.addLast("inflater", new HttpContentDecompressor(false));
Review Comment:
Addressed this in a subsequent commit.
##########
server/src/main/java/org/apache/druid/messages/client/MessageRelayClientImpl.java:
##########
@@ -76,7 +76,7 @@
return FutureUtils.transform(
asyncRequest,
holder -> {
- if (holder.getResponse().getStatus().getCode() ==
HttpStatus.NO_CONTENT_204) {
+ if (holder.getResponse().getStatus().code() ==
HttpStatus.NO_CONTENT_204) {
Review Comment:
Addressed this in a subsequent commit.
##########
server/src/main/java/org/apache/druid/discovery/DataServerResponseHandler.java:
##########
@@ -215,11 +223,11 @@ public void exceptionCaught(ClientResponse<InputStream>
clientResponse, Throwabl
setupResponseReadFailure(msg, e);
}
- private boolean enqueue(ChannelBuffer buffer, long chunkNum) throws
InterruptedException
+ private boolean enqueue(ByteBuf buffer, long chunkNum) throws
InterruptedException
{
// Increment queuedByteCount before queueing the object, so
queuedByteCount is at least as high as
// the actual number of queued bytes at any particular time.
- final InputStreamHolder holder =
InputStreamHolder.fromChannelBuffer(buffer, chunkNum);
+ final InputStreamHolder holder = InputStreamHolder.fromByteBuf(buffer,
chunkNum);
Review Comment:
Addressed this in a subsequent commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
