KAFKA-2614; No more clients can connect after TooManyConnectionsException threshold (max.connections.per.ip) is reached

clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java

@@ -20,6 +20,7 @@
 
 import java.io.IOException;
 
+import java.net.InetAddress;
 import java.net.Socket;
 import java.nio.channels.SelectionKey;
 
@@ -108,14 +109,21 @@ public boolean hasSend() {
         return send != null;
     }
 
+    /**
+     * Returns the address to which this channel's socket is connected or `null` if the socket has never been connected.
+     *
+     * If the socket was connected prior to being closed, then this method will continue to return the
+     * connected address after the socket is closed.
+     */
+    public InetAddress socketAddress() {
+        return transportLayer.socketChannel().socket().getInetAddress();
+    }
+
     public String socketDescription() {
         Socket socket = transportLayer.socketChannel().socket();
-        if (socket == null)
-            return "[unconnected socket]";
-        else if (socket.getInetAddress() != null)
-            return socket.getInetAddress().toString();
-        else
+        if (socket.getInetAddress() == null)
             return socket.getLocalAddress().toString();
+        return socket.getInetAddress().toString();
     }
 
     public void setSend(Send send) {
@@ -132,7 +140,7 @@ public NetworkReceive read() throws IOException {
             receive = new NetworkReceive(maxReceiveSize, id);
         }
 
-        long x = receive(receive);
+        receive(receive);
         if (receive.complete()) {
             receive.payload().rewind();
             result = receive;

clients/src/main/java/org/apache/kafka/common/network/Selectable.java

@@ -50,7 +50,7 @@ public interface Selectable {
     /**
      * Close the connection identified by the given id
      */
-    public void close(String nodeId);
+    public void close(String id);
 
     /**
      * Queue the given request for sending in the subsequent {@link #poll(long) poll()} calls

clients/src/main/java/org/apache/kafka/common/network/Selector.java

@@ -207,10 +207,7 @@ public void close() {
      * @param send The request to send
      */
     public void send(Send send) {
-        KafkaChannel channel = channelForId(send.destination());
-        if (channel == null)
-            throw new IllegalStateException("channel is not connected");
-
+        KafkaChannel channel = channelOrFail(send.destination());
         try {
             channel.setSend(send);
         } catch (CancelledKeyException e) {
@@ -243,6 +240,7 @@ public void send(Send send) {
      * @throws IllegalArgumentException If `timeout` is negative
      * @throws IllegalStateException If a send is given for which we have no existing connection or for which there is
      *         already an in-progress send
+     * @throws InvalidReceiveException If invalid data is received
      */
     @Override
     public void poll(long timeout) throws IOException {
@@ -351,7 +349,7 @@ public List<String> connected() {
 
     @Override
     public void mute(String id) {
-        KafkaChannel channel = channelForId(id);
+        KafkaChannel channel = channelOrFail(id);
         mute(channel);
     }
 
@@ -361,7 +359,7 @@ private void mute(KafkaChannel channel) {
 
     @Override
     public void unmute(String id) {
-        KafkaChannel channel = channelForId(id);
+        KafkaChannel channel = channelOrFail(id);
         unmute(channel);
     }
 
@@ -433,8 +431,7 @@ private int select(long ms) throws IOException {
     }
 
     /**
-     * Begin closing this connection
-     * @param id channel id
+     * Close the connection identified by the given id
      */
     public void close(String id) {
         KafkaChannel channel = this.channels.get(id);
@@ -463,22 +460,34 @@ private void close(KafkaChannel channel) {
      */
     @Override
     public boolean isChannelReady(String id) {
-        KafkaChannel channel = channelForId(id);
+        KafkaChannel channel = this.channels.get(id);
+        if (channel == null)
+            return false;
         return channel.ready();
     }
 
-    /**
-     * Get the channel associated with this connection
-     * Exposing this to allow SocketServer get the Principal from the channel when creating a request
-     * without making Selector know about Principals
-     */
-    public KafkaChannel channelForId(String id) {
+    private KafkaChannel channelOrFail(String id) {
         KafkaChannel channel = this.channels.get(id);
         if (channel == null)
-            throw new IllegalStateException("Attempt to write to socket for which there is no open connection. Connection id " + id + " existing connections " + channels.keySet().toString());
+            throw new IllegalStateException("Attempt to retrieve channel for which there is no open connection. Connection id " + id + " existing connections " + channels.keySet().toString());
         return channel;
     }
 
+    /**
+     * Return the selector channels.
+     */
+    public List<KafkaChannel> channels() {
+        return new ArrayList<>(channels.values());
+    }
+
+    /**
+     * Return the channel associated with this connection or `null` if there is no channel associated with the
+     * connection.
+     */
+    public KafkaChannel channel(String id) {
+        return this.channels.get(id);
+    }
+
     /**
      * Get the channel associated with selectionKey
      */

clients/src/test/java/org/apache/kafka/common/network/SSLTransportLayerTest.java

@@ -418,9 +418,7 @@ private void waitForChannelClose(String node) throws IOException {
         boolean closed = false;
         for (int i = 0; i < 30; i++) {
             selector.poll(1000L);
-            try {
-                selector.channelForId(node);
-            } catch (IllegalStateException e) {
+            if (selector.channel(node) == null) {
                 closed = true;
                 break;
             }
@@ -581,7 +579,7 @@ public void run() {
                     newChannels.clear();
                     while (true) {
                         NetworkSend send = inflightSends.peek();
-                        if (send != null && !selector.channelForId(send.destination()).hasSend()) {
+                        if (send != null && !selector.channel(send.destination()).hasSend()) {
                             send = inflightSends.poll();
                             selector.send(send);
                         } else
@@ -590,7 +588,7 @@ public void run() {
                     List<NetworkReceive> completedReceives = selector.completedReceives();
                     for (NetworkReceive rcv : completedReceives) {
                         NetworkSend send = new NetworkSend(rcv.source(), rcv.payload());
-                        if (!selector.channelForId(send.destination()).hasSend())
+                        if (!selector.channel(send.destination()).hasSend())
                             selector.send(send);
                         else
                             inflightSends.add(send);

core/src/main/scala/kafka/cluster/BrokerEndPoint.scala

@@ -23,15 +23,27 @@ import kafka.common.KafkaException
 import org.apache.kafka.common.utils.Utils._
 
 object BrokerEndPoint {
-  def createBrokerEndPoint(brokerId: Int, connectionString: String): BrokerEndPoint = {
 
-    // BrokerEndPoint URI is host:port or [ipv6_host]:port
-    // Note that unlike EndPoint (or listener) this URI has no security information.
-    val uriParseExp = """\[?([0-9a-z\-.:]*)\]?:([0-9]+)""".r
+  private val uriParseExp = """\[?([0-9a-z\-.:]*)\]?:([0-9]+)""".r
 
+  /**
+   * BrokerEndPoint URI is host:port or [ipv6_host]:port
+   * Note that unlike EndPoint (or listener) this URI has no security information.
+   */
+  def parseHostPort(connectionString: String): Option[(String, Int)] = {
     connectionString match {
-      case uriParseExp(host, port) => new BrokerEndPoint(brokerId, host, port.toInt)
-      case _ => throw new KafkaException("Unable to parse " + connectionString + " to a broker endpoint")
+      case uriParseExp(host, port) => try Some(host, port.toInt) catch { case e: NumberFormatException => None }
+      case _ => None
+    }
+  }
+
+  /**
+   * BrokerEndPoint URI is host:port or [ipv6_host]:port
+   * Note that unlike EndPoint (or listener) this URI has no security information.
+   */
+  def createBrokerEndPoint(brokerId: Int, connectionString: String): BrokerEndPoint = {
+    parseHostPort(connectionString).map { case (host, port) => new BrokerEndPoint(brokerId, host, port) }.getOrElse {
+      throw new KafkaException("Unable to parse " + connectionString + " to a broker endpoint")
     }
   }