Fix and enable TestConsumerInterleavedClose

consumer_test.go

@@ -336,8 +336,6 @@ func TestConsumerRebalancingMultiplePartitions(t *testing.T) {
 }
 
 func TestConsumerInterleavedClose(t *testing.T) {
-	t.Skip("Enable once bug #325 is fixed.")
-
 	seedBroker := newMockBroker(t, 1)
 	leader := newMockBroker(t, 2)
 
@@ -370,6 +368,7 @@ func TestConsumerInterleavedClose(t *testing.T) {
 	fetchResponse := new(FetchResponse)
 	fetchResponse.AddMessage("my_topic", 0, nil, ByteEncoder([]byte{0x00, 0x0E}), int64(0))
 	leader.Returns(fetchResponse)
+	time.Sleep(50 * time.Millisecond)
 
 	offsetResponseNewest1 := new(OffsetResponse)
 	offsetResponseNewest1.AddTopicPartition("my_topic", 1, 1234)
@@ -383,7 +382,9 @@ func TestConsumerInterleavedClose(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	<-c0.Messages()
 
+	fetchResponse.AddMessage("my_topic", 0, nil, ByteEncoder([]byte{0x00, 0x0E}), int64(1))
 	fetchResponse.AddMessage("my_topic", 1, nil, ByteEncoder([]byte{0x00, 0x0E}), int64(0))
 	leader.Returns(fetchResponse)