Fix KafkaConsumerTest.testGracefulClose transient failure
Author: Rajini Sivaram
Reviewers: Ismael Juma
Closes from rajinisivaram/KAFKA-4583
@@ -33,6 +33,7 @@ import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedDeque;
/**
* A mock network client for use testing code
@@ -67,7 +68,8 @@ public class MockClient implements KafkaClient {
private final Set<String> ready = new HashSet<>();
private final Map<Node, Long> blackedOut = new HashMap<>();
private final Queue<ClientRequest> requests = new ArrayDeque<>();
- private final Queue<ClientResponse> responses = new ArrayDeque<>();
+ // Use concurrent queue for responses so that responses may be updated during poll() from a different thread.
+ private final Queue<ClientResponse> responses = new ConcurrentLinkedDeque<>();
private final Queue<FutureResponse> futureResponses = new ArrayDeque<>();
private final Queue<Cluster> metadataUpdates = new ArrayDeque<>();
