optimized compatibility

consumer.go

@@ -448,14 +448,22 @@ feederLoop:
 		for i, msg := range msgs {
 		messageSelect:
 			select {
+			case <-child.dying:
+				child.broker.acks.Done()
+				continue feederLoop
 			case child.messages <- msg:
 				firstAttempt = true
 			case <-expiryTicker.C:
 				if !firstAttempt {
 					child.responseResult = errTimedOut
 					child.broker.acks.Done()
+				remainingLoop:
 					for _, msg = range msgs[i:] {
-						child.messages <- msg
+						select {
+						case child.messages <- msg:
+						case <-child.dying:
+							break remainingLoop
+						}
 					}
 					child.broker.input <- child
 					continue feederLoop

consumer_test.go

@@ -807,28 +807,8 @@ func TestConsumerInterleavedClose(t *testing.T) {
 	assertMessageOffset(t, <-c0.Messages(), 1001)
 	assertMessageOffset(t, <-c0.Messages(), 1002)
 
-	var wg sync.WaitGroup
-	wg.Add(2)
-
-	go func() {
-		defer wg.Done()
-
-		for range c0.Messages() {
-			// drain
-		}
-	}()
-	go func() {
-		defer wg.Done()
-
-		for range c1.Messages() {
-			// drain
-		}
-	}()
-
 	safeClose(t, c1)
 	safeClose(t, c0)
-	wg.Wait()
-
 	safeClose(t, master)
 	broker0.Close()
 }