code: public void doReconnect() {
          if (channelRef.get() != null) {
              // if (channelRef.get().isWritable()) {
              // LOG.info("already exist a writable channel, give up reconnect, {}",
              // channelRef.get());
              // return;
              // }
              return;
          }
  
          if (isClosed()) {
              return;
          }
  
          if (isConnecting.getAndSet(true)) {
              LOG.info("Connect twice {}", name());
              return;
          }
  
          long sleepMs = getSleepTimeMs();
          LOG.info("Reconnect ... [{}], {}, sleep {}ms", retries.get(), name, sleepMs);
          ChannelFuture future = bootstrap.connect(remoteAddr);
          future.addListener(new ChannelFutureListener() {
              public void operationComplete(ChannelFuture future) throws Exception {
                  isConnecting.set(false);
                  Channel channel = future.getChannel();
                  if (future.isSuccess()) {
                      // do something else
                      LOG.info("Connection established, channel = :{}", channel);
                      setChannel(channel);
                      // handleResponse();
                      BATCH_THRESHOLD_WARN = ConfigExtension.getNettyBufferThresholdSize(stormConf);
  
                      // Check if any pending message
                      /*synchronized (writeLock) {
                          if (channel != null && messageBuffer.size() > 0) {
                              MessageBatch messageBatch = messageBuffer.drain();
                              flushRequest(channel, messageBatch);
                          } else {
                              LOG.warn("Failed to flush pending message after reconnecting, channel={}, messageBuffer.size={}",
                                      channel, messageBuffer.size());
                          }
                      }*/
                  } else {
                      if (!isClosed()) {
                          LOG.info("Failed to reconnect ... [{}], {}, channel = {}, cause = {}",
                                  retries.get(), name, channel, future.getCause());
                          reconnect();
                      }
                  }
              }
          });
          JStormUtils.sleepMs(sleepMs);
      }
docs: The function can't be synchronized, otherwise it will cause deadlock