diff --git a/utils.go b/utils.go
index 70ecab247..abb063678 100644
--- a/utils.go
+++ b/utils.go
@@ -2,7 +2,9 @@ package sarama
 
 import (
 	"bufio"
+	"fmt"
 	"net"
+	"regexp"
 )
 
 type none struct{}
@@ -150,3 +152,33 @@ var (
 	V1_0_0_0   = newKafkaVersion(1, 0, 0, 0)
 	minVersion = V0_8_2_0
 )
+
+func ParseKafkaVersion(s string) (KafkaVersion, error) {
+	var major, minor, veryMinor, patch uint
+	var err error
+	if s[0] == '0' {
+		err = scanKafkaVersion(s, `^0.\d+.\d+.\d+$`, "0.%d.%d.%d", [3]*uint{&minor, &veryMinor, &patch})
+	} else {
+		err = scanKafkaVersion(s, `^\d+.\d+.\d+$`, "%d.%d.%d", [3]*uint{&major, &minor, &veryMinor})
+	}
+	if err != nil {
+		return minVersion, err
+	}
+	return newKafkaVersion(major, minor, veryMinor, patch), nil
+}
+
+func scanKafkaVersion(s string, pattern string, format string, v [3]*uint) error {
+	if !regexp.MustCompile(pattern).MatchString(s) {
+		return fmt.Errorf("invalid version `%s`", s)
+	}
+	_, err := fmt.Sscanf(s, format, v[0], v[1], v[2])
+	return err
+}
+
+func (v KafkaVersion) String() string {
+	if v.version[0] == 0 {
+		return fmt.Sprintf("0.%d.%d.%d", v.version[1], v.version[2], v.version[3])
+	} else {
+		return fmt.Sprintf("%d.%d.%d", v.version[0], v.version[1], v.version[2])
+	}
+}
diff --git a/utils_test.go b/utils_test.go
index a9e09502c..013620e55 100644
--- a/utils_test.go
+++ b/utils_test.go
@@ -19,3 +19,23 @@ func TestVersionCompare(t *testing.T) {
 		t.Error("0.8.2.1 >= 0.10.0.0")
 	}
 }
+
+func TestVersionParsing(t *testing.T) {
+	validVersions := []string{"0.8.2.0", "0.8.2.1", "0.9.0.0", "0.10.2.0", "1.0.0"}
+	for _, s := range validVersions {
+		v, err := ParseKafkaVersion(s)
+		if err != nil {
+			t.Errorf("could not parse valid version %s: %s", s, err)
+		}
+		if v.String() != s {
+			t.Errorf("version %s != %s", v.String(), s)
+		}
+	}
+
+	invalidVersions := []string{"0.8.2-4", "0.8.20", "1.19.0.0", "1.0.x"}
+	for _, s := range invalidVersions {
+		if _, err := ParseKafkaVersion(s); err == nil {
+			t.Errorf("invalid version %s parsed without error", s)
+		}
+	}
+}