kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jun...@apache.org
Subject [3/3] kafka git commit: kafka-1928; Move kafka.network over to using the network classes in org.apache.kafka.common.network; patched by Gwen Shapira; reviewed by Joel Koshy, Jay Kreps, Jiangjie Qin, Guozhang Wang and Jun Rao
Date Thu, 04 Jun 2015 04:40:46 GMT
kafka-1928; Move kafka.network over to using the network classes in org.apache.kafka.common.network; patched by Gwen Shapira; reviewed by Joel Koshy, Jay Kreps, Jiangjie Qin, Guozhang Wang and Jun Rao


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/78ba492e
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/78ba492e
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/78ba492e

Branch: refs/heads/trunk
Commit: 78ba492e3e70fd9db61bc82469371d04a8d6b762
Parents: d22987f
Author: Gwen Shapira <cshapi@gmail.com>
Authored: Wed Jun 3 21:40:35 2015 -0700
Committer: Jun Rao <junrao@gmail.com>
Committed: Wed Jun 3 21:40:35 2015 -0700

----------------------------------------------------------------------
 .../kafka/clients/ClusterConnectionStates.java  |  78 ++---
 .../kafka/clients/CommonClientConfigs.java      |   2 +
 .../apache/kafka/clients/InFlightRequests.java  |  18 +-
 .../org/apache/kafka/clients/KafkaClient.java   |   8 +-
 .../org/apache/kafka/clients/NetworkClient.java |  59 ++--
 .../kafka/clients/consumer/ConsumerConfig.java  |  11 +-
 .../kafka/clients/consumer/KafkaConsumer.java   |   3 +-
 .../clients/consumer/internals/Coordinator.java |   4 +-
 .../clients/consumer/internals/Fetcher.java     |  10 +-
 .../kafka/clients/producer/KafkaProducer.java   |   3 +-
 .../kafka/clients/producer/ProducerConfig.java  |   6 +-
 .../clients/producer/internals/Sender.java      |   6 +-
 .../main/java/org/apache/kafka/common/Node.java |  10 +
 .../kafka/common/network/ByteBufferReceive.java |  10 +-
 .../kafka/common/network/ByteBufferSend.java    |  20 +-
 .../common/network/InvalidReceiveException.java |  30 ++
 .../apache/kafka/common/network/MultiSend.java  | 100 ++++++
 .../kafka/common/network/NetworkReceive.java    |  59 +++-
 .../kafka/common/network/NetworkSend.java       |   2 +-
 .../apache/kafka/common/network/Receive.java    |   8 +-
 .../apache/kafka/common/network/Selectable.java |  16 +-
 .../apache/kafka/common/network/Selector.java   | 230 ++++++++-----
 .../org/apache/kafka/common/network/Send.java   |  18 +-
 .../kafka/common/requests/RequestSend.java      |   2 +-
 .../kafka/common/requests/ResponseSend.java     |  41 +++
 .../org/apache/kafka/clients/MockClient.java    |   6 +-
 .../apache/kafka/clients/NetworkClientTest.java |   8 +-
 .../kafka/common/network/SelectorTest.java      |  86 ++---
 .../org/apache/kafka/test/MockSelector.java     |  25 +-
 core/src/main/scala/kafka/Kafka.scala           |  12 +-
 .../kafka/admin/ConsumerGroupCommand.scala      |   2 +-
 .../kafka/api/ConsumerMetadataRequest.scala     |   7 +-
 .../kafka/api/ControlledShutdownRequest.scala   |   9 +-
 .../src/main/scala/kafka/api/FetchRequest.scala |   2 +-
 .../main/scala/kafka/api/FetchResponse.scala    |  73 ++--
 .../scala/kafka/api/LeaderAndIsrRequest.scala   |  12 +-
 .../scala/kafka/api/OffsetCommitRequest.scala   |  10 +-
 .../scala/kafka/api/OffsetFetchRequest.scala    |  15 +-
 .../main/scala/kafka/api/OffsetRequest.scala    |   7 +-
 .../main/scala/kafka/api/ProducerRequest.scala  |   7 +-
 core/src/main/scala/kafka/api/RequestKeys.scala |   4 +-
 .../scala/kafka/api/StopReplicaRequest.scala    |   4 +-
 .../scala/kafka/api/TopicMetadataRequest.scala  |   8 +-
 .../scala/kafka/api/UpdateMetadataRequest.scala |   4 +-
 .../main/scala/kafka/client/ClientUtils.scala   |   2 +-
 .../scala/kafka/consumer/SimpleConsumer.scala   |  19 +-
 .../consumer/ZookeeperConsumerConnector.scala   |   4 +-
 .../controller/ControllerChannelManager.scala   |  11 +-
 .../kafka/javaapi/TopicMetadataRequest.scala    |   7 +-
 .../scala/kafka/network/BlockingChannel.scala   |  21 +-
 .../network/BoundedByteBufferReceive.scala      |  90 -----
 .../kafka/network/BoundedByteBufferSend.scala   |  71 ----
 .../scala/kafka/network/ByteBufferSend.scala    |  40 ---
 core/src/main/scala/kafka/network/Handler.scala |   6 +-
 .../scala/kafka/network/RequestChannel.scala    |  35 +-
 .../kafka/network/RequestOrResponseSend.scala   |  57 ++++
 .../main/scala/kafka/network/SocketServer.scala | 334 ++++++++-----------
 .../main/scala/kafka/network/Transmission.scala | 122 -------
 .../scala/kafka/producer/SyncProducer.scala     |  19 +-
 .../src/main/scala/kafka/server/KafkaApis.scala |  44 +--
 .../main/scala/kafka/server/KafkaConfig.scala   |  56 +++-
 .../main/scala/kafka/server/KafkaServer.scala   |  33 +-
 .../scala/kafka/server/MessageSetSend.scala     |  71 ----
 .../kafka/tools/ConsumerOffsetChecker.scala     |   2 +-
 .../scala/other/kafka/TestOffsetManager.scala   |   6 +-
 .../test/scala/unit/kafka/KafkaConfigTest.scala |  17 +-
 .../unit/kafka/network/SocketServerTest.scala   |  41 +--
 .../kafka/server/KafkaConfigConfigDefTest.scala |   8 +
 68 files changed, 1075 insertions(+), 1096 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java b/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java
