kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rsiva...@apache.org
Subject [kafka] branch trunk updated: KAFKA-7453: Expire registered channels not selected within idle timeout (#5712)
Date Fri, 28 Sep 2018 20:08:12 GMT
This is an automated email from the ASF dual-hosted git repository.

rsivaram pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 0a8205c  KAFKA-7453: Expire registered channels not selected within idle timeout
(#5712)
0a8205c is described below

commit 0a8205cb45667fd7d001f21e4e0412220b1abec1
Author: Rajini Sivaram <rajinisivaram@googlemail.com>
AuthorDate: Fri Sep 28 21:07:57 2018 +0100

    KAFKA-7453: Expire registered channels not selected within idle timeout (#5712)
    
    Reviewers: Jun Rao <junrao@gmail.com>. Ismael Juma <ismael@juma.me.uk>
---
 .../main/java/org/apache/kafka/common/network/Selector.java |  2 ++
 .../java/org/apache/kafka/common/network/SelectorTest.java  | 13 +++++++++++++
 2 files changed, 15 insertions(+)

diff --git a/clients/src/main/java/org/apache/kafka/common/network/Selector.java b/clients/src/main/java/org/apache/kafka/common/network/Selector.java
index 44223e7..93325d5 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/Selector.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/Selector.java
@@ -319,6 +319,8 @@ public class Selector implements Selectable, AutoCloseable {
         SelectionKey key = socketChannel.register(nioSelector, interestedOps);
         KafkaChannel channel = buildAndAttachKafkaChannel(socketChannel, id, key);
         this.channels.put(id, channel);
+        if (idleExpiryManager != null)
+            idleExpiryManager.update(channel.id(), time.nanoseconds());
         return key;
     }
 
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
index cfd7fb3..cef7c7f 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
@@ -364,6 +364,19 @@ public class SelectorTest {
     }
 
     @Test
+    public void testIdleExpiryWithoutReadyKeys() throws IOException {
+        String id = "0";
+        selector.connect(id, new InetSocketAddress("localhost", server.port), BUFFER_SIZE,
BUFFER_SIZE);
+        KafkaChannel channel = selector.channel(id);
+        channel.selectionKey().interestOps(0);
+
+        time.sleep(6000); // The max idle time is 5000ms
+        selector.poll(0);
+        assertTrue("The idle connection should have been closed", selector.disconnected().containsKey(id));
+        assertEquals(ChannelState.EXPIRED, selector.disconnected().get(id));
+    }
+
+    @Test
     public void testImmediatelyConnectedCleaned() throws Exception {
         Metrics metrics = new Metrics(); // new metrics object to avoid metric registration
conflicts
         Selector selector = new Selector(5000, metrics, time, "MetricGroup", channelBuilder,
new LogContext()) {


Mime
View raw message