[FAB-11574] Use AdminClient to create topics

The Kafka AdminClient API is now used in order
to create the topic required for a channel.
Prior to this, the orderer relied on the Kafka
cluster to have auto create topics enabled.

This API is only supported for Kafka v0.10.1.0
and higher.

If the API fails, the code still falls through
to the auto create path.

Change-Id: Id23d5b5f5826b24cb0cc3202e3ac6088e5ae3a49
Signed-off-by: Gari Singh <gari.r.singh@gmail.com>

Author: mastersingh24

integration/nwo/fabricconfig/orderer.go

@@ -67,6 +67,10 @@ type OrdererAuthentication struct {
 	TimeWindow time.Duration `yaml:"TimeWindow,omitempty"`
 }
 
+type OrdererTopic struct {
+	ReplicationFactor int16
+}
+
 type FileLedger struct {
 	Location string `yaml:"Location,omitempty"`
 	Prefix   string `yaml:"Prefix,omitempty"`
@@ -81,6 +85,7 @@ type Kafka struct {
 	Verbose   bool              `yaml:"Verbose"`
 	TLS       *OrdererTLS       `yaml:"TLS,omitempty"`
 	SASLPlain *OrdererSASLPlain `yaml:"SASLPlain,omitempty"`
+	Topic     *OrdererTopic     `yaml:"Topic,omitempty"`
 }
 
 type Retry struct {

integration/nwo/orderer_template.go

@@ -68,6 +68,8 @@ Kafka:
       RetryMax: 3
     Consumer:
       RetryBackoff: 2s
+  Topic:
+    ReplicationFactor: 1
   Verbose: false
   TLS:
     Enabled: false

integration/runner/kafka.go

@@ -211,6 +211,7 @@ func (k *Kafka) Run(sigCh <-chan os.Signal, ready chan<- struct{}) error {
 func (k *Kafka) buildEnv() []string {
 	env := []string{
 		"KAFKA_LOG_RETENTION_MS=-1",
+		//"KAFKA_AUTO_CREATE_TOPICS_ENABLE=false",
 		fmt.Sprintf("KAFKA_MESSAGE_MAX_BYTES=%d", k.MessageMaxBytes),
 		fmt.Sprintf("KAFKA_REPLICA_FETCH_MAX_BYTES=%d", k.ReplicaFetchMaxBytes),
 		fmt.Sprintf("KAFKA_UNCLEAN_LEADER_ELECTION_ENABLE=%s", strconv.FormatBool(k.UncleanLeaderElectionEnable)),

orderer/common/localconfig/config.go

@@ -110,6 +110,7 @@ type Kafka struct {
 	Version   sarama.KafkaVersion // TODO Move this to global config
 	TLS       TLS
 	SASLPlain SASLPlain
+	Topic     Topic
 }
 
 // Retry contains configuration related to retries and timeouts when the
@@ -155,6 +156,11 @@ type Consumer struct {
 	RetryBackoff time.Duration
 }
 
+// Topic contains the settings to use when creating Kafka topics
+type Topic struct {
+	ReplicationFactor int16
+}
+
 // Debug contains configuration for the orderer's debug parameters.
 type Debug struct {
 	BroadcastTraceDir string
@@ -219,6 +225,9 @@ var Defaults = TopLevel{
 		TLS: TLS{
 			Enabled: false,
 		},
+		Topic: Topic{
+			ReplicationFactor: 3,
+		},
 	},
 	Debug: Debug{
 		BroadcastTraceDir: "",

orderer/consensus/kafka/chain.go

@@ -255,6 +255,13 @@ func (chain *chainImpl) enqueue(kafkaMsg *ab.KafkaMessage) bool {
 func startThread(chain *chainImpl) {
 	var err error
 
+	// Create topic if it does not exist (requires Kafka v0.10.1.0)
+	err = setupTopicForChannel(chain.consenter.retryOptions(), chain.haltChan, chain.SharedConfig().KafkaBrokers(), chain.consenter.brokerConfig(), chain.consenter.topicDetail(), chain.channel)
+	if err != nil {
+		// log for now and fallback to auto create topics setting for broker
+		logger.Infof("[channel: %s]: failed to create Kafka topic = %s", chain.channel.topic(), err)
+	}
+
 	// Set up the producer
 	chain.producer, err = setupProducerForChannel(chain.consenter.retryOptions(), chain.haltChan, chain.SharedConfig().KafkaBrokers(), chain.consenter.brokerConfig(), chain.channel)
 	if err != nil {
@@ -950,3 +957,137 @@ func setupProducerForChannel(retryOptions localconfig.Retry, haltChan chan struc
 
 	return producer, setupProducer.retry()
 }
+
+// Creates the Kafka topic for the channel if it does not already exist
+func setupTopicForChannel(retryOptions localconfig.Retry, haltChan chan struct{}, brokers []string, brokerConfig *sarama.Config, topicDetail *sarama.TopicDetail, channel channel) error {
+
+	// requires Kafka v0.10.1.0 or higher
+	if !brokerConfig.Version.IsAtLeast(sarama.V0_10_1_0) {
+		return nil
+	}
+
+	logger.Infof("[channel: %s] Setting up the topic for this channel...",
+		channel.topic())
+
+	retryMsg := fmt.Sprintf("Creating Kafka topic [%s] for channel [%s]",
+		channel.topic(), channel.String())
+
+	setupTopic := newRetryProcess(
+		retryOptions,
+		haltChan,
+		channel,
+		retryMsg,
+		func() error {
+
+			var err error
+			clusterMembers := map[int32]*sarama.Broker{}
+			var controllerId int32
+
+			// loop through brokers to access metadata
+			for _, address := range brokers {
+				broker := sarama.NewBroker(address)
+				err = broker.Open(brokerConfig)
+
+				if err != nil {
+					continue
+				}
+
+				var ok bool
+				ok, err = broker.Connected()
+				if !ok {
+					continue
+				}
+				defer broker.Close()
+
+				// metadata request which includes the topic
+				var apiVersion int16
+				if brokerConfig.Version.IsAtLeast(sarama.V0_11_0_0) {
+					// use API version 4 to disable auto topic creation for
+					// metadata requests
+					apiVersion = 4
+				} else {
+					apiVersion = 1
+				}
+				metadata, err := broker.GetMetadata(&sarama.MetadataRequest{
+					Version:                apiVersion,
+					Topics:                 []string{channel.topic()},
+					AllowAutoTopicCreation: false})
+
+				if err != nil {
+					continue
+				}
+
+				controllerId = metadata.ControllerID
+				for _, broker := range metadata.Brokers {
+					clusterMembers[broker.ID()] = broker
+				}
+
+				for _, topic := range metadata.Topics {
+					if topic.Name == channel.topic() {
+						if topic.Err != sarama.ErrUnknownTopicOrPartition {
+							// auto create topics must be enabled so return
+							return nil
+						}
+					}
+				}
+				break
+			}
+
+			// check to see if we got any metadata from any of the brokers in the list
+			if len(clusterMembers) == 0 {
+				return fmt.Errorf(
+					"error creating topic [%s]; failed to retrieve metadata for the cluster",
+					channel.topic())
+			}
+
+			// get the controller
+			controller := clusterMembers[controllerId]
+			err = controller.Open(brokerConfig)
+
+			if err != nil {
+				return err
+			}
+
+			var ok bool
+			ok, err = controller.Connected()
+			if !ok {
+				return err
+			}
+			defer controller.Close()
+
+			// create the topic
+			req := &sarama.CreateTopicsRequest{
+				Version: 0,
+				TopicDetails: map[string]*sarama.TopicDetail{
+					channel.topic(): topicDetail},
+				Timeout: 3 * time.Second}
+			resp := &sarama.CreateTopicsResponse{}
+			resp, err = controller.CreateTopics(req)
+			if err != nil {
+				return err
+			}
+
+			// check the response
+			if topicErr, ok := resp.TopicErrors[channel.topic()]; ok {
+				// treat no error and topic exists error as success
+				if topicErr.Err == sarama.ErrNoError ||
+					topicErr.Err == sarama.ErrTopicAlreadyExists {
+					return nil
+				}
+				if topicErr.Err == sarama.ErrInvalidTopic {
+					// topic is invalid so abort
+					logger.Warningf("[channel: %s] Failed to set up topic = %s",
+						channel.topic(), topicErr.Err.Error())
+					go func() {
+						haltChan <- struct{}{}
+					}()
+				}
+				return fmt.Errorf("error creating topic: [%s]",
+					topicErr.Err.Error())
+			}
+
+			return nil
+		})
+
+	return setupTopic.retry()
+}

orderer/consensus/kafka/chain_test.go

@@ -447,6 +447,161 @@ func TestChain(t *testing.T) {
 	})
 }
 
+func TestSetupTopicForChannel(t *testing.T) {
+
+	mockChannel := newChannel(channelNameForTest(t), defaultPartition)
+	haltChan := make(chan struct{})
+
+	mockBrokerNoError := sarama.NewMockBroker(t, 0)
+	defer mockBrokerNoError.Close()
+	metadataResponse := sarama.NewMockMetadataResponse(t)
+	metadataResponse.SetBroker(mockBrokerNoError.Addr(),
+		mockBrokerNoError.BrokerID())
+	metadataResponse.SetController(mockBrokerNoError.BrokerID())
+
+	mdrUnknownTopicOrPartition := &sarama.MetadataResponse{
+		Version:      1,
+		Brokers:      []*sarama.Broker{sarama.NewBroker(mockBrokerNoError.Addr())},
+		ControllerID: -1,
+		Topics: []*sarama.TopicMetadata{
+			{
+				Err:  sarama.ErrUnknownTopicOrPartition,
+				Name: mockChannel.topic(),
+			},
+		},
+	}
+
+	mockBrokerNoError.SetHandlerByMap(map[string]sarama.MockResponse{
+		"CreateTopicsRequest": sarama.NewMockWrapper(
+			&sarama.CreateTopicsResponse{
+				TopicErrors: map[string]*sarama.TopicError{
+					mockChannel.topic(): {
+						Err: sarama.ErrNoError}}}),
+		"MetadataRequest": sarama.NewMockWrapper(mdrUnknownTopicOrPartition)})
+
+	mockBrokerTopicExists := sarama.NewMockBroker(t, 1)
+	defer mockBrokerTopicExists.Close()
+	mockBrokerTopicExists.SetHandlerByMap(map[string]sarama.MockResponse{
+		"CreateTopicsRequest": sarama.NewMockWrapper(
+			&sarama.CreateTopicsResponse{
+				TopicErrors: map[string]*sarama.TopicError{
+					mockChannel.topic(): {
+						Err: sarama.ErrTopicAlreadyExists}}}),
+		"MetadataRequest": sarama.NewMockWrapper(&sarama.MetadataResponse{
+			Version: 1,
+			Topics: []*sarama.TopicMetadata{
+				{
+					Name: channelNameForTest(t),
+					Err:  sarama.ErrNoError}}})})
+
+	mockBrokerInvalidTopic := sarama.NewMockBroker(t, 2)
+	defer mockBrokerInvalidTopic.Close()
+	metadataResponse = sarama.NewMockMetadataResponse(t)
+	metadataResponse.SetBroker(mockBrokerInvalidTopic.Addr(),
+		mockBrokerInvalidTopic.BrokerID())
+	metadataResponse.SetController(mockBrokerInvalidTopic.BrokerID())
+	mockBrokerInvalidTopic.SetHandlerByMap(map[string]sarama.MockResponse{
+		"CreateTopicsRequest": sarama.NewMockWrapper(
+			&sarama.CreateTopicsResponse{
+				TopicErrors: map[string]*sarama.TopicError{
+					mockChannel.topic(): {
+						Err: sarama.ErrInvalidTopic}}}),
+		"MetadataRequest": metadataResponse})
+
+	mockBrokerInvalidTopic2 := sarama.NewMockBroker(t, 3)
+	defer mockBrokerInvalidTopic2.Close()
+	mockBrokerInvalidTopic2.SetHandlerByMap(map[string]sarama.MockResponse{
+		"CreateTopicsRequest": sarama.NewMockWrapper(
+			&sarama.CreateTopicsResponse{
+				TopicErrors: map[string]*sarama.TopicError{
+					mockChannel.topic(): {
+						Err: sarama.ErrInvalidTopic}}}),
+		"MetadataRequest": sarama.NewMockWrapper(&sarama.MetadataResponse{
+			Version:      1,
+			Brokers:      []*sarama.Broker{sarama.NewBroker(mockBrokerInvalidTopic2.Addr())},
+			ControllerID: mockBrokerInvalidTopic2.BrokerID()})})
+
+	closedBroker := sarama.NewMockBroker(t, 99)
+	badAddress := closedBroker.Addr()
+	closedBroker.Close()
+
+	var tests = []struct {
+		name         string
+		brokers      []string
+		brokerConfig *sarama.Config
+		version      sarama.KafkaVersion
+		expectErr    bool
+		errorMsg     string
+	}{
+		{
+			name:         "Unsupported Version",
+			brokers:      []string{mockBrokerNoError.Addr()},
+			brokerConfig: sarama.NewConfig(),
+			version:      sarama.V0_9_0_0,
+			expectErr:    false,
+		},
+		{
+			name:         "No Error",
+			brokers:      []string{mockBrokerNoError.Addr()},
+			brokerConfig: sarama.NewConfig(),
+			version:      sarama.V0_10_2_0,
+			expectErr:    false,
+		},
+		{
+			name:         "Topic Exists",
+			brokers:      []string{mockBrokerTopicExists.Addr()},
+			brokerConfig: sarama.NewConfig(),
+			version:      sarama.V0_10_2_0,
+			expectErr:    false,
+		},
+		{
+			name:         "Invalid Topic",
+			brokers:      []string{mockBrokerInvalidTopic.Addr()},
+			brokerConfig: sarama.NewConfig(),
+			version:      sarama.V0_10_2_0,
+			expectErr:    true,
+			errorMsg:     "process asked to exit",
+		},
+		{
+			name:         "Multiple Brokers - One No Error",
+			brokers:      []string{badAddress, mockBrokerNoError.Addr()},
+			brokerConfig: sarama.NewConfig(),
+			version:      sarama.V0_10_2_0,
+			expectErr:    false,
+		},
+		{
+			name:         "Multiple Brokers - All Errors",
+			brokers:      []string{badAddress, badAddress},
+			brokerConfig: sarama.NewConfig(),
+			version:      sarama.V0_10_2_0,
+			expectErr:    true,
+			errorMsg:     "failed to retrieve metadata",
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+		t.Run(test.name, func(t *testing.T) {
+			test.brokerConfig.Version = test.version
+			err := setupTopicForChannel(
+				mockRetryOptions,
+				haltChan,
+				test.brokers,
+				test.brokerConfig,
+				&sarama.TopicDetail{
+					NumPartitions:     1,
+					ReplicationFactor: 2},
+				mockChannel)
+			if test.expectErr {
+				assert.Contains(t, err.Error(), test.errorMsg)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+
+}
+
 func TestSetupProducerForChannel(t *testing.T) {
 	if testing.Short() {
 		t.Skip("Skipping test in short mode")