index da76cc2..9ebda5e 100644
--- a/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java
+++ b/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java
@@ -21,22 +21,22 @@ import java.util.Map;
  */
 final class ClusterConnectionStates {
     private final long reconnectBackoffMs;
-    private final Map<Integer, NodeConnectionState> nodeState;
+    private final Map<String, NodeConnectionState> nodeState;
 
     public ClusterConnectionStates(long reconnectBackoffMs) {
         this.reconnectBackoffMs = reconnectBackoffMs;
-        this.nodeState = new HashMap<Integer, NodeConnectionState>();
+        this.nodeState = new HashMap<String, NodeConnectionState>();
     }
 
     /**
-     * Return true iff we can currently initiate a new connection to the given node. This will be the case if we are not
+     * Return true iff we can currently initiate a new connection. This will be the case if we are not
      * connected and haven't been connected for at least the minimum reconnection backoff period.
-     * @param node The node id to check
+     * @param id The connection id to check
      * @param now The current time in MS
      * @return true if we can initiate a new connection
      */
-    public boolean canConnect(int node, long now) {
-        NodeConnectionState state = nodeState.get(node);
+    public boolean canConnect(String id, long now) {
+        NodeConnectionState state = nodeState.get(id);
         if (state == null)
             return true;
         else
@@ -45,11 +45,11 @@ final class ClusterConnectionStates {
 
     /**
      * Return true if we are disconnected from the given node and can't re-establish a connection yet
-     * @param node The node to check
+     * @param id The connection to check
      * @param now The current time in ms
      */
-    public boolean isBlackedOut(int node, long now) {
-        NodeConnectionState state = nodeState.get(node);
+    public boolean isBlackedOut(String id, long now) {
+        NodeConnectionState state = nodeState.get(id);
         if (state == null)
             return false;
         else
@@ -60,11 +60,11 @@ final class ClusterConnectionStates {
      * Returns the number of milliseconds to wait, based on the connection state, before attempting to send data. When
      * disconnected, this respects the reconnect backoff time. When connecting or connected, this handles slow/stalled
      * connections.
-     * @param node The node to check
+     * @param id The connection to check
      * @param now The current time in ms
      */
-    public long connectionDelay(int node, long now) {
-        NodeConnectionState state = nodeState.get(node);
+    public long connectionDelay(String id, long now) {
+        NodeConnectionState state = nodeState.get(id);
         if (state == null) return 0;
         long timeWaited = now - state.lastConnectAttemptMs;
         if (state.state == ConnectionState.DISCONNECTED) {
@@ -77,67 +77,67 @@ final class ClusterConnectionStates {
     }
 
     /**
-     * Enter the connecting state for the given node.
-     * @param node The id of the node we are connecting to
+     * Enter the connecting state for the given connection.
+     * @param id The id of the connection
      * @param now The current time.
      */
-    public void connecting(int node, long now) {
-        nodeState.put(node, new NodeConnectionState(ConnectionState.CONNECTING, now));
+    public void connecting(String id, long now) {
+        nodeState.put(id, new NodeConnectionState(ConnectionState.CONNECTING, now));
     }
 
     /**
-     * Return true iff we have a connection to the give node
-     * @param node The id of the node to check
+     * Return true iff a specific connection is connected
+     * @param id The id of the connection to check
      */
-    public boolean isConnected(int node) {
-        NodeConnectionState state = nodeState.get(node);
+    public boolean isConnected(String id) {
+        NodeConnectionState state = nodeState.get(id);
         return state != null && state.state == ConnectionState.CONNECTED;
     }
 
     /**
-     * Return true iff we are in the process of connecting to the given node
-     * @param node The id of the node
+     * Return true iff we are in the process of connecting
+     * @param id The id of the connection
      */
-    public boolean isConnecting(int node) {
-        NodeConnectionState state = nodeState.get(node);
+    public boolean isConnecting(String id) {
+        NodeConnectionState state = nodeState.get(id);
         return state != null && state.state == ConnectionState.CONNECTING;
     }
 
     /**
-     * Enter the connected state for the given node
-     * @param node The node we have connected to
+     * Enter the connected state for the given connection
+     * @param id The connection identifier
      */
-    public void connected(int node) {
-        NodeConnectionState nodeState = nodeState(node);
+    public void connected(String id) {
+        NodeConnectionState nodeState = nodeState(id);
         nodeState.state = ConnectionState.CONNECTED;
     }
 
     /**
      * Enter the disconnected state for the given node
-     * @param node The node we have disconnected from
+     * @param id The connection we have disconnected
      */
-    public void disconnected(int node) {
-        NodeConnectionState nodeState = nodeState(node);
+    public void disconnected(String id) {
+        NodeConnectionState nodeState = nodeState(id);
         nodeState.state = ConnectionState.DISCONNECTED;
     }
     
     /**
-     * Get the state of our connection to the given node
-     * @param node The id of the node
+     * Get the state of a given connection
+     * @param id The id of the connection
      * @return The state of our connection
      */
-    public ConnectionState connectionState(int node) {
-        return nodeState(node).state;
+    public ConnectionState connectionState(String id) {
+        return nodeState(id).state;
     }
     
     /**
      * Get the state of a given node
-     * @param node The node to fetch the state for
+     * @param id The connection to fetch the state for
      */
-    private NodeConnectionState nodeState(int node) {
-        NodeConnectionState state = this.nodeState.get(node);
+    private NodeConnectionState nodeState(String id) {
+        NodeConnectionState state = this.nodeState.get(id);
         if (state == null)
-            throw new IllegalStateException("No entry found for node " + node);
+            throw new IllegalStateException("No entry found for connection " + id);
         return state;
     }
     

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java
index cf32e4e..2c421f4 100644
--- a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java
+++ b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java
@@ -55,4 +55,6 @@ public class CommonClientConfigs {
     public static final String METRIC_REPORTER_CLASSES_CONFIG = "metric.reporters";
     public static final String METRIC_REPORTER_CLASSES_DOC = "A list of classes to use as metrics reporters. Implementing the <code>MetricReporter</code> interface allows plugging in classes that will be notified of new metric creation. The JmxReporter is always included to register JMX statistics.";
 
+    public static final String CONNECTIONS_MAX_IDLE_MS_CONFIG = "connections.max.idle.ms";
+    public static final String CONNECTIONS_MAX_IDLE_MS_DOC = "Close idle connections after the number of milliseconds specified by this config.";
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java b/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java
index 936487b..15d00d4 100644
--- a/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java
+++ b/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java
@@ -24,14 +24,14 @@ import java.util.Map;
 final class InFlightRequests {
 
     private final int maxInFlightRequestsPerConnection;
-    private final Map<Integer, Deque<ClientRequest>> requests = new HashMap<Integer, Deque<ClientRequest>>();
+    private final Map<String, Deque<ClientRequest>> requests = new HashMap<String, Deque<ClientRequest>>();
 
     public InFlightRequests(int maxInFlightRequestsPerConnection) {
         this.maxInFlightRequestsPerConnection = maxInFlightRequestsPerConnection;
     }
 
     /**
-     * Add the given request to the queue for the node it was directed to
+     * Add the given request to the queue for the connection it was directed to
      */
     public void add(ClientRequest request) {
         Deque<ClientRequest> reqs = this.requests.get(request.request().destination());
@@ -45,7 +45,7 @@ final class InFlightRequests {
     /**
      * Get the request queue for the given node
      */
-    private Deque<ClientRequest> requestQueue(int node) {
+    private Deque<ClientRequest> requestQueue(String node) {
         Deque<ClientRequest> reqs = requests.get(node);
         if (reqs == null || reqs.isEmpty())
             throw new IllegalStateException("Response from server for which there are no in-flight requests.");
@@ -55,7 +55,7 @@ final class InFlightRequests {
     /**
      * Get the oldest request (the one that that will be completed next) for the given node
      */
-    public ClientRequest completeNext(int node) {
+    public ClientRequest completeNext(String node) {
         return requestQueue(node).pollLast();
     }
 
@@ -63,7 +63,7 @@ final class InFlightRequests {
      * Get the last request we sent to the given node (but don't remove it from the queue)
      * @param node The node id
      */
-    public ClientRequest lastSent(int node) {
+    public ClientRequest lastSent(String node) {
         return requestQueue(node).peekFirst();
     }
 
@@ -72,7 +72,7 @@ final class InFlightRequests {
      * @param node The node the request was sent to
      * @return The request
      */
-    public ClientRequest completeLastSent(int node) {
+    public ClientRequest completeLastSent(String node) {
         return requestQueue(node).pollFirst();
     }
 
@@ -82,7 +82,7 @@ final class InFlightRequests {
      * @param node Node in question
      * @return true iff we have no requests still being sent to the given node
      */
-    public boolean canSendMore(int node) {
+    public boolean canSendMore(String node) {
         Deque<ClientRequest> queue = requests.get(node);
         return queue == null || queue.isEmpty() ||
                (queue.peekFirst().request().completed() && queue.size() < this.maxInFlightRequestsPerConnection);
@@ -93,7 +93,7 @@ final class InFlightRequests {
      * @param node The node
      * @return The request count.
      */
-    public int inFlightRequestCount(int node) {
+    public int inFlightRequestCount(String node) {
         Deque<ClientRequest> queue = requests.get(node);
         return queue == null ? 0 : queue.size();
     }
@@ -114,7 +114,7 @@ final class InFlightRequests {
      * @param node The node
      * @return All the in-flight requests for that node that have been removed
      */
-    public Iterable<ClientRequest> clearAll(int node) {
+    public Iterable<ClientRequest> clearAll(String node) {
         Deque<ClientRequest> reqs = requests.get(node);
         if (reqs == null) {
             return Collections.emptyList();

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java b/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
index 1311f85..7ab2503 100644
--- a/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
@@ -81,13 +81,13 @@ public interface KafkaClient extends Closeable {
     public List<ClientResponse> poll(long timeout, long now);
 
     /**
-     * Complete all in-flight requests for a given node
+     * Complete all in-flight requests for a given connection
      * 
-     * @param node The node to complete requests for
+     * @param id The connection to complete requests for
      * @param now The current time in ms
      * @return All requests that complete during this time period.
      */
-    public List<ClientResponse> completeAll(int node, long now);
+    public List<ClientResponse> completeAll(String id, long now);
 
     /**
      * Complete all in-flight requests
@@ -117,7 +117,7 @@ public interface KafkaClient extends Closeable {
      * 
      * @param nodeId The id of the node
      */
-    public int inFlightRequestCount(int nodeId);
+    public int inFlightRequestCount(String nodeId);
 
     /**
      * Generate a request header for the next request

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
index 435fbb5..48fe796 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -22,8 +22,8 @@ import java.util.Set;
 import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.network.NetworkReceive;
-import org.apache.kafka.common.network.NetworkSend;
 import org.apache.kafka.common.network.Selectable;
+import org.apache.kafka.common.network.Send;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ProtoUtils;
 import org.apache.kafka.common.protocol.types.Struct;
@@ -111,7 +111,7 @@ public class NetworkClient implements KafkaClient {
         if (isReady(node, now))
             return true;
 
-        if (connectionStates.canConnect(node.id(), now))
+        if (connectionStates.canConnect(node.idString(), now))
             // if we are interested in sending to a node and we don't have a connection to it, initiate one
             initiateConnect(node, now);
 
@@ -129,7 +129,7 @@ public class NetworkClient implements KafkaClient {
      */
     @Override
     public long connectionDelay(Node node, long now) {
-        return connectionStates.connectionDelay(node.id(), now);
+        return connectionStates.connectionDelay(node.idString(), now);
     }
 
     /**
@@ -142,7 +142,7 @@ public class NetworkClient implements KafkaClient {
      */
     @Override
     public boolean connectionFailed(Node node) {
-        return connectionStates.connectionState(node.id()).equals(ConnectionState.DISCONNECTED);
+        return connectionStates.connectionState(node.idString()).equals(ConnectionState.DISCONNECTED);
     }
 
     /**
@@ -154,7 +154,7 @@ public class NetworkClient implements KafkaClient {
      */
     @Override
     public boolean isReady(Node node, long now) {
-        int nodeId = node.id();
+        String nodeId = node.idString();
         if (!this.metadataFetchInProgress && this.metadata.timeToNextUpdate(now) == 0)
             // if we need to update our metadata now declare all requests unready to make metadata requests first
             // priority
@@ -165,11 +165,11 @@ public class NetworkClient implements KafkaClient {
     }
 
     /**
-     * Are we connected and ready and able to send more requests to the given node?
+     * Are we connected and ready and able to send more requests to the given connection?
      * 
      * @param node The node
      */
-    private boolean isSendable(int node) {
+    private boolean isSendable(String node) {
         return connectionStates.isConnected(node) && inFlightRequests.canSendMore(node);
     }
 
@@ -179,7 +179,7 @@ public class NetworkClient implements KafkaClient {
      * @param node The node to check
      * @return The connection state
      */
-    public ConnectionState connectionState(int node) {
+    public ConnectionState connectionState(String node) {
         return connectionStates.connectionState(node);
     }
 
@@ -190,7 +190,7 @@ public class NetworkClient implements KafkaClient {
      */
     @Override
     public void send(ClientRequest request) {
-        int nodeId = request.request().destination();
+        String nodeId = request.request().destination();
         if (!isSendable(nodeId))
             throw new IllegalStateException("Attempt to send a request to node " + nodeId + " which is not ready.");
 
@@ -252,7 +252,7 @@ public class NetworkClient implements KafkaClient {
      * @return All the collected responses
      */
     @Override
-    public List<ClientResponse> completeAll(int node, long now) {
+    public List<ClientResponse> completeAll(String node, long now) {
         try {
             this.selector.muteAll();
             this.selector.unmute(node);
@@ -288,8 +288,8 @@ public class NetworkClient implements KafkaClient {
      * Get the number of in-flight requests for a given node
      */
     @Override
-    public int inFlightRequestCount(int nodeId) {
-        return this.inFlightRequests.inFlightRequestCount(nodeId);
+    public int inFlightRequestCount(String node) {
+        return this.inFlightRequests.inFlightRequestCount(node);
     }
 
     /**
@@ -334,11 +334,11 @@ public class NetworkClient implements KafkaClient {
         for (int i = 0; i < nodes.size(); i++) {
             int idx = Utils.abs((this.nodeIndexOffset + i) % nodes.size());
             Node node = nodes.get(idx);
-            int currInflight = this.inFlightRequests.inFlightRequestCount(node.id());
-            if (currInflight == 0 && this.connectionStates.isConnected(node.id())) {
+            int currInflight = this.inFlightRequests.inFlightRequestCount(node.idString());
+            if (currInflight == 0 && this.connectionStates.isConnected(node.idString())) {
                 // if we find an established connection with no in-flight requests we can stop right away
                 return node;
-            } else if (!this.connectionStates.isBlackedOut(node.id(), now) && currInflight < inflight) {
+            } else if (!this.connectionStates.isBlackedOut(node.idString(), now) && currInflight < inflight) {
                 // otherwise if this is the best we have found so far, record that
                 inflight = currInflight;
                 found = node;
@@ -355,7 +355,7 @@ public class NetworkClient implements KafkaClient {
      */
     private void handleCompletedSends(List<ClientResponse> responses, long now) {
         // if no response is expected then when the send is completed, return it
-        for (NetworkSend send : this.selector.completedSends()) {
+        for (Send send : this.selector.completedSends()) {
             ClientRequest request = this.inFlightRequests.lastSent(send.destination());
             if (!request.expectResponse()) {
                 this.inFlightRequests.completeLastSent(send.destination());
@@ -372,7 +372,7 @@ public class NetworkClient implements KafkaClient {
      */
     private void handleCompletedReceives(List<ClientResponse> responses, long now) {
         for (NetworkReceive receive : this.selector.completedReceives()) {
-            int source = receive.source();
+            String source = receive.source();
             ClientRequest req = inFlightRequests.completeNext(source);
             ResponseHeader header = ResponseHeader.parse(receive.payload());
             short apiKey = req.request().header().apiKey();
@@ -412,7 +412,7 @@ public class NetworkClient implements KafkaClient {
      * @param now The current time
      */
     private void handleDisconnections(List<ClientResponse> responses, long now) {
-        for (int node : this.selector.disconnected()) {
+        for (String node : this.selector.disconnected()) {
             connectionStates.disconnected(node);
             log.debug("Node {} disconnected.", node);
             for (ClientRequest request : this.inFlightRequests.clearAll(node)) {
@@ -433,9 +433,9 @@ public class NetworkClient implements KafkaClient {
      * Record any newly completed connections
      */
     private void handleConnections() {
-        for (Integer id : this.selector.connected()) {
-            log.debug("Completed connection to node {}", id);
-            this.connectionStates.connected(id);
+        for (String node : this.selector.connected()) {
+            log.debug("Completed connection to node {}", node);
+            this.connectionStates.connected(node);
         }
     }
 
@@ -451,7 +451,7 @@ public class NetworkClient implements KafkaClient {
     /**
      * Create a metadata request for the given topics
      */
-    private ClientRequest metadataRequest(long now, int node, Set<String> topics) {
+    private ClientRequest metadataRequest(long now, String node, Set<String> topics) {
         MetadataRequest metadata = new MetadataRequest(new ArrayList<String>(topics));
         RequestSend send = new RequestSend(node, nextRequestHeader(ApiKeys.METADATA), metadata.toStruct());
         return new ClientRequest(now, true, send, null);
@@ -470,15 +470,17 @@ public class NetworkClient implements KafkaClient {
             this.lastNoNodeAvailableMs = now;
             return;
         }
+        String nodeConnectionId = node.idString();
 
-        if (connectionStates.isConnected(node.id()) && inFlightRequests.canSendMore(node.id())) {
+
+        if (connectionStates.isConnected(nodeConnectionId) && inFlightRequests.canSendMore(nodeConnectionId)) {
             Set<String> topics = metadata.topics();
             this.metadataFetchInProgress = true;
-            ClientRequest metadataRequest = metadataRequest(now, node.id(), topics);
+            ClientRequest metadataRequest = metadataRequest(now, nodeConnectionId, topics);
             log.debug("Sending metadata request {} to node {}", metadataRequest, node.id());
             this.selector.send(metadataRequest.request());
             this.inFlightRequests.add(metadataRequest);
-        } else if (connectionStates.canConnect(node.id(), now)) {
+        } else if (connectionStates.canConnect(nodeConnectionId, now)) {
             // we don't have a connection to this node right now, make one
             log.debug("Initialize connection to node {} for sending metadata request", node.id());
             initiateConnect(node, now);
@@ -497,16 +499,17 @@ public class NetworkClient implements KafkaClient {
      * Initiate a connection to the given node
      */
     private void initiateConnect(Node node, long now) {
+        String nodeConnectionId = node.idString();
         try {
             log.debug("Initiating connection to node {} at {}:{}.", node.id(), node.host(), node.port());
-            this.connectionStates.connecting(node.id(), now);
-            selector.connect(node.id(),
+            this.connectionStates.connecting(nodeConnectionId, now);
+            selector.connect(nodeConnectionId,
                              new InetSocketAddress(node.host(), node.port()),
                              this.socketSendBuffer,
                              this.socketReceiveBuffer);
         } catch (IOException e) {
             /* attempt failed, we'll try again after the backoff */
-            connectionStates.disconnected(node.id());
+            connectionStates.disconnected(nodeConnectionId);
             /* maybe the problem is our metadata, update it */
             metadata.requestUpdate();
             log.debug("Error connecting to node {} at {}:{}:", node.id(), node.host(), node.port(), e);

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
index bdff518..1e90524 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
@@ -156,6 +156,9 @@ public class ConsumerConfig extends AbstractConfig {
     public static final String VALUE_DESERIALIZER_CLASS_CONFIG = "value.deserializer";
     private static final String VALUE_DESERIALIZER_CLASS_DOC = "Deserializer class for value that implements the <code>Deserializer</code> interface.";
 
+    /** <code>connections.max.idle.ms</code> */
+    public static final String CONNECTIONS_MAX_IDLE_MS_CONFIG = CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG;
+
 
     static {
         CONFIG = new ConfigDef().define(BOOTSTRAP_SERVERS_CONFIG,
@@ -277,7 +280,13 @@ public class ConsumerConfig extends AbstractConfig {
                                 .define(VALUE_DESERIALIZER_CLASS_CONFIG,
                                         Type.CLASS,
                                         Importance.HIGH,
-                                        VALUE_DESERIALIZER_CLASS_DOC);
+                                        VALUE_DESERIALIZER_CLASS_DOC)
+                                /* default is set to be a bit lower than the server default (10 min), to avoid both client and server closing connection at same time */
+                                .define(CONNECTIONS_MAX_IDLE_MS_CONFIG,
+                                        Type.LONG,
+                                        9 * 60 * 1000,
+                                        Importance.MEDIUM,
+                                        CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_DOC);
     }
 
     public static Map<String, Object> addDeserializerToConfig(Map<String, Object> configs,

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
index d301be4..d1d1ec1 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
@@ -472,7 +472,8 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
             String metricGrpPrefix = "consumer";
             Map<String, String> metricsTags = new LinkedHashMap<String, String>();
             metricsTags.put("client-id", clientId);
-            this.client = new NetworkClient(new Selector(metrics, time, metricGrpPrefix, metricsTags),
+            this.client = new NetworkClient(
+                    new Selector(config.getLong(ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG), metrics, time, metricGrpPrefix, metricsTags),
                     this.metadata,
                     clientId,
                     100, // a fixed large enough value will suffice

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Coordinator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Coordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Coordinator.java
index fac7995..c1496a0 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Coordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Coordinator.java
@@ -445,7 +445,7 @@ public final class Coordinator {
         log.debug("Issuing consumer metadata request to broker {}", node.id());
 
         ConsumerMetadataRequest request = new ConsumerMetadataRequest(this.groupId);
-        RequestSend send = new RequestSend(node.id(),
+        RequestSend send = new RequestSend(node.idString(),
             this.client.nextRequestHeader(ApiKeys.CONSUMER_METADATA),
             request.toStruct());
         long now = time.milliseconds();
@@ -464,7 +464,7 @@ public final class Coordinator {
         log.debug("Issuing request ({}: {}) to coordinator {}", api, request, this.consumerCoordinator.id());
 
         RequestHeader header = this.client.nextRequestHeader(api);
-        RequestSend send = new RequestSend(this.consumerCoordinator.id(), header, request);
+        RequestSend send = new RequestSend(this.consumerCoordinator.idString(), header, request);
         return new ClientRequest(now, true, send, handler);
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
index c5e577f..56281ee 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
@@ -124,7 +124,7 @@ public class Fetcher<K, V> {
      */
     public void initFetches(Cluster cluster, long now) {
         for (ClientRequest request : createFetchRequests(cluster)) {
-            Node node = cluster.nodeById(request.request().destination());
+            Node node = cluster.nodeById(Integer.parseInt(request.request().destination()));
             if (client.ready(node, now)) {
                 log.trace("Initiating fetch to node {}: {}", node.id(), request);
                 client.send(request);
@@ -209,12 +209,12 @@ public class Fetcher<K, V> {
             } else if (this.client.ready(info.leader(), now)) {
                 Node node = info.leader();
                 ListOffsetRequest request = new ListOffsetRequest(-1, partitions);
-                RequestSend send = new RequestSend(node.id(),
+                RequestSend send = new RequestSend(node.idString(),
                     this.client.nextRequestHeader(ApiKeys.LIST_OFFSETS),
                     request.toStruct());
                 ClientRequest clientRequest = new ClientRequest(now, true, send, null);
                 this.client.send(clientRequest);
-                List<ClientResponse> responses = this.client.completeAll(node.id(), now);
+                List<ClientResponse> responses = this.client.completeAll(node.idString(), now);
                 if (responses.isEmpty())
                     throw new IllegalStateException("This should not happen.");
                 ClientResponse response = responses.get(responses.size() - 1);
@@ -258,7 +258,7 @@ public class Fetcher<K, V> {
         for (TopicPartition partition : subscriptions.assignedPartitions()) {
             Node node = cluster.leaderFor(partition);
             // if there is a leader and no in-flight requests, issue a new fetch
-            if (node != null && this.client.inFlightRequestCount(node.id()) == 0) {
+            if (node != null && this.client.inFlightRequestCount(node.idString()) == 0) {
                 Map<TopicPartition, FetchRequest.PartitionData> fetch = fetchable.get(node.id());
                 if (fetch == null) {
                     fetch = new HashMap<TopicPartition, FetchRequest.PartitionData>();
@@ -274,7 +274,7 @@ public class Fetcher<K, V> {
         for (Map.Entry<Integer, Map<TopicPartition, FetchRequest.PartitionData>> entry : fetchable.entrySet()) {
             int nodeId = entry.getKey();
             final FetchRequest fetch = new FetchRequest(this.maxWaitMs, this.minBytes, entry.getValue());
-            RequestSend send = new RequestSend(nodeId, this.client.nextRequestHeader(ApiKeys.FETCH), fetch.toStruct());
+            RequestSend send = new RequestSend(Integer.toString(nodeId), this.client.nextRequestHeader(ApiKeys.FETCH), fetch.toStruct());
             RequestCompletionHandler handler = new RequestCompletionHandler() {
                 public void onComplete(ClientResponse response) {
                     handleFetchResponse(response, fetch);

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
index ded19d8..5a37580 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -227,7 +227,8 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
             List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(config.getList(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
             this.metadata.update(Cluster.bootstrap(addresses), time.milliseconds());
 
-            NetworkClient client = new NetworkClient(new Selector(this.metrics, time, "producer", metricTags),
+            NetworkClient client = new NetworkClient(
+                    new Selector(config.getLong(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG), this.metrics, time, "producer", metricTags),
                     this.metadata,
                     clientId,
                     config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION),

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
index 023bd2e..aa26420 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
@@ -169,11 +169,13 @@ public class ProducerConfig extends AbstractConfig {
     public static final String VALUE_SERIALIZER_CLASS_CONFIG = "value.serializer";
     private static final String VALUE_SERIALIZER_CLASS_DOC = "Serializer class for value that implements the <code>Serializer</code> interface.";
 
+    /** <code>connections.max.idle.ms</code> */
+    public static final String CONNECTIONS_MAX_IDLE_MS_CONFIG = CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG;
+
     /** <code>partitioner.class</code> */
     public static final String PARTITIONER_CLASS_CONFIG = "partitioner.class";
     private static final String PARTITIONER_CLASS_DOC = "Partitioner class that implements the <code>Partitioner</code> interface.";
 
-
     static {
         CONFIG = new ConfigDef().define(BOOTSTRAP_SERVERS_CONFIG, Type.LIST, Importance.HIGH, CommonClientConfigs.BOOSTRAP_SERVERS_DOC)
                                 .define(BUFFER_MEMORY_CONFIG, Type.LONG, 32 * 1024 * 1024L, atLeast(0L), Importance.HIGH, BUFFER_MEMORY_DOC)
@@ -223,6 +225,8 @@ public class ProducerConfig extends AbstractConfig {
                                         MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_DOC)
                                 .define(KEY_SERIALIZER_CLASS_CONFIG, Type.CLASS, Importance.HIGH, KEY_SERIALIZER_CLASS_DOC)
                                 .define(VALUE_SERIALIZER_CLASS_CONFIG, Type.CLASS, Importance.HIGH, VALUE_SERIALIZER_CLASS_DOC)
+                                /* default is set to be a bit lower than the server default (10 min), to avoid both client and server closing connection at same time */
+                                .define(CONNECTIONS_MAX_IDLE_MS_CONFIG, Type.LONG, 9 * 60 * 1000, Importance.MEDIUM, CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_DOC)
                                 .define(PARTITIONER_CLASS_CONFIG, Type.CLASS, "org.apache.kafka.clients.producer.internals.DefaultPartitioner", Importance.MEDIUM, PARTITIONER_CLASS_DOC);
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
index 1e943d6..07e65d4 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
@@ -320,7 +320,7 @@ public class Sender implements Runnable {
             recordsByPartition.put(tp, batch);
         }
         ProduceRequest request = new ProduceRequest(acks, timeout, produceRecordsByPartition);
-        RequestSend send = new RequestSend(destination,
+        RequestSend send = new RequestSend(Integer.toString(destination),
                                            this.client.nextRequestHeader(ApiKeys.PRODUCE),
                                            request.toStruct());
         RequestCompletionHandler callback = new RequestCompletionHandler() {
@@ -505,10 +505,10 @@ public class Sender implements Runnable {
                 topicErrorSensor.record(count, now);
         }
 
-        public void recordLatency(int node, long latency) {
+        public void recordLatency(String node, long latency) {
             long now = time.milliseconds();
             this.requestTimeSensor.record(latency, now);
-            if (node >= 0) {
+            if (!node.isEmpty()) {
                 String nodeTimeName = "node-" + node + ".latency";
                 Sensor nodeRequestTime = this.metrics.getSensor(nodeTimeName);
                 if (nodeRequestTime != null)

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/common/Node.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/Node.java b/clients/src/main/java/org/apache/kafka/common/Node.java
index f4e4186..644cd71 100644
--- a/clients/src/main/java/org/apache/kafka/common/Node.java
+++ b/clients/src/main/java/org/apache/kafka/common/Node.java
@@ -18,12 +18,14 @@ package org.apache.kafka.common;
 public class Node {
 
     private final int id;
+    private final String idString;
     private final String host;
     private final int port;
 
     public Node(int id, String host, int port) {
         super();
         this.id = id;
+        this.idString = Integer.toString(id);
         this.host = host;
         this.port = port;
     }
@@ -40,6 +42,14 @@ public class Node {
     }
 
     /**
+     * String representation of the node id.
+     * Typically the integer id is used to serialize over the wire, the string representation is used as an identifier with NetworkClient code
+     */
+    public String idString() {
+        return idString;
+    }
+
+    /**
      * The host name for this node
      */
     public String host() {

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/common/network/ByteBufferReceive.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/ByteBufferReceive.java b/clients/src/main/java/org/apache/kafka/common/network/ByteBufferReceive.java
index 129ae82..159c301 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/ByteBufferReceive.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/ByteBufferReceive.java
@@ -25,11 +25,11 @@ import java.nio.channels.ScatteringByteChannel;
  */
 public class ByteBufferReceive implements Receive {
 
-    private final int source;
+    private final String source;
     private final ByteBuffer[] buffers;
     private int remaining;
 
-    public ByteBufferReceive(int source, ByteBuffer... buffers) {
+    public ByteBufferReceive(String source, ByteBuffer... buffers) {
         super();
         this.source = source;
         this.buffers = buffers;
@@ -38,7 +38,7 @@ public class ByteBufferReceive implements Receive {
     }
 
     @Override
-    public int source() {
+    public String source() {
         return source;
     }
 
@@ -54,8 +54,4 @@ public class ByteBufferReceive implements Receive {
         return read;
     }
 
-    public ByteBuffer[] reify() {
-        return buffers;
-    }
-
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/common/network/ByteBufferSend.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/ByteBufferSend.java b/clients/src/main/java/org/apache/kafka/common/network/ByteBufferSend.java
index c8213e1..df0e6d5 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/ByteBufferSend.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/ByteBufferSend.java
@@ -22,12 +22,12 @@ import java.nio.channels.GatheringByteChannel;
  */
 public class ByteBufferSend implements Send {
 
-    private final int destination;
+    private final String destination;
     protected final ByteBuffer[] buffers;
     private int remaining;
     private int size;
 
-    public ByteBufferSend(int destination, ByteBuffer... buffers) {
+    public ByteBufferSend(String destination, ByteBuffer... buffers) {
         super();
         this.destination = destination;
         this.buffers = buffers;
@@ -37,7 +37,7 @@ public class ByteBufferSend implements Send {
     }
 
     @Override
-    public int destination() {
+    public String destination() {
         return destination;
     }
 
@@ -47,16 +47,7 @@ public class ByteBufferSend implements Send {
     }
 
     @Override
-    public ByteBuffer[] reify() {
-        return this.buffers;
-    }
-
-    @Override
-    public int remaining() {
-        return this.remaining;
-    }
-
-    public int size() {
+    public long size() {
         return this.size;
     }
 
@@ -64,9 +55,8 @@ public class ByteBufferSend implements Send {
     public long writeTo(GatheringByteChannel channel) throws IOException {
         long written = channel.write(buffers);
         if (written < 0)
-            throw new EOFException("This shouldn't happen.");
+            throw new EOFException("Wrote negative bytes to channel. This shouldn't happen.");
         remaining -= written;
         return written;
     }
-
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/common/network/InvalidReceiveException.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/InvalidReceiveException.java b/clients/src/main/java/org/apache/kafka/common/network/InvalidReceiveException.java
new file mode 100644
index 0000000..a5bdd62
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/network/InvalidReceiveException.java
@@ -0,0 +1,30 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.network;
+
+import org.apache.kafka.common.KafkaException;
+
+public class InvalidReceiveException extends KafkaException {
+
+    public InvalidReceiveException(String message) {
+        super(message);
+    }
+
+    public InvalidReceiveException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/common/network/MultiSend.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/MultiSend.java b/clients/src/main/java/org/apache/kafka/common/network/MultiSend.java
new file mode 100644
index 0000000..0e14a39
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/network/MultiSend.java
@@ -0,0 +1,100 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.network;
+
+import org.apache.kafka.common.KafkaException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.channels.GatheringByteChannel;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * A set of composite sends, sent one after another
+ */
+
+public class MultiSend implements Send {
+
+    private static final Logger log = LoggerFactory.getLogger(MultiSend.class);
+    private String dest;
+    private long totalWritten = 0;
+    private List<Send> sends;
+    private Iterator<Send> sendsIterator;
+    private Send current;
+    private boolean doneSends = false;
+    private long size = 0;
+
+    public MultiSend(String dest, List<Send> sends) {
+        this.dest = dest;
+        this.sends = sends;
+        this.sendsIterator = sends.iterator();
+        nextSendOrDone();
+        for (Send send: sends)
+            this.size += send.size();
+    }
+
+    @Override
+    public long size() {
+        return size;
+    }
+
+    @Override
+    public String destination() {
+        return dest;
+    }
+
+    @Override
+    public boolean completed() {
+        if (doneSends) {
+            if (totalWritten != size)
+                log.error("mismatch in sending bytes over socket; expected: " + size + " actual: " + totalWritten);
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    @Override
+    public long writeTo(GatheringByteChannel channel) throws IOException {
+        if (completed())
+            throw new KafkaException("This operation cannot be completed on a complete request.");
+
+        int totalWrittenPerCall = 0;
+        boolean sendComplete = false;
+        do {
+            long written = current.writeTo(channel);
+            totalWritten += written;
+            totalWrittenPerCall += written;
+            sendComplete = current.completed();
+            if (sendComplete)
+                nextSendOrDone();
+        } while (!completed() && sendComplete);
+        if (log.isTraceEnabled())
+            log.trace("Bytes written as part of multisend call : " + totalWrittenPerCall +  "Total bytes written so far : " + totalWritten + "Expected bytes to write : " + size);
+        return totalWrittenPerCall;
+    }
+
+    // update current if there's a next Send, mark sends as done if there isn't
+    private void nextSendOrDone() {
+        if (sendsIterator.hasNext())
+            current = sendsIterator.next();
+        else
+            doneSends = true;
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java b/clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java
index fc0d168..3ca0098 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java
@@ -15,6 +15,7 @@ package org.apache.kafka.common.network;
 import java.io.EOFException;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
 import java.nio.channels.ScatteringByteChannel;
 
 /**
@@ -22,24 +23,42 @@ import java.nio.channels.ScatteringByteChannel;
  */
 public class NetworkReceive implements Receive {
 
-    private final int source;
+    public final static String UNKNOWN_SOURCE = "";
+    public final static int UNLIMITED = -1;
+
+    private final String source;
     private final ByteBuffer size;
+    private final int maxSize;
     private ByteBuffer buffer;
 
-    public NetworkReceive(int source, ByteBuffer buffer) {
+
+    public NetworkReceive(String source, ByteBuffer buffer) {
         this.source = source;
         this.buffer = buffer;
         this.size = null;
+        this.maxSize = UNLIMITED;
+    }
+
+    public NetworkReceive(String source) {
+        this.source = source;
+        this.size = ByteBuffer.allocate(4);
+        this.buffer = null;
+        this.maxSize = UNLIMITED;
     }
 
-    public NetworkReceive(int source) {
+    public NetworkReceive(int maxSize, String source) {
         this.source = source;
         this.size = ByteBuffer.allocate(4);
         this.buffer = null;
+        this.maxSize = maxSize;
+    }
+
+    public NetworkReceive() {
+        this(UNKNOWN_SOURCE);
     }
 
     @Override
-    public int source() {
+    public String source() {
         return source;
     }
 
@@ -48,13 +67,15 @@ public class NetworkReceive implements Receive {
         return !size.hasRemaining() && !buffer.hasRemaining();
     }
 
-    @Override
-    public ByteBuffer[] reify() {
-        return new ByteBuffer[] {this.buffer};
+    public long readFrom(ScatteringByteChannel channel) throws IOException {
+        return readFromReadableChannel(channel);
     }
 
-    @Override
-    public long readFrom(ScatteringByteChannel channel) throws IOException {
+    // Need a method to read from ReadableByteChannel because BlockingChannel requires read with timeout
+    // See: http://stackoverflow.com/questions/2866557/timeout-for-socketchannel-doesnt-work
+    // This can go away after we get rid of BlockingChannel
+    @Deprecated
+    public long readFromReadableChannel(ReadableByteChannel channel) throws IOException {
         int read = 0;
         if (size.hasRemaining()) {
             int bytesRead = channel.read(size);
@@ -63,10 +84,12 @@ public class NetworkReceive implements Receive {
             read += bytesRead;
             if (!size.hasRemaining()) {
                 size.rewind();
-                int requestSize = size.getInt();
-                if (requestSize < 0)
-                    throw new IllegalStateException("Invalid request (size = " + requestSize + ")");
-                this.buffer = ByteBuffer.allocate(requestSize);
+                int receiveSize = size.getInt();
+                if (receiveSize < 0)
+                    throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + ")");
+                if (maxSize != UNLIMITED && receiveSize > maxSize)
+                    throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + " larger than " + maxSize + ")");
+                this.buffer = ByteBuffer.allocate(receiveSize);
             }
         }
         if (buffer != null) {
@@ -83,4 +106,14 @@ public class NetworkReceive implements Receive {
         return this.buffer;
     }
 
+    // Used only by BlockingChannel, so we may be able to get rid of this when/if we get rid of BlockingChannel
+    @Deprecated
+    public long readCompletely(ReadableByteChannel channel) throws IOException {
+        int totalRead = 0;
+        while (!complete()) {
+            totalRead += readFromReadableChannel(channel);
+        }
+        return totalRead;
+    }
+
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java b/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java
index 68327cd..49964b0 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java
@@ -23,7 +23,7 @@ import java.nio.ByteBuffer;
  */
 public class NetworkSend extends ByteBufferSend {
 
-    public NetworkSend(int destination, ByteBuffer... buffers) {
+    public NetworkSend(String destination, ByteBuffer... buffers) {
         super(destination, sizeDelimit(buffers));
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/common/network/Receive.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/Receive.java b/clients/src/main/java/org/apache/kafka/common/network/Receive.java
index 4e33078..4b14431 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/Receive.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/Receive.java
@@ -17,7 +17,6 @@
 package org.apache.kafka.common.network;
 
 import java.io.IOException;
-import java.nio.ByteBuffer;
 import java.nio.channels.ScatteringByteChannel;
 
 /**
@@ -28,7 +27,7 @@ public interface Receive {
     /**
      * The numeric id of the source from which we are receiving data.
      */
-    public int source();
+    public String source();
 
     /**
      * Are we done receiving data?
@@ -36,11 +35,6 @@ public interface Receive {
     public boolean complete();
 
     /**
-     * Turn this receive into ByteBuffer instances, if possible (otherwise returns null).
-     */
-    public ByteBuffer[] reify();
-
-    /**
      * Read bytes into this receive from the given channel
      * @param channel The channel to read from
      * @return The number of bytes read

http://git-wip-us.apache.org/repos/asf/kafka/blob/78ba492e/clients/src/main/java/org/apache/kafka/common/network/Selectable.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/Selectable.java b/clients/src/main/java/org/apache/kafka/common/network/Selectable.java
index b5f8d83..618a0fa 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/Selectable.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/Selectable.java
@@ -29,12 +29,12 @@ public interface Selectable {
      * @param receiveBufferSize The receive buffer for the socket
      * @throws IOException If we cannot begin connecting
      */
-    public void connect(int id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize) throws IOException;
+    public void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize) throws IOException;
 
     /**
      * Begin disconnecting the connection identified by the given id
      */
-    public void disconnect(int id);
+    public void disconnect(String id);
 
     /**
      * Wakeup this selector if it is blocked on I/O
@@ -50,7 +50,7 @@ public interface Selectable {
      * Queue the given request for sending in the subsequent {@poll(long)} calls
      * @param send The request to send
      */
-    public void send(NetworkSend send);
+    public void send(Send send);
 
     /**
      * Do I/O. Reads, writes, connection establishment, etc.
@@ -62,7 +62,7 @@ public interface Selectable {
     /**
      * The list of sends that completed on the last {@link #poll(long, List) poll()} call.
      */
-    public List<NetworkSend> completedSends();
+    public List<Send> completedSends();
 
     /**
      * The list of receives that completed on the last {@link #poll(long, List) poll()} call.
@@ -73,25 +73,25 @@ public interface Selectable {
      * The list of connections that finished disconnecting on the last {@link #poll(long, List) poll()}
      * call.
      */
-    public List<Integer> disconnected();
+    public List<String> disconnected();
 
     /**
      * The list of connections that completed their connection on the last {@link #poll(long, List) poll()}
      * call.
      */
-    public List<Integer> connected();
+    public List<String> connected();
 
     /**
      * Disable reads from the given connection
      * @param id The id for the connection
      */
-    public void mute(int id);
+    public void mute(String id);
 
     /**
      * Re-enable reads from the given connection
      * @param id The id for the connection
      */
-    public void unmute(int id);
+    public void unmute(String id);
 
     /**
      * Disable reads from all connections


Mime
View raw message