config validation + test

config.go

@@ -607,6 +607,13 @@ func (c *Config) Validate() error {
 		return ConfigurationError("Consumer.Offsets.Initial must be OffsetOldest or OffsetNewest")
 	case c.Consumer.Offsets.Retry.Max < 0:
 		return ConfigurationError("Consumer.Offsets.Retry.Max must be >= 0")
+	case c.Consumer.IsolationLevel != ReadUncommitted && c.Consumer.IsolationLevel != ReadCommitted:
+		return ConfigurationError("Consumer.IsolationLevel must be ReadUncommitted or ReadCommitted")
+	}
+
+	// validate IsolationLevel
+	if c.Consumer.IsolationLevel == ReadCommitted && !c.Version.IsAtLeast(V0_11_0_0) {
+		return ConfigurationError("ReadCommitted requires Version >= V0_11_0_0")
 	}
 
 	// validate the Consumer Group values

config_test.go

@@ -272,6 +272,24 @@ func TestProducerConfigValidates(t *testing.T) {
 				cfg.Producer.RequiredAcks = WaitForAll
 			},
 			"Idempotent producer requires Net.MaxOpenRequests to be 1"},
+		{"ReadCommitted Version",
+			func(cfg *Config) {
+				cfg.Version = V0_10_0_0
+				cfg.Consumer.IsolationLevel = ReadCommitted
+			},
+			"ReadCommitted requires Version >= V0_11_0_0"},
+		{"ReadCommitted Version",
+			func(cfg *Config) {
+				cfg.Version = V0_10_0_0
+				cfg.Consumer.IsolationLevel = ReadCommitted
+			},
+			"ReadCommitted requires Version >= V0_11_0_0"},
+		{"ReadCommitted Version",
+			func(cfg *Config) {
+				cfg.Version = V0_11_0_0
+				cfg.Consumer.IsolationLevel = IsolationLevel(42)
+			},
+			"Consumer.IsolationLevel must be ReadUncommitted or ReadCommitted"},
 	}
 
 	for i, test := range tests {