[FAB-4537] Sporadic CI failures in orderer/kafka

When halting fabric channel it is possible to select on the
closure of the kafka consumer channel before the haltChan.

Also reverts 26d71e0, re-enabling the TestChain/EnqueueProper
test.

Change-Id: I01d63801bf29ce1e73b4bdc84a3945570c2c515a
Signed-off-by: Luis Sanchez <[email protected]>

Author: Luis Sanchez

orderer/kafka/chain.go

@@ -205,7 +205,7 @@ func (chain *chainImpl) processMessagesToBlocks() ([]uint64, error) {
 			counts[indexExitChanPass]++
 			return counts, nil
 		case kafkaErr := <-chain.channelConsumer.Errors():
-			logger.Error(kafkaErr)
+			logger.Errorf("[channel: %s] Error during consumption: %s", chain.support.ChainID(), kafkaErr)
 			counts[indexRecvError]++
 			select {
 			case <-chain.errorChan: // If already closed, don't do anything
@@ -221,14 +221,17 @@ func (chain *chainImpl) processMessagesToBlocks() ([]uint64, error) {
 			// mark the chain as available, so we have to force that trigger via
 			// the emission of a CONNECT message. TODO Consider rate limiting
 			go sendConnectMessage(chain.consenter.retryOptions(), chain.haltChan, chain.producer, chain.channel)
-		case in := <-chain.channelConsumer.Messages():
+		case in, ok := <-chain.channelConsumer.Messages():
+			if !ok {
+				logger.Criticalf("[channel: %s] Kafka consumer closed.", chain.support.ChainID())
+				return counts, nil
+			}
 			select {
 			case <-chain.errorChan: // If this channel was closed...
 				chain.errorChan = make(chan struct{}) // ...make a new one.
 				logger.Infof("[channel: %s] Marked consenter as available again", chain.support.ChainID())
 			default:
 			}
-
 			if err := proto.Unmarshal(in.Value, msg); err != nil {
 				// This shouldn't happen, it should be filtered at ingress
 				logger.Criticalf("[channel: %s] Unable to unmarshal consumed message = %s", chain.support.ChainID(), err)

orderer/kafka/chain_test.go

@@ -31,36 +31,38 @@ var (
 )
 
 func TestChain(t *testing.T) {
-	mockChannel := newChannel("foo.channel", defaultPartition)
 
 	oldestOffset := int64(0)
 	newestOffset := int64(5)
 
 	message := sarama.StringEncoder("messageFoo")
 
-	mockBroker := sarama.NewMockBroker(t, 0)
-	defer func() { mockBroker.Close() }()
-
-	mockBroker.SetHandlerByMap(map[string]sarama.MockResponse{
-		"MetadataRequest": sarama.NewMockMetadataResponse(t).
-			SetBroker(mockBroker.Addr(), mockBroker.BrokerID()).
-			SetLeader(mockChannel.topic(), mockChannel.partition(), mockBroker.BrokerID()),
-		"ProduceRequest": sarama.NewMockProduceResponse(t).
-			SetError(mockChannel.topic(), mockChannel.partition(), sarama.ErrNoError),
-		"OffsetRequest": sarama.NewMockOffsetResponse(t).
-			SetOffset(mockChannel.topic(), mockChannel.partition(), sarama.OffsetOldest, oldestOffset).
-			SetOffset(mockChannel.topic(), mockChannel.partition(), sarama.OffsetNewest, newestOffset),
-		"FetchRequest": sarama.NewMockFetchResponse(t, 1).
-			SetMessage(mockChannel.topic(), mockChannel.partition(), newestOffset, message),
-	})
-
-	mockSupport := &mockmultichain.ConsenterSupport{
-		ChainIDVal:      mockChannel.topic(),
-		HeightVal:       uint64(3),
-		SharedConfigVal: &mockconfig.Orderer{KafkaBrokersVal: []string{mockBroker.Addr()}},
+	newMocks := func(t *testing.T) (mockChannel channel, mockBroker *sarama.MockBroker, mockSupport *mockmultichain.ConsenterSupport) {
+		mockChannel = newChannel(channelNameForTest(t), defaultPartition)
+		mockBroker = sarama.NewMockBroker(t, 0)
+		mockBroker.SetHandlerByMap(map[string]sarama.MockResponse{
+			"MetadataRequest": sarama.NewMockMetadataResponse(t).
+				SetBroker(mockBroker.Addr(), mockBroker.BrokerID()).
+				SetLeader(mockChannel.topic(), mockChannel.partition(), mockBroker.BrokerID()),
+			"ProduceRequest": sarama.NewMockProduceResponse(t).
+				SetError(mockChannel.topic(), mockChannel.partition(), sarama.ErrNoError),
+			"OffsetRequest": sarama.NewMockOffsetResponse(t).
+				SetOffset(mockChannel.topic(), mockChannel.partition(), sarama.OffsetOldest, oldestOffset).
+				SetOffset(mockChannel.topic(), mockChannel.partition(), sarama.OffsetNewest, newestOffset),
+			"FetchRequest": sarama.NewMockFetchResponse(t, 1).
+				SetMessage(mockChannel.topic(), mockChannel.partition(), newestOffset, message),
+		})
+		mockSupport = &mockmultichain.ConsenterSupport{
+			ChainIDVal:      mockChannel.topic(),
+			HeightVal:       uint64(3),
+			SharedConfigVal: &mockconfig.Orderer{KafkaBrokersVal: []string{mockBroker.Addr()}},
+		}
+		return
 	}
 
 	t.Run("New", func(t *testing.T) {
+		_, mockBroker, mockSupport := newMocks(t)
+		defer func() { mockBroker.Close() }()
 		chain, err := newChain(mockConsenter, mockSupport, newestOffset-1)
 
 		assert.NoError(t, err, "Expected newChain to return without errors")
@@ -87,6 +89,8 @@ func TestChain(t *testing.T) {
 	})
 
 	t.Run("Start", func(t *testing.T) {
+		_, mockBroker, mockSupport := newMocks(t)
+		defer func() { mockBroker.Close() }()
 		// Set to -1 because we haven't sent the CONNECT message yet
 		chain, _ := newChain(mockConsenter, mockSupport, newestOffset-1)
 
@@ -103,6 +107,8 @@ func TestChain(t *testing.T) {
 	})
 
 	t.Run("Halt", func(t *testing.T) {
+		_, mockBroker, mockSupport := newMocks(t)
+		defer func() { mockBroker.Close() }()
 		chain, _ := newChain(mockConsenter, mockSupport, newestOffset-1)
 
 		chain.Start()
@@ -132,6 +138,8 @@ func TestChain(t *testing.T) {
 	})
 
 	t.Run("DoubleHalt", func(t *testing.T) {
+		_, mockBroker, mockSupport := newMocks(t)
+		defer func() { mockBroker.Close() }()
 		chain, _ := newChain(mockConsenter, mockSupport, newestOffset-1)
 
 		chain.Start()
@@ -148,6 +156,8 @@ func TestChain(t *testing.T) {
 	})
 
 	t.Run("StartWithProducerForChannelError", func(t *testing.T) {
+		_, mockBroker, mockSupport := newMocks(t)
+		defer func() { mockBroker.Close() }()
 		// Point to an empty brokers list
 		mockSupportCopy := *mockSupport
 		mockSupportCopy.SharedConfigVal = &mockconfig.Orderer{KafkaBrokersVal: []string{}}
@@ -164,6 +174,8 @@ func TestChain(t *testing.T) {
 		// - Net.ReadTimeout
 		// - Consumer.Retry.Backoff
 		// - Metadata.Retry.Max
+		mockChannel, mockBroker, mockSupport := newMocks(t)
+		defer func() { mockBroker.Close() }()
 		chain, _ := newChain(mockConsenter, mockSupport, newestOffset-1)
 
 		// Have the broker return an ErrNotLeaderForPartition error
@@ -184,6 +196,8 @@ func TestChain(t *testing.T) {
 	})
 
 	t.Run("EnqueueIfNotStarted", func(t *testing.T) {
+		mockChannel, mockBroker, mockSupport := newMocks(t)
+		defer func() { mockBroker.Close() }()
 		chain, _ := newChain(mockConsenter, mockSupport, newestOffset-1)
 
 		// As in StartWithConnectMessageError, have the broker return an
@@ -211,6 +225,9 @@ func TestChain(t *testing.T) {
 		// - Consumer.Retry.Backoff
 		// - Metadata.Retry.Max
 
+		mockChannel, mockBroker, mockSupport := newMocks(t)
+		defer func() { mockBroker.Close() }()
+
 		// Provide an out-of-range offset
 		chain, _ := newChain(mockConsenter, mockSupport, newestOffset)
 
@@ -231,8 +248,8 @@ func TestChain(t *testing.T) {
 	})
 
 	t.Run("EnqueueProper", func(t *testing.T) {
-		t.Skip("Skipping test until FAB-4537 is resolved")
-
+		mockChannel, mockBroker, mockSupport := newMocks(t)
+		defer func() { mockBroker.Close() }()
 		chain, _ := newChain(mockConsenter, mockSupport, newestOffset-1)
 
 		mockBroker.SetHandlerByMap(map[string]sarama.MockResponse{
@@ -264,6 +281,8 @@ func TestChain(t *testing.T) {
 	})
 
 	t.Run("EnqueueIfHalted", func(t *testing.T) {
+		mockChannel, mockBroker, mockSupport := newMocks(t)
+		defer func() { mockBroker.Close() }()
 		chain, _ := newChain(mockConsenter, mockSupport, newestOffset-1)
 
 		mockBroker.SetHandlerByMap(map[string]sarama.MockResponse{
@@ -293,6 +312,8 @@ func TestChain(t *testing.T) {
 	})
 
 	t.Run("EnqueueError", func(t *testing.T) {
+		mockChannel, mockBroker, mockSupport := newMocks(t)
+		defer func() { mockBroker.Close() }()
 		chain, _ := newChain(mockConsenter, mockSupport, newestOffset-1)
 
 		// Use the "good" handler map that allows the Stage to complete without
@@ -336,7 +357,7 @@ func TestSetupProducerForChannel(t *testing.T) {
 	mockBroker := sarama.NewMockBroker(t, 0)
 	defer mockBroker.Close()
 
-	mockChannel := newChannel("foo.channel", defaultPartition)
+	mockChannel := newChannel(channelNameForTest(t), defaultPartition)
 
 	haltChan := make(chan struct{})
 
@@ -361,7 +382,7 @@ func TestSetupConsumerForChannel(t *testing.T) {
 	mockBroker := sarama.NewMockBroker(t, 0)
 	defer func() { mockBroker.Close() }()
 
-	mockChannel := newChannel("foo.channel", defaultPartition)
+	mockChannel := newChannel(channelNameForTest(t), defaultPartition)
 
 	oldestOffset := int64(0)
 	newestOffset := int64(5)
@@ -412,7 +433,7 @@ func TestSetupConsumerForChannel(t *testing.T) {
 }
 
 func TestCloseKafkaObjects(t *testing.T) {
-	mockChannel := newChannel("foo.channel", defaultPartition)
+	mockChannel := newChannel(channelNameForTest(t), defaultPartition)
 
 	mockSupport := &mockmultichain.ConsenterSupport{
 		ChainIDVal: mockChannel.topic(),
@@ -526,7 +547,7 @@ func TestGetLastCutBlockNumber(t *testing.T) {
 }
 
 func TestGetLastOffsetPersisted(t *testing.T) {
-	mockChannel := newChannel("foo.channel", defaultPartition)
+	mockChannel := newChannel(channelNameForTest(t), defaultPartition)
 	mockMetadata := &cb.Metadata{Value: utils.MarshalOrPanic(&ab.KafkaMetadata{LastOffsetPersisted: int64(5)})}
 
 	testCases := []struct {

orderer/kafka/channel_test.go

@@ -24,9 +24,9 @@ import (
 )
 
 func TestChannel(t *testing.T) {
-	chn := newChannel("foo.channel", defaultPartition)
+	chn := newChannel(channelNameForTest(t), defaultPartition)
 
-	expectedTopic := fmt.Sprintf("%s", "foo.channel")
+	expectedTopic := fmt.Sprintf("%s", channelNameForTest(t))
 	actualTopic := chn.topic()
 	assert.Equal(t, expectedTopic, actualTopic, "Got the wrong topic, expected %s, got %s instead", expectedTopic, actualTopic)
 

orderer/kafka/config_test.go

@@ -17,10 +17,10 @@ import (
 )
 
 func TestBrokerConfig(t *testing.T) {
-	mockChannel1 := newChannel("foo.channel", defaultPartition)
+	mockChannel1 := newChannel(channelNameForTest(t), defaultPartition)
 	// Use a partition ID that is not the 'default' (defaultPartition)
 	var differentPartition int32 = defaultPartition + 1
-	mockChannel2 := newChannel("foo.channel", differentPartition)
+	mockChannel2 := newChannel(channelNameForTest(t), differentPartition)
 
 	mockBroker := sarama.NewMockBroker(t, 0)
 	defer func() { mockBroker.Close() }()

orderer/kafka/consenter_test.go

@@ -7,8 +7,10 @@ SPDX-License-Identifier: Apache-2.0
 package kafka
 
 import (
+	"fmt"
 	"log"
 	"os"
+	"strings"
 	"testing"
 	"time"
 
@@ -67,7 +69,7 @@ func TestHandleChain(t *testing.T) {
 	newestOffset := int64(5)
 	message := sarama.StringEncoder("messageFoo")
 
-	mockChannel := newChannel("foo.channel", defaultPartition)
+	mockChannel := newChannel(channelNameForTest(t), defaultPartition)
 
 	mockBroker := sarama.NewMockBroker(t, 0)
 	mockBroker.SetHandlerByMap(map[string]sarama.MockResponse{
@@ -171,3 +173,8 @@ func tamperBytes(original []byte) []byte {
 	byteCount := len(original)
 	return original[:byteCount-1]
 }
+
+func channelNameForTest(t *testing.T) string {
+	name := strings.Split(fmt.Sprint(t), " ")[18] // w/golang 1.8, use t.Name()
+	return fmt.Sprintf("%s.channel", strings.Replace(strings.ToLower(name), "/", ".", -1))
+}

orderer/kafka/partitioner_test.go

@@ -18,7 +18,7 @@ func TestStaticPartitioner(t *testing.T) {
 	var numberOfPartitions int32 = 6
 
 	partitionerConstructor := newStaticPartitioner(partition)
-	partitioner := partitionerConstructor("foo.channel")
+	partitioner := partitionerConstructor(channelNameForTest(t))
 
 	for i := 0; i < 10; i++ {
 		assignedPartition, err := partitioner.Partition(new(sarama.ProducerMessage), numberOfPartitions)

orderer/kafka/retry_test.go

@@ -16,7 +16,7 @@ import (
 func TestRetry(t *testing.T) {
 	var rp *retryProcess
 
-	mockChannel := newChannel("foo.channel", defaultPartition)
+	mockChannel := newChannel(channelNameForTest(t), defaultPartition)
 	flag := false
 
 	noErrorFn := func() error {