code: private void negotiate(final ChannelHandlerContext ctx) {
      if (controlSettings.hasNext()) {
        Map.Entry<String, String> setting = controlSettings.next();
  
        LOGGER.debug("Negotiating DCP Control {}: {}", setting.getKey(), setting.getValue());
        ByteBuf request = ctx.alloc().buffer();
        DcpControlRequest.init(request);
        DcpControlRequest.key(setting.getKey(), request);
        DcpControlRequest.value(Unpooled.copiedBuffer(setting.getValue(), CharsetUtil.UTF_8), request);
  
        ctx.writeAndFlush(request);
      } else {
        originalPromise().setSuccess();
        ctx.pipeline().remove(this);
        ctx.fireChannelActive();
        LOGGER.debug("Negotiated all DCP Control settings against Node {}", ctx.channel().remoteAddress());
      }
    }
docs: Helper method to walk the iterator and create a new request that defines which control param
  should be negotiated right now.