kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rsiva...@apache.org
Subject kafka git commit: KAFKA-5872; Fix transient failure in SslSelectorTest.testMuteOnOOM
Date Tue, 12 Sep 2017 16:40:02 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 8bd2a68b5 -> 9ed8bf273


KAFKA-5872; Fix transient failure in SslSelectorTest.testMuteOnOOM

Author: Rajini Sivaram <rajinisivaram@googlemail.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>

Closes #3836 from rajinisivaram/KAFKA-5872-sslselectortest-failure


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

Branch: refs/heads/trunk
Commit: 9ed8bf273c6ee5569ed68b71782ec7ab947f9c93
Parents: 8bd2a68
Author: Rajini Sivaram <rajinisivaram@googlemail.com>
Authored: Tue Sep 12 17:39:47 2017 +0100
Committer: Rajini Sivaram <rajinisivaram@googlemail.com>
Committed: Tue Sep 12 17:39:47 2017 +0100

----------------------------------------------------------------------
 .../org/apache/kafka/common/network/Selector.java |  4 +++-
 .../kafka/common/network/SslSelectorTest.java     | 18 +++++++-----------
 2 files changed, 10 insertions(+), 12 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/9ed8bf27/clients/src/main/java/org/apache/kafka/common/network/Selector.java
----------------------------------------------------------------------
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 1e5d969..23dbc0f 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
@@ -395,6 +395,9 @@ public class Selector implements Selectable, AutoCloseable {
             //poll from channels where the underlying socket has more data
             pollSelectionKeys(readyKeys, false, endSelect);
             pollSelectionKeys(immediatelyConnectedKeys, true, endSelect);
+
+            // Clear all selected keys so that they are included in the ready count for the
next select
+            readyKeys.clear();
         } else {
             madeReadProgressLastPoll = true; //no work is also "progress"
         }
@@ -424,7 +427,6 @@ public class Selector implements Selectable, AutoCloseable {
         Iterator<SelectionKey> iterator = determineHandlingOrder(selectionKeys).iterator();
         while (iterator.hasNext()) {
             SelectionKey key = iterator.next();
-            iterator.remove();
             KafkaChannel channel = channel(key);
             long channelStartTimeNanos = recordTimePerConnection ? time.nanoseconds() : 0;
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/9ed8bf27/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
index bea8084..5683068 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
@@ -18,6 +18,7 @@ package org.apache.kafka.common.network;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
 import java.nio.channels.SelectionKey;
@@ -42,7 +43,6 @@ import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.common.utils.MockTime;
 import org.apache.kafka.test.TestSslUtils;
 import org.junit.After;
-import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -203,7 +203,7 @@ public class SslSelectorTest extends SelectorTest {
             selector.register("clientX", channelX);
             selector.register("clientY", channelY);
 
-            boolean success = false;
+            boolean handshaked = false;
             NetworkReceive firstReceive = null;
             long deadline = System.currentTimeMillis() + 5000;
             //keep calling poll until:
@@ -224,22 +224,18 @@ public class SslSelectorTest extends SelectorTest {
                     assertTrue("only expecting single request", completed.isEmpty());
                 }
 
-                boolean handshaked = sender1.waitForHandshake(1);
-                handshaked = handshaked && sender2.waitForHandshake(1);
+                handshaked = sender1.waitForHandshake(1) && sender2.waitForHandshake(1);
 
-                if (handshaked && firstReceive != null) {
-                    success = true;
+                if (handshaked && firstReceive != null && selector.isOutOfMemory())
                     break;
-                }
-            }
-            if (!success) {
-                Assert.fail("could not initiate connections within timeout");
             }
+            assertTrue("could not initiate connections within timeout", handshaked);
 
             selector.poll(10);
             assertTrue(selector.completedReceives().isEmpty());
             assertEquals(0, pool.availableMemory());
-            assertTrue(selector.isOutOfMemory());
+            assertNotNull("First receive not complete", firstReceive);
+            assertTrue("Selector not out of memory", selector.isOutOfMemory());
 
             firstReceive.close();
             assertEquals(900, pool.availableMemory()); //memory has been released back to
pool


Mime
View raw message