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 ; }
Tries to start the server at the provided port .