Merge pull request #1115 from Shopify/move-control-check

Fix block on control messages

consumer.go

@@ -579,10 +579,6 @@ func (child *partitionConsumer) parseResponse(response *FetchResponse) ([]*Consu
 
 	messages := []*ConsumerMessage{}
 	for _, records := range block.RecordsSet {
-		if control, err := records.isControl(); err != nil || control {
-			continue
-		}
-
 		switch records.recordsType {
 		case legacyRecords:
 			messageSetMessages, err := child.parseMessages(records.MsgSet)
@@ -596,6 +592,9 @@ func (child *partitionConsumer) parseResponse(response *FetchResponse) ([]*Consu
 			if err != nil {
 				return nil, err
 			}
+			if control, err := records.isControl(); err != nil || control {
+				continue
+			}
 
 			messages = append(messages, recordBatchMessages...)
 		default: