private void configureNettyWatermarks(Configuration conf) {
    int watermarkLow =
      conf.getInt(CHANNEL_WRITABLE_LOW_WATERMARK_KEY, CHANNEL_WRITABLE_LOW_WATERMARK_DEFAULT);
    int watermarkHigh =
      conf.getInt(CHANNEL_WRITABLE_HIGH_WATERMARK_KEY, CHANNEL_WRITABLE_HIGH_WATERMARK_DEFAULT);
    int fatalThreshold =
      conf.getInt(CHANNEL_WRITABLE_FATAL_WATERMARK_KEY, CHANNEL_WRITABLE_FATAL_WATERMARK_DEFAULT);

    WriteBufferWaterMark oldWaterMark = writeBufferWaterMark;
    int oldFatalThreshold = writeBufferFatalThreshold;

    boolean disabled = false;
    if (watermarkHigh == 0 && watermarkLow == 0) {
      // if both are 0, use the netty default, which we will treat as "disabled".
      // when disabled, we won't manage autoRead in response to writability changes.
      writeBufferWaterMark = WriteBufferWaterMark.DEFAULT;
      disabled = true;
    } else {
      // netty checks pendingOutboundBytes < watermarkLow. It can never be less than 0, so set to
      // 1 to avoid confusing behavior.
      if (watermarkLow == 0) {
        LOG.warn(
          "Detected a {} value of 0, which is impossible to achieve "
            + "due to how netty evaluates these thresholds, setting to 1",
          CHANNEL_WRITABLE_LOW_WATERMARK_KEY);
        watermarkLow = 1;
      }

      // netty validates the watermarks and throws an exception if high < low, fail more gracefully
      // by disabling the watermarks and warning.
      if (watermarkHigh <= watermarkLow) {
        LOG.warn(
          "Detected {} value {}, lower than {} value {}. This will fail netty validation, "
            + "so disabling",
          CHANNEL_WRITABLE_HIGH_WATERMARK_KEY, watermarkHigh, CHANNEL_WRITABLE_LOW_WATERMARK_KEY,
          watermarkLow);
        writeBufferWaterMark = WriteBufferWaterMark.DEFAULT;
      } else {
        writeBufferWaterMark = new WriteBufferWaterMark(watermarkLow, watermarkHigh);
      }

      // only apply this check when watermark is enabled. this way we give the operator some
      // flexibility if they want to try enabling fatal threshold without backpressure.
      if (fatalThreshold > 0 && fatalThreshold <= watermarkHigh) {
        LOG.warn("Detected a {} value of {}, which is lower than the {} value of {}, ignoring.",
          CHANNEL_WRITABLE_FATAL_WATERMARK_KEY, fatalThreshold, CHANNEL_WRITABLE_HIGH_WATERMARK_KEY,
          watermarkHigh);
        fatalThreshold = 0;
      }
    }

    writeBufferFatalThreshold = fatalThreshold;

    if (
      oldWaterMark != null && (oldWaterMark.low() != writeBufferWaterMark.low()
        || oldWaterMark.high() != writeBufferWaterMark.high()
        || oldFatalThreshold != writeBufferFatalThreshold)
    ) {
      LOG.info("Updated netty outbound write buffer watermarks: low={}, high={}, fatal={}",
        disabled ? "disabled" : writeBufferWaterMark.low(),
        disabled ? "disabled" : writeBufferWaterMark.high(),
        writeBufferFatalThreshold <= 0 ? "disabled" : writeBufferFatalThreshold);
    }

    // update any existing channels
    for (Channel channel : allChannels) {
      channel.config().setWriteBufferWaterMark(writeBufferWaterMark);
      // if disabling watermark, set auto read to true in case channel had been exceeding
      // previous watermark
      if (disabled) {
        channel.config().setAutoRead(true);
      }
    }
  }
Fatal watermark for pending outbound bytes of a single netty channel. If the number of pending
outbound bytes exceeds this threshold, the connection will be forcibly closed so that memory
can be reclaimed. The client will have to re-establish a new connection and retry any in-flight
requests.
<p>
Note: must be higher than the high watermark, otherwise it's ignored.
hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcServer.java