Improve experience trying to get Controller on old versions

client.go

@@ -17,7 +17,7 @@ type Client interface {
 	// altered after it has been created.
 	Config() *Config
 
-	// Controller returns the cluster controller broker.
+	// Controller returns the cluster controller broker. Requires Kafka 0.10 or higher.
 	Controller() (*Broker, error)
 
 	// Brokers returns the current set of active brokers as retrieved from cluster metadata.
@@ -388,6 +388,10 @@ func (client *client) Controller() (*Broker, error) {
 		return nil, ErrClosedClient
 	}
 
+	if !client.conf.Version.IsAtLeast(V0_10_0_0) {
+		return nil, ErrUnsupportedVersion
+	}
+
 	controller := client.cachedController()
 	if controller == nil {
 		if err := client.refreshMetadata(); err != nil {