[FAB-4357] Modify retry options for Kafka config

This changeset modifies the `Retry` options of the Kakfa-based orderer
configuration. It adds short/long retry interval switches as suggested
in FAB-4121 and exposes some retry/timeout-related settings from the
underlying sarama client library. Defaults are introduced for all of the
new options.

This is part of fixing FAB-4136.

Change-Id: I75b9a726a9948e0c68e062f6500b6376c3a1bba7
Signed-off-by: Kostas Christidis <[email protected]>

Author: kchristidis

bddtests/dc-orderer-base.yml

@@ -14,8 +14,10 @@ services:
       - ORDERER_GENERAL_GENESISFILE=${ORDERER_GENERAL_GENESISFILE}
       - ORDERER_GENERAL_LISTENADDRESS=0.0.0.0
       - ORDERER_GENERAL_LOGLEVEL=debug
-      - ORDERER_KAFKA_VERBOSE=true
       - ORDERER_GENERAL_TLS_ENABLED=true
+      - ORDERER_KAFKA_RETRY_SHORTINTERVAL=1s
+      - ORDERER_KAFKA_RETRY_SHORTTOTAL=30s
+      - ORDERER_KAFKA_VERBOSE=true
     volumes:
       - ./volumes/orderer:/var/hyperledger/bddtests/volumes/orderer
       - /etc/hyperledger/msp:/etc/hyperledger/msp

examples/cluster/config/orderer.yaml

@@ -138,17 +138,59 @@ RAMLedger:
 #
 #   SECTION: Kafka
 #
-#   - This section applies to the configuration of the Kafka-based orderer.
+#   - This section applies to the configuration of the Kafka-based orderer, and
+#     its interaction with the Kafka cluster.
 #
 ################################################################################
 Kafka:
 
-    # Retry: What to do if none of the Kafka brokers are available.
+    # Retry: What do if a connection to the Kafka cluster cannot be established,
+    # or if a metadata request to the Kafka cluster needs to be repeated.
     Retry:
-        # The orderer should attempt to reconnect every <Period>
-        Period: 3s
-        # Panic if <Stop> has elapsed and no connection has been established
-        Stop: 60s
+        # When a new channel is created, or when an existing channel is reloaded
+        # (as part of the orderer's bootstrapping), the orderer interacts with
+        # the Kafka cluster in the following ways:
+        # 1. It creates a Kafka producer (writer) for the Kafka partition that
+        # corresponds to the channel.
+        # 2. It uses that producer to post a no-op CONNECT message to that
+        # partition
+        # 3. It creates a Kafka consumer (reader) for that partition.
+        # If any of these steps fails, they will be re-attempted every
+        # <ShortInterval> for a total of <ShortTotal>, and then every
+        # <LongInterval> for a total of <LongTotal> until they succeed:
+        # 1. Creation of Kafka producer (writer) for the channel
+        # 2. Creation of Kafka consumer (reader) for the channel
+        # 3. Posting of no-op CONNECT message to the channel
+        # Note that the orderer will be unable to write to or read from a
+        # channel until all of the steps above have been completed successfully.
+        ShortInterval: 5s
+        ShortTotal: 10m
+        LongInterval: 5m
+        LongTotal: 12h
+        # Affects the socket timeouts when waiting for an initial connection, a
+        # response, or a transmission. See Config.Net for more info:
+        # https://godoc.org/github.com/Shopify/sarama#Config
+        NetworkTimeouts:
+            DialTimeout: 10s
+            ReadTimeout: 10s
+            WriteTimeout: 10s
+        # Affects the metadata requests when the Kafka cluster is in the middle
+        # of a leader election.See Config.Metadata for more info:
+        # https://godoc.org/github.com/Shopify/sarama#Config
+        Metadata:
+            RetryBackoff: 250ms
+            RetryMax: 3
+        # What to do if posting a message to the Kafka cluster fails. See
+        # Config.Producer for more info:
+        # https://godoc.org/github.com/Shopify/sarama#Config
+        Producer:
+            RetryBackoff: 100ms
+            RetryMax: 3
+        # What to do if reading from the Kafka cluster fails. See
+        # Config.Consumer for more info:
+        # https://godoc.org/github.com/Shopify/sarama#Config
+        Consumer:
+            RetryBackoff: 2s
 
     # Verbose: Enable logging for interactions with the Kafka cluster.
     Verbose: false

