Merge pull request #859 from ueokande/support-errors

Additional error codes

errors.go

@@ -108,12 +108,20 @@ const (
 	ErrUnsupportedSASLMechanism        KError = 33
 	ErrIllegalSASLState                KError = 34
 	ErrUnsupportedVersion              KError = 35
+	ErrTopicAlreadyExists              KError = 36
+	ErrInvalidPartitions               KError = 37
+	ErrInvalidReplicationFactor        KError = 38
+	ErrInvalidReplicaAssignment        KError = 39
+	ErrInvalidConfig                   KError = 40
+	ErrNotController                   KError = 41
+	ErrInvalidRequest                  KError = 42
 	ErrUnsupportedForMessageFormat     KError = 43
+	ErrPolicyViolation                 KError = 44
 )
 
 func (err KError) Error() string {
 	// Error messages stolen/adapted from
-	// https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol
+	// https://kafka.apache.org/protocol#protocol_error_codes
 	switch err {
 	case ErrNoError:
 		return "kafka server: Not an error, why are you printing me?"
@@ -189,8 +197,24 @@ func (err KError) Error() string {
 		return "kafka server: Request is not valid given the current SASL state."
 	case ErrUnsupportedVersion:
 		return "kafka server: The version of API is not supported."
+	case ErrTopicAlreadyExists:
+		return "kafka server: Topic with this name already exists."
+	case ErrInvalidPartitions:
+		return "kafka server: Number of partitions is invalid."
+	case ErrInvalidReplicationFactor:
+		return "kafka server: Replication-factor is invalid."
+	case ErrInvalidReplicaAssignment:
+		return "kafka server: Replica assignment is invalid."
+	case ErrInvalidConfig:
+		return "kafka server: Configuration is invalid."
+	case ErrNotController:
+		return "kafka server: This is not the correct controller for this cluster."
+	case ErrInvalidRequest:
+		return "kafka server: This most likely occurs because of a request being malformed by the client library or the message was sent to an incompatible broker. See the broker logs for more details."
 	case ErrUnsupportedForMessageFormat:
 		return "kafka server: The requested operation is not supported by the message format version."
+	case ErrPolicyViolation:
+		return "kafka server: Request parameters do not satisfy the configured policy."
 	}
 
 	return fmt.Sprintf("Unknown error, how did this happen? Error code = %d", err)