code: public void close() {
          LOG.info("Close netty connection to {}", name());
          if (!beingClosed.compareAndSet(false, true)) {
              LOG.info("Netty client has been closed.");
              return;
          }
  
          if (!connectMyself) {
              unregisterMetrics();
          }
  
          Channel channel = channelRef.get();
          if (channel == null) {
              LOG.info("Channel {} has been closed already", name());
              return;
          }
  
          // wait for pendings to exit
          final long timeoutMilliSeconds = 10 * 1000;
          final long start = System.currentTimeMillis();
  
          LOG.info("Waiting for pending batchs to be sent with " + name() + "..., timeout: {}ms, pendings: {}",
                  timeoutMilliSeconds, pendings.get());
  
          while (pendings.get() != 0) {
              try {
                  long delta = System.currentTimeMillis() - start;
                  if (delta > timeoutMilliSeconds) {
                      LOG.error("Timeout when sending pending batchs with {}..., there are still {} pending batchs not sent",
                              name(), pendings.get());
                      break;
                  }
                  Thread.sleep(1000); // sleep 1s
              } catch (InterruptedException e) {
                  break;
              }
          }
  
          close_n_release();
      }
docs: gracefully close this client.
  <p/>
  We will send all existing requests, and then invoke close_n_release() method