{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/130078682","pull_request_review_id":52907497,"id":130078682,"diff_hunk":"@@ -184,6 +185,121 @@ class SocketServerTest extends JUnitSuite {\n   }\n \n   @Test\n+  def testConnectionId() {\n+    val sockets = (1 to 5).map(_ => connect(protocol = SecurityProtocol.PLAINTEXT))\n+    val serializedBytes = producerRequestBytes\n+\n+    val requests = sockets.map{ socket =>\n+      sendRequest(socket, serializedBytes)\n+      server.requestChannel.receiveRequest(2000)\n+    }\n+    requests.zipWithIndex.foreach { case (request, i) =>\n+      val index = requests(i).connectionId.split(\"-\").last\n+      assertEquals(i.toString, index)\n+    }\n+\n+    sockets.foreach(_.close)\n+  }\n+\n+  @Test\n+  def testIdleConnection() {\n+    val idleTimeMs = 60000\n+    val time = new MockTime()\n+    props.put(KafkaConfig.ConnectionsMaxIdleMsProp, idleTimeMs.toString)\n+    val serverMetrics = new Metrics\n+    val overrideServer = new SocketServer(KafkaConfig.fromProps(props), serverMetrics, time, credentialProvider)\n+\n+    def openChannel(request: RequestChannel.Request): Option[KafkaChannel] =\n+      overrideServer.processor(request.processor).channel(request.connectionId)\n+    def openOrClosingChannel(request: RequestChannel.Request): Option[KafkaChannel] =\n+      overrideServer.processor(request.processor).openOrClosingChannel(request.connectionId)\n+\n+    try {\n+      overrideServer.startup()\n+      val serializedBytes = producerRequestBytes\n+\n+      // Connection with no staged receives\n+      val socket1 = connect(overrideServer, protocol = SecurityProtocol.PLAINTEXT)\n+      sendRequest(socket1, serializedBytes)\n+      val request1 = overrideServer.requestChannel.receiveRequest(2000)\n+      assertTrue(\"Channel not open\", openChannel(request1).nonEmpty)\n+      assertEquals(openChannel(request1), openOrClosingChannel(request1))\n+\n+      time.sleep(idleTimeMs + 1)\n+      TestUtils.waitUntilTrue(() => openOrClosingChannel(request1).isEmpty, \"Failed to close idle channel\")\n+      assertTrue(\"Channel not removed\", openChannel(request1).isEmpty)\n+      processRequest(overrideServer.requestChannel, request1)\n+\n+      // Connection with staged receives\n+      val socket2 = connect(overrideServer, protocol = SecurityProtocol.PLAINTEXT)\n+      sendRequest(socket2, serializedBytes)\n+      sendRequest(socket2, serializedBytes)\n+      val request2 = overrideServer.requestChannel.receiveRequest(2000)\n+\n+      time.sleep(idleTimeMs + 1)\n+      TestUtils.waitUntilTrue(() => openChannel(request2).isEmpty, \"Failed to close idle channel\")\n+      assertTrue(\"Channel removed without processing staging receives\", openOrClosingChannel(request2).nonEmpty)\n+      processRequest(overrideServer.requestChannel, request2) // this triggers a failed send since channel has been closed\n+      TestUtils.waitUntilTrue(() => openOrClosingChannel(request2).isEmpty, \"Failed to remove channel with failed sends\")\n+      assertNull(\"Received request after failed send\", overrideServer.requestChannel.receiveRequest(200))\n+\n+    } finally {\n+      overrideServer.shutdown()\n+      serverMetrics.close()\n+    }\n+  }\n+\n+  @Test\n+  def testConnectionIdReuse() {\n+    val idleTimeMs = 60000\n+    val time = new MockTime()\n+    props.put(KafkaConfig.ConnectionsMaxIdleMsProp, idleTimeMs.toString)\n+    val serverMetrics = new Metrics\n+    val overrideConnectionId = \"127.0.0.1:1-127.0.0.1:2-0\"\n+    val overrideServer = new SocketServer(KafkaConfig.fromProps(props), serverMetrics, time, credentialProvider) {\n+      override def newProcessor(id: Int, connectionQuotas: ConnectionQuotas, listenerName: ListenerName,\n+                                protocol: SecurityProtocol): Processor = {\n+        new Processor(id, time, config.socketRequestMaxBytes, requestChannel, connectionQuotas,\n+          config.connectionsMaxIdleMs, listenerName, protocol, config, metrics, credentialProvider) {\n+          override private[network] def connectionId(channel: SocketChannel): String = overrideConnectionId\n+        }\n+      }\n+    }\n+\n+    def openChannel: Option[KafkaChannel] = overrideServer.processor(0).channel(overrideConnectionId)\n+    def openOrClosingChannel: Option[KafkaChannel] = overrideServer.processor(0).openOrClosingChannel(overrideConnectionId)\n+    def connectionCount = overrideServer.connectionCount(InetAddress.getByName(\"127.0.0.1\"))\n+\n+    try {\n+      overrideServer.startup()\n+      val socket1 = connect(overrideServer)\n+      TestUtils.waitUntilTrue(() => connectionCount == 1, \"Failed to create channel\")\n+      val channel1 = openChannel\n+\n+      // Create new connection with same id when channel1 is still open and in Selector.channels\n+      connect(overrideServer)\n+      TestUtils.waitUntilTrue(() => connectionCount == 1, \"Failed to close channel\")\n+      assertEquals(channel1, openChannel)\n+\n+      // Create new connection with same id when channel1 is closed with staged receives and in Selector.closingChannels\n+      val serializedBytes = producerRequestBytes\n+      (1 to 3).foreach(_ => sendRequest(socket1, serializedBytes))\n+      val request = overrideServer.requestChannel.receiveRequest(2000)\n+      time.sleep(idleTimeMs + 1)\n+      TestUtils.waitUntilTrue(() => openChannel.isEmpty, \"Idle channel not closed\")\n+      assertTrue(\"Channel removed without processing staging receives\", openOrClosingChannel.nonEmpty)\n+\n+      connect(overrideServer)\n+      TestUtils.waitUntilTrue(() => connectionCount == 1, \"Failed to close channel\")\n+      assertEquals(channel1, openOrClosingChannel)","path":"core/src/test/scala/unit/kafka/network/SocketServerTest.scala","position":133,"original_position":133,"commit_id":"debed9bc389b34bf11dd551ff8557e6a7d74c3da","original_commit_id":"debed9bc389b34bf11dd551ff8557e6a7d74c3da","user":{"login":"ijuma","id":24747,"avatar_url":"https://avatars2.githubusercontent.com/u/24747?v=4","gravatar_id":"","url":"https://api.github.com/users/ijuma","html_url":"https://github.com/ijuma","followers_url":"https://api.github.com/users/ijuma/followers","following_url":"https://api.github.com/users/ijuma/following{/other_user}","gists_url":"https://api.github.com/users/ijuma/gists{/gist_id}","starred_url":"https://api.github.com/users/ijuma/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ijuma/subscriptions","organizations_url":"https://api.github.com/users/ijuma/orgs","repos_url":"https://api.github.com/users/ijuma/repos","events_url":"https://api.github.com/users/ijuma/events{/privacy}","received_events_url":"https://api.github.com/users/ijuma/received_events","type":"User","site_admin":false},"body":"It would be nice to allow the staging receives to complete and verify that the closing channel is also empty like in the other test.","created_at":"2017-07-28T12:34:14Z","updated_at":"2017-07-28T12:41:04Z","html_url":"https://github.com/apache/kafka/pull/3530#discussion_r130078682","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/3530","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/130078682"},"html":{"href":"https://github.com/apache/kafka/pull/3530#discussion_r130078682"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/3530"}}},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/3530","id":130716120,"html_url":"https://github.com/apache/kafka/pull/3530","diff_url":"https://github.com/apache/kafka/pull/3530.diff","patch_url":"https://github.com/apache/kafka/pull/3530.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/3530","number":3530,"state":"open","locked":false,"title":"KAFKA-5595: Ensure client connection ids are not reused too quickly","user":{"login":"rajinisivaram","id":13164074,"avatar_url":"https://avatars0.githubusercontent.com/u/13164074?v=4","gravatar_id":"","url":"https://api.github.com/users/rajinisivaram","html_url":"https://github.com/rajinisivaram","followers_url":"https://api.github.com/users/rajinisivaram/followers","following_url":"https://api.github.com/users/rajinisivaram/following{/other_user}","gists_url":"https://api.github.com/users/rajinisivaram/gists{/gist_id}","starred_url":"https://api.github.com/users/rajinisivaram/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rajinisivaram/subscriptions","organizations_url":"https://api.github.com/users/rajinisivaram/orgs","repos_url":"https://api.github.com/users/rajinisivaram/repos","events_url":"https://api.github.com/users/rajinisivaram/events{/privacy}","received_events_url":"https://api.github.com/users/rajinisivaram/received_events","type":"User","site_admin":false},"body":"When there are broker delays that cause a response to take longer than `connections.max.idle.ms`, connections may be closed by the broker (as well as by the client) before the response is processed.\r\nIf the port is reused, broker may send the outstanding response to a new connection with the reused port. The new connection will end up with correlation id mismatch, requiring process restart. This is also a security exposure since clients receive response intended for the wrong connection.","created_at":"2017-07-15T13:36:24Z","updated_at":"2017-07-28T12:41:04Z","closed_at":null,"merged_at":null,"merge_commit_sha":"72119fb5ccba345de61b405fda96fc2a93e6fb68","assignee":null,"assignees":[],"requested_reviewers":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/3530/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/3530/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/3530/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/debed9bc389b34bf11dd551ff8557e6a7d74c3da","head":{"label":"rajinisivaram:KAFKA-5595","ref":"KAFKA-5595","sha":"debed9bc389b34bf11dd551ff8557e6a7d74c3da","user":{"login":"rajinisivaram","id":13164074,"avatar_url":"https://avatars0.githubusercontent.com/u/13164074?v=4","gravatar_id":"","url":"https://api.github.com/users/rajinisivaram","html_url":"https://github.com/rajinisivaram","followers_url":"https://api.github.com/users/rajinisivaram/followers","following_url":"https://api.github.com/users/rajinisivaram/following{/other_user}","gists_url":"https://api.github.com/users/rajinisivaram/gists{/gist_id}","starred_url":"https://api.github.com/users/rajinisivaram/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rajinisivaram/subscriptions","organizations_url":"https://api.github.com/users/rajinisivaram/orgs","repos_url":"https://api.github.com/users/rajinisivaram/repos","events_url":"https://api.github.com/users/rajinisivaram/events{/privacy}","received_events_url":"https://api.github.com/users/rajinisivaram/received_events","type":"User","site_admin":false},"repo":{"id":42296329,"name":"kafka","full_name":"rajinisivaram/kafka","owner":{"login":"rajinisivaram","id":13164074,"avatar_url":"https://avatars0.githubusercontent.com/u/13164074?v=4","gravatar_id":"","url":"https://api.github.com/users/rajinisivaram","html_url":"https://github.com/rajinisivaram","followers_url":"https://api.github.com/users/rajinisivaram/followers","following_url":"https://api.github.com/users/rajinisivaram/following{/other_user}","gists_url":"https://api.github.com/users/rajinisivaram/gists{/gist_id}","starred_url":"https://api.github.com/users/rajinisivaram/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rajinisivaram/subscriptions","organizations_url":"https://api.github.com/users/rajinisivaram/orgs","repos_url":"https://api.github.com/users/rajinisivaram/repos","events_url":"https://api.github.com/users/rajinisivaram/events{/privacy}","received_events_url":"https://api.github.com/users/rajinisivaram/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/rajinisivaram/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/rajinisivaram/kafka","forks_url":"https://api.github.com/repos/rajinisivaram/kafka/forks","keys_url":"https://api.github.com/repos/rajinisivaram/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/rajinisivaram/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/rajinisivaram/kafka/teams","hooks_url":"https://api.github.com/repos/rajinisivaram/kafka/hooks","issue_events_url":"https://api.github.com/repos/rajinisivaram/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/rajinisivaram/kafka/events","assignees_url":"https://api.github.com/repos/rajinisivaram/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/rajinisivaram/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/rajinisivaram/kafka/tags","blobs_url":"https://api.github.com/repos/rajinisivaram/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/rajinisivaram/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/rajinisivaram/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/rajinisivaram/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/rajinisivaram/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/rajinisivaram/kafka/languages","stargazers_url":"https://api.github.com/repos/rajinisivaram/kafka/stargazers","contributors_url":"https://api.github.com/repos/rajinisivaram/kafka/contributors","subscribers_url":"https://api.github.com/repos/rajinisivaram/kafka/subscribers","subscription_url":"https://api.github.com/repos/rajinisivaram/kafka/subscription","commits_url":"https://api.github.com/repos/rajinisivaram/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/rajinisivaram/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/rajinisivaram/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/rajinisivaram/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/rajinisivaram/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/rajinisivaram/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/rajinisivaram/kafka/merges","archive_url":"https://api.github.com/repos/rajinisivaram/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/rajinisivaram/kafka/downloads","issues_url":"https://api.github.com/repos/rajinisivaram/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/rajinisivaram/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/rajinisivaram/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/rajinisivaram/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/rajinisivaram/kafka/labels{/name}","releases_url":"https://api.github.com/repos/rajinisivaram/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/rajinisivaram/kafka/deployments","created_at":"2015-09-11T08:16:17Z","updated_at":"2017-05-26T20:03:07Z","pushed_at":"2017-07-19T12:12:36Z","git_url":"git://github.com/rajinisivaram/kafka.git","ssh_url":"git@github.com:rajinisivaram/kafka.git","clone_url":"https://github.com/rajinisivaram/kafka.git","svn_url":"https://github.com/rajinisivaram/kafka","homepage":null,"size":51405,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"e5e88f636fa9857105324f62cc758b1fed3602bb","user":{"login":"apache","id":47359,"avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":2211243,"name":"kafka","full_name":"apache/kafka","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/kafka","description":"Mirror of Apache Kafka","fork":false,"url":"https://api.github.com/repos/apache/kafka","forks_url":"https://api.github.com/repos/apache/kafka/forks","keys_url":"https://api.github.com/repos/apache/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/kafka/teams","hooks_url":"https://api.github.com/repos/apache/kafka/hooks","issue_events_url":"https://api.github.com/repos/apache/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/apache/kafka/events","assignees_url":"https://api.github.com/repos/apache/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/apache/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/apache/kafka/tags","blobs_url":"https://api.github.com/repos/apache/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/kafka/languages","stargazers_url":"https://api.github.com/repos/apache/kafka/stargazers","contributors_url":"https://api.github.com/repos/apache/kafka/contributors","subscribers_url":"https://api.github.com/repos/apache/kafka/subscribers","subscription_url":"https://api.github.com/repos/apache/kafka/subscription","commits_url":"https://api.github.com/repos/apache/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/apache/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/kafka/merges","archive_url":"https://api.github.com/repos/apache/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/kafka/downloads","issues_url":"https://api.github.com/repos/apache/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/apache/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/kafka/labels{/name}","releases_url":"https://api.github.com/repos/apache/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/apache/kafka/deployments","created_at":"2011-08-15T18:06:16Z","updated_at":"2017-07-28T11:05:26Z","pushed_at":"2017-07-28T08:05:58Z","git_url":"git://github.com/apache/kafka.git","ssh_url":"git@github.com:apache/kafka.git","clone_url":"https://github.com/apache/kafka.git","svn_url":"https://github.com/apache/kafka","homepage":null,"size":53020,"stargazers_count":5703,"watchers_count":5703,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":3540,"mirror_url":"git://git.apache.org/kafka.git","open_issues_count":411,"forks":3540,"open_issues":411,"watchers":5703,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/3530"},"html":{"href":"https://github.com/apache/kafka/pull/3530"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/3530"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/3530/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/3530/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/3530/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/debed9bc389b34bf11dd551ff8557e6a7d74c3da"}}}}
{"actor":{"display_login":"ijuma"}}