Selector implements Selectable { @Override public void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize) throws IOException { if (this.channels.containsKey(id)) { throw new IllegalStateException("There is already a connection for id " + id); } SocketChannel socketChannel = SocketChannel.open(); socketChannel.configureBlocking(false); Socket socket = socketChannel.socket(); socket.setKeepAlive(true); if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE) { socket.setSendBufferSize(sendBufferSize); } if (receiveBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE) { socket.setReceiveBufferSize(receiveBufferSize); } socket.setTcpNoDelay(true); boolean connected; try { connected = socketChannel.connect(address); } catch (UnresolvedAddressException e) { socketChannel.close(); throw new IOException("Can't resolve address: " + address, e); } catch (IOException e) { socketChannel.close(); throw e; } SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_CONNECT); KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize); key.attach(channel); this.channels.put(id, channel); if (connected) { log.debug("Immediately connected to node {}", channel.id()); immediatelyConnectedKeys.add(key); key.interestOps(0); } } Selector(int maxReceiveSize, long connectionMaxIdleMs, Metrics metrics, Time time, String metricGrpPrefix, Map<String, String> metricTags, boolean metricsPerConnection,
        ChannelBuilder channelBuilder);  Selector(long connectionMaxIdleMS, Metrics metrics, Time time, String metricGrpPrefix, ChannelBuilder channelBuilder); @Override void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize); void register(String id, SocketChannel socketChannel); @Override void wakeup(); @Override void close(); void send(Send send); @Override void poll(long timeout); static void main(String[] args); @Override List<Send> completedSends(); @Override List<NetworkReceive> completedReceives(); @Override List<String> disconnected(); @Override List<String> connected(); void close(String id); @Override boolean isChannelReady(String id); List<KafkaChannel> channels(); KafkaChannel channel(String id); @Override void mute(String id); @Override void unmute(String id); @Override void muteAll(); @Override void unmuteAll();  }
@Test(expected = IOException.class) public void testNoRouteToHost() throws Exception { selector.connect("0", new InetSocketAddress("asdf.asdf.dsc", server.port), BUFFER_SIZE, BUFFER_SIZE); }
@Test public void testLargeMessageSequence() throws Exception { int bufferSize = 512 * 1024; String node = "0"; int reqs = 50; InetSocketAddress addr = new InetSocketAddress("localhost", server.port); connect(node, addr); String requestPrefix = TestUtils.randomString(bufferSize); sendAndReceive(node, requestPrefix, 0, reqs); }