docs: private boolean attemptToBind(final int port) throws Throwable {
  		log.debug("Attempting to start {} on port {}.", serverName, port);
  
  		this.queryExecutor = createQueryExecutor();
  		this.handler = initializeHandler();
  
  		final NettyBufferPool bufferPool = new NettyBufferPool(numEventLoopThreads);
  
  		final ThreadFactory threadFactory = new ThreadFactoryBuilder()
  				.setDaemon(true)
  				.setNameFormat("Flink " + serverName + " EventLoop Thread %d")
  				.build();
  
  		final NioEventLoopGroup nioGroup = new NioEventLoopGroup(numEventLoopThreads, threadFactory);
  
  		this.bootstrap = new ServerBootstrap()
  				.localAddress(bindAddress, port)
  				.group(nioGroup)
  				.channel(NioServerSocketChannel.class)
  				.option(ChannelOption.ALLOCATOR, bufferPool)
  				.childOption(ChannelOption.ALLOCATOR, bufferPool)
  				.childHandler(new ServerChannelInitializer<>(handler));
  
  		final int defaultHighWaterMark = 64 * 1024; // from DefaultChannelConfig (not exposed)
  		//noinspection ConstantConditions
  		// (ignore warning here to make this flexible in case the configuration values change)
  		if (LOW_WATER_MARK > defaultHighWaterMark) {
  			bootstrap.childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, HIGH_WATER_MARK);
  			bootstrap.childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, LOW_WATER_MARK);
  		} else { // including (newHighWaterMark < defaultLowWaterMark)
  			bootstrap.childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, LOW_WATER_MARK);
  			bootstrap.childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, HIGH_WATER_MARK);
  		}
  
  		try {
  			final ChannelFuture future = bootstrap.bind().sync();
  			if (future.isSuccess()) {
  				final InetSocketAddress localAddress = (InetSocketAddress) future.channel().localAddress();
  				serverAddress = new InetSocketAddress(localAddress.getAddress(), localAddress.getPort());
  				return true;
  			}
  
  			// the following throw is to bypass Netty's "optimization magic"
  			// and catch the bind exception.
  			// the exception is thrown by the sync() call above.
  
  			throw future.cause();
  		} catch (BindException e) {
  			log.debug("Failed to start {} on port {}: {}.", serverName, port, e.getMessage());
  			try {
  				// we shutdown the server but we reset the future every time because in
  				// case of failure to bind, we will call attemptToBind() here, and not resetting
  				// the flag will interfere with future shutdown attempts.
  
  				shutdownServer()
  						.whenComplete((ignoredV, ignoredT) -> serverShutdownFuture.getAndSet(null))
  						.get();
  			} catch (Exception r) {
  
  				// Here we were seeing this problem:
  				// https://github.com/netty/netty/issues/4357 if we do a get().
  				// this is why we now simply wait a bit so that everything is shut down.
  
  				log.warn("Problem while shutting down {}: {}", serverName, r.getMessage());
  			}
  		}
  		// any other type of exception we let it bubble up.
  		return false;
  	}