@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
      if (msg instanceof DataTransferEncryptorMessageProto) {
        DataTransferEncryptorMessageProto proto = (DataTransferEncryptorMessageProto) msg;
        check(proto);
        byte[] challenge = proto.getPayload().toByteArray();
        byte[] response = saslClient.evaluateChallenge(challenge);
        switch (step) {
          case 1: {
            List<CipherOption> cipherOptions = null;
            if (requestedQopContainsPrivacy()) {
              cipherOptions = getCipherOptions();
            }
            sendSaslMessage(ctx, response, cipherOptions);
            ctx.flush();
            step++;
            break;
          }
          case 2: {
            assert response == null;
            checkSaslComplete();
            CipherOption cipherOption =
              getCipherOption(proto, isNegotiatedQopPrivacy(), saslClient);
            ChannelPipeline p = ctx.pipeline();
            while (p.first() != null) {
              p.removeFirst();
            }
            if (cipherOption != null) {
              CryptoCodec codec = CryptoCodec.getInstance(conf, cipherOption.getCipherSuite());
              p.addLast(new EncryptHandler(codec, cipherOption.getInKey(), cipherOption.getInIv()),
                new DecryptHandler(codec, cipherOption.getOutKey(), cipherOption.getOutIv()));
            } else {
              if (useWrap()) {
                p.addLast(new SaslWrapHandler(saslClient),
                  new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4),
                  new SaslUnwrapHandler(saslClient));
              }
            }
            promise.trySuccess(null);
            break;
          }
          default:
            throw new IllegalArgumentException("Unrecognized negotiation step: " + step);
        }
      } else {
        ctx.fireChannelRead(msg);
      }
    }
Create a ByteString from byte array without copying (wrap), and then set it as the payload
for the builder.
@param builder builder for HDFS DataTransferEncryptorMessage.
@param payload byte array of payload.
hbase-asyncfs/src/main/java/org/apache/hadoop/hbase/io/asyncfs/FanOutOneBlockAsyncDFSOutputSaslHelper.java