[FAB-10949] fix race in TestResubmission

The test was beginning to marshal an object that had not been fully
unmarshaled from the wire.

Change-Id: I6f6b019405197d556459f26e1637cfa03e2dc893
Signed-off-by: Matthew Sykes <sykesmat@us.ibm.com>

Author: sykesm

orderer/consensus/kafka/chain_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/hyperledger/fabric/protos/utils"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
 )
 
 var (
@@ -2464,9 +2465,12 @@ func TestResubmission(t *testing.T) {
 			}
 			defer close(mockSupport.BlockCutterVal.Block)
 
-			expectedKafkaMsg := &ab.KafkaMessage{}
+			expectedKafkaMsgCh := make(chan *ab.KafkaMessage, 1)
 			producer := mocks.NewSyncProducer(t, mockBrokerConfig)
 			producer.ExpectSendMessageWithCheckerFunctionAndSucceed(func(val []byte) error {
+				defer close(expectedKafkaMsgCh)
+
+				expectedKafkaMsg := &ab.KafkaMessage{}
 				if err := proto.Unmarshal(val, expectedKafkaMsg); err != nil {
 					return err
 				}
@@ -2484,6 +2488,7 @@ func TestResubmission(t *testing.T) {
 					return fmt.Errorf("Expect Original Offset to be non-zero if resubmission")
 				}
 
+				expectedKafkaMsgCh <- expectedKafkaMsg
 				return nil
 			})
 
@@ -2537,8 +2542,14 @@ func TestResubmission(t *testing.T) {
 			}
 
 			// Emits the kafka message produced by consenter
-			mpc.YieldMessage(newMockConsumerMessage(expectedKafkaMsg))
-			mockSupport.BlockCutterVal.Block <- struct{}{}
+			select {
+			case expectedKafkaMsg := <-expectedKafkaMsgCh:
+				require.NotNil(t, expectedKafkaMsg)
+				mpc.YieldMessage(newMockConsumerMessage(expectedKafkaMsg))
+				mockSupport.BlockCutterVal.Block <- struct{}{}
+			case <-time.After(shortTimeout):
+				t.Fatalf("Expected to receive kafka message")
+			}
 
 			select {
 			case <-mockSupport.Blocks:
@@ -2922,9 +2933,12 @@ func TestResubmission(t *testing.T) {
 			}
 			defer close(mockSupport.BlockCutterVal.Block)
 
-			expectedKafkaMsg := &ab.KafkaMessage{}
+			expectedKafkaMsgCh := make(chan *ab.KafkaMessage, 1)
 			producer := mocks.NewSyncProducer(t, mockBrokerConfig)
 			producer.ExpectSendMessageWithCheckerFunctionAndSucceed(func(val []byte) error {
+				defer close(expectedKafkaMsgCh)
+
+				expectedKafkaMsg := &ab.KafkaMessage{}
 				if err := proto.Unmarshal(val, expectedKafkaMsg); err != nil {
 					return err
 				}
@@ -2942,6 +2956,7 @@ func TestResubmission(t *testing.T) {
 					return fmt.Errorf("Expect Original Offset to be non-zero if resubmission")
 				}
 
+				expectedKafkaMsgCh <- expectedKafkaMsg
 				return nil
 			})
 
@@ -2986,8 +3001,14 @@ func TestResubmission(t *testing.T) {
 			// check that WaitReady is actually blocked because of in-flight reprocessed messages
 			blockIngressMsg(t, true, bareMinimumChain.WaitReady)
 
-			// Emits the kafka message produced by consenter
-			mpc.YieldMessage(newMockConsumerMessage(expectedKafkaMsg))
+			select {
+			case expectedKafkaMsg := <-expectedKafkaMsgCh:
+				require.NotNil(t, expectedKafkaMsg)
+				// Emits the kafka message produced by consenter
+				mpc.YieldMessage(newMockConsumerMessage(expectedKafkaMsg))
+			case <-time.After(shortTimeout):
+				t.Fatalf("Expected to receive kafka message")
+			}
 
 			select {
 			case <-mockSupport.Blocks: