[FAB-7054] more flexibility setting Kafka.Version

Made the parsing of Kafka.Version in orderer.yaml more
forgiving, and mapping the value to a supported
underlying value.

Vendored hashicorp/go-version utility for parsing
version strings.

Change-Id: I0e10809cc572ba560995c47718c3fbc8f9866494
Signed-off-by: Luis Sanchez <sanchezl@us.ibm.com>

Author: Luis Sanchez

common/viperutil/config_test.go

@@ -77,22 +77,34 @@ func TestKafkaVersionDecode(t *testing.T) {
 		expected    sarama.KafkaVersion
 		errExpected bool
 	}{
+		{"0.8", sarama.KafkaVersion{}, true},
 		{"0.8.2.0", sarama.V0_8_2_0, false},
 		{"0.8.2.1", sarama.V0_8_2_1, false},
 		{"0.8.2.2", sarama.V0_8_2_2, false},
 		{"0.9.0.0", sarama.V0_9_0_0, false},
+		{"0.9", sarama.V0_9_0_0, false},
+		{"0.9.0", sarama.V0_9_0_0, false},
 		{"0.9.0.1", sarama.V0_9_0_1, false},
+		{"0.9.0.3", sarama.V0_9_0_1, false},
 		{"0.10.0.0", sarama.V0_10_0_0, false},
+		{"0.10", sarama.V0_10_0_0, false},
+		{"0.10.0", sarama.V0_10_0_0, false},
 		{"0.10.0.1", sarama.V0_10_0_1, false},
 		{"0.10.1.0", sarama.V0_10_1_0, false},
 		{"0.10.2.0", sarama.V0_10_2_0, false},
-		{"Unsupported", sarama.KafkaVersion{}, true},
+		{"0.10.2.1", sarama.V0_10_2_0, false},
+		{"0.10.2.2", sarama.V0_10_2_0, false},
+		{"0.10.2.3", sarama.V0_10_2_0, false},
+		{"0.11", sarama.KafkaVersion{}, true},
+		{"0.11.0", sarama.KafkaVersion{}, true},
+		{"0.11.0.0", sarama.KafkaVersion{}, true},
+		{"Malformed", sarama.KafkaVersion{}, true},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.data, func(t *testing.T) {
 
-			data := fmt.Sprintf("---\nInner:\n    Version: %s", tc.data)
+			data := fmt.Sprintf("---\nInner:\n    Version: '%s'", tc.data)
 			err := config.ReadConfig(bytes.NewReader([]byte(data)))
 			if err != nil {
 				t.Fatalf("Error reading config: %s", err)

common/viperutil/config_util.go

@@ -20,6 +20,7 @@ import (
 	"encoding/pem"
 
 	"github.com/Shopify/sarama"
+	version "github.com/hashicorp/go-version"
 	"github.com/hyperledger/fabric/common/flogging"
 	"github.com/mitchellh/mapstructure"
 	"github.com/spf13/viper"
@@ -245,33 +246,39 @@ func pemBlocksFromFileDecodeHook() mapstructure.DecodeHookFunc {
 	}
 }
 
+var kafkaVersionConstraints map[sarama.KafkaVersion]version.Constraints
+
+func init() {
+	kafkaVersionConstraints = make(map[sarama.KafkaVersion]version.Constraints)
+	kafkaVersionConstraints[sarama.V0_8_2_0], _ = version.NewConstraint(">=0.8.2,<0.8.2.1")
+	kafkaVersionConstraints[sarama.V0_8_2_1], _ = version.NewConstraint(">=0.8.2.1,<0.8.2.2")
+	kafkaVersionConstraints[sarama.V0_8_2_2], _ = version.NewConstraint(">=0.8.2.2,<0.9.0.0")
+	kafkaVersionConstraints[sarama.V0_9_0_0], _ = version.NewConstraint(">=0.9.0.0,<0.9.0.1")
+	kafkaVersionConstraints[sarama.V0_9_0_1], _ = version.NewConstraint(">=0.9.0.1,<0.10.0.0")
+	kafkaVersionConstraints[sarama.V0_10_0_0], _ = version.NewConstraint(">=0.10.0.0,<0.10.0.1")
+	kafkaVersionConstraints[sarama.V0_10_0_1], _ = version.NewConstraint(">=0.10.0.1,<0.10.1.0")
+	kafkaVersionConstraints[sarama.V0_10_1_0], _ = version.NewConstraint(">=0.10.1.0,<0.10.2.0")
+	kafkaVersionConstraints[sarama.V0_10_2_0], _ = version.NewConstraint(">=0.10.2.0,<0.11.0.0")
+}
+
 func kafkaVersionDecodeHook() mapstructure.DecodeHookFunc {
 	return func(f reflect.Type, t reflect.Type, data interface{}) (interface{}, error) {
 		if f.Kind() != reflect.String || t != reflect.TypeOf(sarama.KafkaVersion{}) {
 			return data, nil
 		}
-		switch data {
-		case "0.8.2.0":
-			return sarama.V0_8_2_0, nil
-		case "0.8.2.1":
-			return sarama.V0_8_2_1, nil
-		case "0.8.2.2":
-			return sarama.V0_8_2_2, nil
-		case "0.9.0.0":
-			return sarama.V0_9_0_0, nil
-		case "0.9.0.1":
-			return sarama.V0_9_0_1, nil
-		case "0.10.0.0":
-			return sarama.V0_10_0_0, nil
-		case "0.10.0.1":
-			return sarama.V0_10_0_1, nil
-		case "0.10.1.0":
-			return sarama.V0_10_1_0, nil
-		case "0.10.2.0":
-			return sarama.V0_10_2_0, nil
-		default:
-			return nil, fmt.Errorf("Unsupported Kafka version: '%s'", data)
+
+		v, err := version.NewVersion(data.(string))
+		if err != nil {
+			return nil, fmt.Errorf("Unable to parse Kafka version: %s", err)
+		}
+
+		for kafkaVersion, constraints := range kafkaVersionConstraints {
+			if constraints.Check(v) {
+				return kafkaVersion, nil
+			}
 		}
+
+		return nil, fmt.Errorf("Unsupported Kafka version: '%s'", data)
 	}
 }
 

docs/source/kafka.rst

@@ -72,15 +72,14 @@ Additional considerations
 Supported Kafka versions and upgrading
 --------------------------------------
 
-Fabric uses the `sarama client library <https://github.com/Shopify/sarama>`_ and vendors a version of it that supportes the following Kafka client versions:
+Fabric uses the `sarama client library <https://github.com/Shopify/sarama>`_ and vendors a version of it that supports the following Kafka client versions:
 
-* ``Version: 0.9.0.1``
-* ``Version: 0.10.0.0``
-* ``Version: 0.10.0.1``
-* ``Version: 0.10.1.0``
-* ``Version: 0.10.2.0``
+* ``Version: 0.9.0``
+* ``Version: 0.10.0``
+* ``Version: 0.10.1``
+* ``Version: 0.10.2``
 
-The sample Kafka server image provided by Fabric contains Kafka server version ``0.10.2.0``. Out of the box, Fabric's ordering service nodes default to configuring their embedded Kafka client to match this version. If you are not using the sample Kafka server image provided by Fabric, ensure that you configure a Kafka client version that is compatible with your Kafka server using the ``Kafka.Version`` key in ``orderer.yaml``.
+The sample Kafka server image provided by Fabric contains Kafka server version ``0.10.2``. Out of the box, Fabric's ordering service nodes default to configuring their embedded Kafka client to match this version. If you are not using the sample Kafka server image provided by Fabric, ensure that you configure a Kafka client version that is compatible with your Kafka server using the ``Kafka.Version`` key in ``orderer.yaml``.
 
 Debugging
 ---------