Retry coordinator lookup to fix transient failure in ConsumerBounceTest
Add logic in ConsumerBounceTest to check the error code in FindCoordinator responses and retry if needed. This should help with transient failures or at least get us closer to the actual problem.
Reviewers: Manikumar Reddy
@@ -26,7 +26,7 @@ import org.apache.kafka.clients.consumer._
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.errors.GroupMaxSizeReachedException
-import org.apache.kafka.common.protocol.ApiKeys
+import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.requests.{FindCoordinatorRequest, FindCoordinatorResponse}
import org.junit.Assert._
import org.junit.{After, Before, Ignore, Test}
@@ -263,9 +263,14 @@ class ConsumerBounceTest extends BaseRequestTest with Logging {
private def findCoordinator(group: String): Int = {
val request = new FindCoordinatorRequest.Builder(FindCoordinatorRequest.CoordinatorType.GROUP, group).build()
+ var nodeId = -1
+ TestUtils.waitUntilTrue(() => {
val resp = connectAndSend(request, ApiKeys.FIND_COORDINATOR)
- val response = FindCoordinatorResponse.parse(resp, ApiKeys.FIND_COORDINATOR.latestVersion())
- response.node().id()
+ val response = FindCoordinatorResponse.parse(resp, ApiKeys.FIND_COORDINATOR.latestVersion)
+ nodeId = response.node.id
+ response.error == Errors.NONE
+ }, s"Failed to find coordinator for group $group")
+ nodeId
}
/**
