title: Netty request timeout
body: <p>I'm trying to write an HTTP service that will take data from HTTP and put it in Kafka using Netty. I need to handle 20K RPS on a m5.large EC2 instance, which seems pretty doable.</p>
  
  <p>The code is simple:</p>
  
  <p><em><code>Server.java</code></em></p>
  
  <pre><code>public class Server {
      public static void main(final String[] args) throws Exception {
          final EventLoopGroup bossGroup = new EpollEventLoopGroup();
          final EventLoopGroup workerGroup = new EpollEventLoopGroup();
  
          try {
              final ServerBootstrap bootstrap = new ServerBootstrap();
  
              bootstrap
                  .group(bossGroup, workerGroup)
                  .channel(EpollServerSocketChannel.class)
                  .childHandler(new RequestChannelInitializer(createProducer()))
                  .childOption(ChannelOption.SO_KEEPALIVE, true);
              bootstrap.bind(PORT).sync().channel().closeFuture().sync();
          } finally {
              bossGroup.shutdownGracefully();
              workerGroup.shutdownGracefully();
          }
      }
  
      private static Producer&lt;String, ByteBuffer&gt; createProducer() {
          final Properties properties = new Properties();
  
          properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_HOST);
          properties.put(ProducerConfig.CLIENT_ID_CONFIG, "KafkaBidRequestProducer");
          properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteBufferSerializer.class.getName());
          properties.put(ProducerConfig.RETRIES_CONFIG, 0);
          properties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 10000);
          properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);
          properties.put(ProducerConfig.SEND_BUFFER_CONFIG, 33554432);
  
          return new KafkaProducer&lt;&gt;(properties);
      }
  }
  </code></pre>
  
  <p><em><code>RequestChannelInitializer.java</code></em></p>
  
  <pre><code>public class RequestChannelInitializer extends io.netty.channel.ChannelInitializer&lt;SocketChannel&gt; {
      private final Producer&lt;String, ByteBuffer&gt; producer;
  
      public BidRequestChannelInitializer(final Producer&lt;String, ByteBuffer&gt; producer) {
          this.producer = producer;
      }
  
      @Override
      public void initChannel(final SocketChannel ch) {
          ch.pipeline().addLast(new HttpServerCodec());
          ch.pipeline().addLast(new HttpObjectAggregator(1048576));
          ch.pipeline().addLast(new RequestHandler(producer));
      }
  }
  </code></pre>
  
  <p><em><code>RequestHandler.java</code></em></p>
  
  <pre><code>public class RequestHandler extends SimpleChannelInboundHandler&lt;FullHttpMessage&gt; {
      private final Producer&lt;String, ByteBuffer&gt; producer;
  
      public BidRequestHandler(final Producer&lt;String, ByteBuffer&gt; producer) {
          this.producer = producer;
      }
  
      @Override
      public void channelReadComplete(final ChannelHandlerContext ctx) {
          ctx.flush();
      }
  
      @Override
      protected void channelRead0(ChannelHandlerContext ctx, FullHttpMessage msg) {
          final DefaultFullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK);
          final ProducerRecord&lt;String, ByteBuffer&gt; record = new ProducerRecord&lt;&gt;(
              "test",
              UUID.randomUUID().toString(),
              msg.content().nioBuffer()
          );
  
          producer.send(record);
  
          if (HttpUtil.isKeepAlive(msg)) {
              response.headers().set(CONNECTION, HttpHeaderValues.KEEP_ALIVE);
          }
  
          ctx.write(response).addListener(ChannelFutureListener.CLOSE);
      }
  }
  </code></pre>
  
  <p>The code is taken from the official documentation. However, sometimes I get <code>Request 'Post BidRequest' failed: j.u.c.TimeoutException: Request timeout after 60000 ms</code> exceptions in my load test.</p>
  
  <p>As far as I understood, this means that the connection was established between my load test instance and service instance but it took longer then 60 seconds to complete. What part of this simple program can block for so long?</p>
  
  <p>I've tuned Kafka producer: decreased it's timeout. I know that <code>send</code> can be blocking, so I've increased send buffer, but it didn't help.
  I've also increased <code>ulimits</code> for service user.
  I'm running on OpenJDK version 1.8.0_171 and <code>securerandom.source</code> is set to <code>file:/dev/urandom</code>, so the call to <code>randomUUID</code> should not block.</p>