[FAB-7054] more flexibility setting Kafka.Version

Synced Kafka version support to match Fabric v1.1.

- 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.

- Updated the sarama Kafka client library version to
  match the version in master.

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

common/viperutil/config_test.go

@@ -77,21 +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},
-		{"Unsupported", sarama.KafkaVersion{}, true},
+		{"0.10.2.0", sarama.V0_10_2_0, false},
+		{"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,31 +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
-		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
 ---------