orderer/kafka/chain.go

@@ -428,9 +428,9 @@ func setupProducerForChannel(brokers []string, brokerConfig *sarama.Config, chan
 	var producer sarama.SyncProducer
 
 	// This will be revised in: https://jira.hyperledger.org/browse/FAB-4136
-	repeatTick := time.NewTicker(retryOptions.Period)
-	panicTick := time.NewTicker(retryOptions.Stop)
-	logger.Debugf("[channel: %s] Retrying every %s for a total of %s", channel.topic(), retryOptions.Period.String(), retryOptions.Stop.String())
+	repeatTick := time.NewTicker(retryOptions.ShortInterval)
+	panicTick := time.NewTicker(retryOptions.ShortTotal)
+	logger.Debugf("[channel: %s] Retrying every %s for a total of %s", channel.topic(), retryOptions.ShortInterval.String(), retryOptions.ShortTotal.String())
 	defer repeatTick.Stop()
 	defer panicTick.Stop()
 

orderer/kafka/chain_test.go

@@ -100,14 +100,8 @@ func TestChain(t *testing.T) {
 	})
 
 	t.Run("StartWithConnectMessageError", func(t *testing.T) {
-		mockBrokerConfigCopy := *mockBrokerConfig
-		mockBrokerConfigCopy.Net.ReadTimeout = 5 * time.Millisecond
-		mockBrokerConfigCopy.Consumer.Retry.Backoff = 5 * time.Millisecond
-		mockBrokerConfigCopy.Metadata.Retry.Max = 1
-
-		mockConsenterCopy := newMockConsenter(&mockBrokerConfigCopy, mockLocalConfig.General.TLS, mockLocalConfig.Kafka.Retry, mockLocalConfig.Kafka.Version)
-
-		chain, _ := newChain(mockConsenterCopy, mockSupport, newestOffset-1)
+		// Affected by Net.ReadTimeout, Consumer.Retry.Backoff, and Metadata.Retry.Max
+		chain, _ := newChain(mockConsenter, mockSupport, newestOffset-1)
 
 		mockBroker.SetHandlerByMap(map[string]sarama.MockResponse{
 			"MetadataRequest": sarama.NewMockMetadataResponse(t).
@@ -131,14 +125,8 @@ func TestChain(t *testing.T) {
 	})
 
 	t.Run("StartWithConsumerForChannelError", func(t *testing.T) {
-		mockBrokerConfigCopy := *mockBrokerConfig
-		mockBrokerConfigCopy.Net.ReadTimeout = 5 * time.Millisecond
-		mockBrokerConfigCopy.Consumer.Retry.Backoff = 5 * time.Millisecond
-		mockBrokerConfigCopy.Metadata.Retry.Max = 1
-
-		mockConsenterCopy := newMockConsenter(&mockBrokerConfigCopy, mockLocalConfig.General.TLS, mockLocalConfig.Kafka.Retry, mockLocalConfig.Kafka.Version)
-
-		chain, _ := newChain(mockConsenterCopy, mockSupport, newestOffset) // Provide an out-of-range offset
+		// Affected by Net.ReadTimeout, Consumer.Retry.Backoff, and Metadata.Retry.Max
+		chain, _ := newChain(mockConsenter, mockSupport, newestOffset) // Provide an out-of-range offset
 
 		mockBroker.SetHandlerByMap(map[string]sarama.MockResponse{
 			"MetadataRequest": sarama.NewMockMetadataResponse(t).
@@ -451,7 +439,7 @@ func TestProcessLoopRegularError(t *testing.T) {
 }
 
 func TestProcessLoopRegularQueueEnvelope(t *testing.T) {
-	batchTimeout, _ := time.ParseDuration("1s")
+	batchTimeout, _ := time.ParseDuration("100s") // Something big
 	newestOffset := int64(5)
 	lastCutBlockNumber := uint64(3)
 	haltedFlag := false
@@ -725,12 +713,9 @@ func TestProcessLoopRegularAndSendTimeToCutError(t *testing.T) {
 	metadataResponse.AddTopicPartition(mockChannel.topic(), mockChannel.partition(), mockBroker.BrokerID(), nil, nil, sarama.ErrNoError)
 	mockBroker.Returns(metadataResponse)
 
-	mockBrokerConfigCopy := *mockBrokerConfig
-	mockBrokerConfigCopy.Net.ReadTimeout = 5 * time.Millisecond
-	mockBrokerConfigCopy.Consumer.Retry.Backoff = 5 * time.Millisecond
-	mockBrokerConfigCopy.Metadata.Retry.Max = 1
+	// Affected by Net.ReadTimeout, Consumer.Retry.Backoff, and Metadata.Retry.Max
 
-	producer, err := sarama.NewSyncProducer([]string{mockBroker.Addr()}, &mockBrokerConfigCopy)
+	producer, err := sarama.NewSyncProducer([]string{mockBroker.Addr()}, mockBrokerConfig)
 	assert.NoError(t, err, "Expected no error when setting up the sarama SyncProducer")
 
 	failureResponse := new(sarama.ProduceResponse)
@@ -987,12 +972,9 @@ func TestSendConnectMessage(t *testing.T) {
 	metadataResponse.AddTopicPartition(mockChannel.topic(), mockChannel.partition(), mockBroker.BrokerID(), nil, nil, sarama.ErrNoError)
 	mockBroker.Returns(metadataResponse)
 
-	mockBrokerConfigCopy := *mockBrokerConfig
-	mockBrokerConfigCopy.Net.ReadTimeout = 5 * time.Millisecond
-	mockBrokerConfigCopy.Consumer.Retry.Backoff = 5 * time.Millisecond
-	mockBrokerConfigCopy.Metadata.Retry.Max = 1
+	// Affected by Net.ReadTimeout, Consumer.Retry.Backoff, and Metadata.Retry.Max
 
-	producer, err := sarama.NewSyncProducer([]string{mockBroker.Addr()}, &mockBrokerConfigCopy)
+	producer, err := sarama.NewSyncProducer([]string{mockBroker.Addr()}, mockBrokerConfig)
 	assert.NoError(t, err, "Expected no error when setting up the sarama SyncProducer")
 
 	t.Run("Proper", func(t *testing.T) {
@@ -1022,12 +1004,9 @@ func TestSendTimeToCut(t *testing.T) {
 	metadataResponse.AddTopicPartition(mockChannel.topic(), mockChannel.partition(), mockBroker.BrokerID(), nil, nil, sarama.ErrNoError)
 	mockBroker.Returns(metadataResponse)
 
-	mockBrokerConfigCopy := *mockBrokerConfig
-	mockBrokerConfigCopy.Net.ReadTimeout = 5 * time.Millisecond
-	mockBrokerConfigCopy.Consumer.Retry.Backoff = 5 * time.Millisecond
-	mockBrokerConfigCopy.Metadata.Retry.Max = 1
+	// Affected by Net.ReadTimeout, Consumer.Retry.Backoff, and Metadata.Retry.Max
 
-	producer, err := sarama.NewSyncProducer([]string{mockBroker.Addr()}, &mockBrokerConfigCopy)
+	producer, err := sarama.NewSyncProducer([]string{mockBroker.Addr()}, mockBrokerConfig)
 	assert.NoError(t, err, "Expected no error when setting up the sarama SyncProducer")
 
 	timeToCutBlockNumber := uint64(3)

orderer/kafka/config.go

@@ -20,10 +20,19 @@ func newBrokerConfig(tlsConfig localconfig.TLS, retryOptions localconfig.Retry,
 	// FIXME https://jira.hyperledger.org/browse/FAB-4136
 	// Use retryOptions to populate `Net`
 
+	brokerConfig.Consumer.Retry.Backoff = retryOptions.Consumer.RetryBackoff
+
 	// Allows us to retrieve errors that occur when consuming a channel, via the
 	// channel's `listenForErrors` goroutine.
 	brokerConfig.Consumer.Return.Errors = true
 
+	brokerConfig.Metadata.Retry.Backoff = retryOptions.Metadata.RetryBackoff
+	brokerConfig.Metadata.Retry.Max = retryOptions.Metadata.RetryMax
+
+	brokerConfig.Net.DialTimeout = retryOptions.NetworkTimeouts.DialTimeout
+	brokerConfig.Net.ReadTimeout = retryOptions.NetworkTimeouts.ReadTimeout
+	brokerConfig.Net.WriteTimeout = retryOptions.NetworkTimeouts.WriteTimeout
+
 	brokerConfig.Net.TLS.Enable = tlsConfig.Enabled
 	if brokerConfig.Net.TLS.Enable {
 		// create public/private key pair structure
@@ -49,6 +58,10 @@ func newBrokerConfig(tlsConfig localconfig.TLS, retryOptions localconfig.Retry,
 	// Set equivalent of Kafka producer config max.request.bytes to the default
 	// value of a Kafka broker's socket.request.max.bytes property (100 MiB).
 	brokerConfig.Producer.MaxMessageBytes = int(sarama.MaxRequestSize) // FIXME https://jira.hyperledger.org/browse/FAB-4083
+
+	brokerConfig.Producer.Retry.Backoff = retryOptions.Producer.RetryBackoff
+	brokerConfig.Producer.Retry.Max = retryOptions.Producer.RetryMax
+
 	// A partitioner is actually not needed the way we do things now,
 	// but we're adding it now to allow for flexibility in the future.
 	brokerConfig.Producer.Partitioner = newStaticPartitioner(chosenStaticPartition)

orderer/kafka/consenter_test.go

@@ -26,8 +26,31 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+var mockRetryOptions = localconfig.Retry{
+	ShortInterval: 50 * time.Millisecond,
+	ShortTotal:    200 * time.Millisecond,
+	LongInterval:  200 * time.Millisecond,
+	LongTotal:     1 * time.Second,
+	NetworkTimeouts: localconfig.NetworkTimeouts{
+		DialTimeout:  5 * time.Millisecond,
+		ReadTimeout:  5 * time.Millisecond,
+		WriteTimeout: 5 * time.Millisecond,
+	},
+	Metadata: localconfig.Metadata{
+		RetryMax:     2,
+		RetryBackoff: 40 * time.Millisecond,
+	},
+	Producer: localconfig.Producer{
+		RetryMax:     2,
+		RetryBackoff: 30 * time.Millisecond,
+	},
+	Consumer: localconfig.Consumer{
+		RetryBackoff: 20 * time.Millisecond,
+	},
+}
+
 func init() {
-	mockLocalConfig = newMockLocalConfig(false, 50, 200, false)
+	mockLocalConfig = newMockLocalConfig(false, mockRetryOptions, false)
 	mockBrokerConfig = newMockBrokerConfig(mockLocalConfig.General.TLS, mockLocalConfig.Kafka.Retry, mockLocalConfig.Kafka.Version, defaultPartition)
 	mockConsenter = newMockConsenter(mockBrokerConfig, mockLocalConfig.General.TLS, mockLocalConfig.Kafka.Retry, mockLocalConfig.Kafka.Version)
 	setupTestLogging("ERROR", mockLocalConfig.Kafka.Verbose)
@@ -113,18 +136,15 @@ func newMockEnvelope(content string) *cb.Envelope {
 	return &cb.Envelope{Payload: []byte(content)}
 }
 
-func newMockLocalConfig(enableTLS bool, retryPeriod int, retryStop int, verboseLog bool) *localconfig.TopLevel {
+func newMockLocalConfig(enableTLS bool, retryOptions localconfig.Retry, verboseLog bool) *localconfig.TopLevel {
 	return &localconfig.TopLevel{
 		General: localconfig.General{
 			TLS: localconfig.TLS{
 				Enabled: enableTLS,
 			},
 		},
 		Kafka: localconfig.Kafka{
-			Retry: localconfig.Retry{
-				Period: time.Duration(retryPeriod) * time.Millisecond,
-				Stop:   time.Duration(retryStop) * time.Millisecond,
-			},
+			Retry:   retryOptions,
 			Verbose: verboseLog,
 			Version: sarama.V0_9_0_1,
 		},