You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by rs...@apache.org on 2018/09/29 09:05:54 UTC
[kafka] branch 1.1 updated: KAFKA-7453: Expire registered channels
not selected within idle timeout (#5712)
This is an automated email from the ASF dual-hosted git repository.
rsivaram pushed a commit to branch 1.1
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/1.1 by this push:
new 9206e4c KAFKA-7453: Expire registered channels not selected within idle timeout (#5712)
9206e4c is described below
commit 9206e4c4b120e8c58ac01b5ff9d29b6c52825d98
Author: Rajini Sivaram <ra...@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 <ju...@gmail.com>. Ismael Juma <is...@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 84b48f0..24e776c 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
@@ -278,6 +278,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 f1c6a5a..9f2fd43 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
@@ -361,6 +361,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()) {