kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gwens...@apache.org
Subject kafka git commit: KAFKA-2386; increase timeouts for transient test failure in ConsumerCoordinatorResponseTests
Date Mon, 03 Aug 2015 22:42:47 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk cd3dc7a5c -> 3c0963084


KAFKA-2386; increase timeouts for transient test failure in ConsumerCoordinatorResponseTests

There are two race conditions in the test case "testGenerationIdIncrementsOnRebalance." First,
a delay before the second join group request can timeout the initial group and cause the generationId
to unexpectedly reset. Second, a delay in the join group request handling will timeout the
request itself and cause the test to fail.  This commit doesn't address these race conditions,
but increases the timeouts to make them more unlikely. If the problem reoccurs, then we'll
probably need a better solution.

Author: Jason Gustafson <jason@confluent.io>

Reviewers: Gwen Shapira <cshapi@gmail.com>

Closes #107 from hachikuji/KAFKA-2386 and squashes the following commits:

a53460a [Jason Gustafson] KAFKA-2386; increase timeouts for transient test failure in ConsumerCoordinatorResponseTest


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

Branch: refs/heads/trunk
Commit: 3c09630844f6e70793f53a9d4f0ef562fe9d91d3
Parents: cd3dc7a
Author: Jason Gustafson <jason@confluent.io>
Authored: Mon Aug 3 15:42:33 2015 -0700
Committer: Chen Shapira <gwen@macbook-pro.gateway.sonic.net>
Committed: Mon Aug 3 15:42:33 2015 -0700

----------------------------------------------------------------------
 .../kafka/coordinator/ConsumerCoordinatorResponseTest.scala | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/3c096308/core/src/test/scala/unit/kafka/coordinator/ConsumerCoordinatorResponseTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/coordinator/ConsumerCoordinatorResponseTest.scala
b/core/src/test/scala/unit/kafka/coordinator/ConsumerCoordinatorResponseTest.scala
index 87a5330..058daef 100644
--- a/core/src/test/scala/unit/kafka/coordinator/ConsumerCoordinatorResponseTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/ConsumerCoordinatorResponseTest.scala
@@ -43,8 +43,8 @@ class ConsumerCoordinatorResponseTest extends JUnitSuite {
   type HeartbeatCallback = Short => Unit
 
   val ConsumerMinSessionTimeout = 10
-  val ConsumerMaxSessionTimeout = 100
-  val DefaultSessionTimeout = 20
+  val ConsumerMaxSessionTimeout = 200
+  val DefaultSessionTimeout = 100
   var consumerCoordinator: ConsumerCoordinator = null
   var offsetManager : OffsetManager = null
 
@@ -238,7 +238,7 @@ class ConsumerCoordinatorResponseTest extends JUnitSuite {
 
     // First start up a group (with a slightly larger timeout to give us time to heartbeat
when the rebalance starts)
     val joinGroupResult = joinGroup(groupId, JoinGroupRequest.UNKNOWN_CONSUMER_ID, partitionAssignmentStrategy,
-      100, isCoordinatorForGroup = true)
+      DefaultSessionTimeout, isCoordinatorForGroup = true)
     val assignedConsumerId = joinGroupResult._2
     val initialGenerationId = joinGroupResult._3
     val joinGroupErrorCode = joinGroupResult._4
@@ -310,7 +310,8 @@ class ConsumerCoordinatorResponseTest extends JUnitSuite {
                         sessionTimeout: Int,
                         isCoordinatorForGroup: Boolean): JoinGroupCallbackParams = {
     val responseFuture = sendJoinGroup(groupId, consumerId, partitionAssignmentStrategy,
sessionTimeout, isCoordinatorForGroup)
-    Await.result(responseFuture, Duration(40, TimeUnit.MILLISECONDS))
+    // should only have to wait as long as session timeout, but allow some extra time in
case of an unexpected delay
+    Await.result(responseFuture, Duration(sessionTimeout+100, TimeUnit.MILLISECONDS))
   }
 
   private def heartbeat(groupId: String,


Mime
View raw message