fix(kafka admin): validations and show proper error messages (#729)

Author: rkpattnaik780

pkg/cmd/kafka/topic/update/update.go

@@ -292,11 +292,31 @@ func runCmd(opts *Options) error {
 
 func runInteractivePrompt(opts *Options) (err error) {
 
-	_, err = opts.Connection(connection.DefaultConfigRequireMasAuth)
+	conn, err := opts.Connection(connection.DefaultConfigRequireMasAuth)
 	if err != nil {
 		return err
 	}
 
+	api, kafkaInstance, err := conn.API().KafkaAdmin(opts.kafkaID)
+	if err != nil {
+		return err
+	}
+
+	// check if topic exists
+	_, httpRes, err := api.GetTopic(context.Background(), opts.topicName).
+		Execute()
+
+	topicNameTmplPair := localize.NewEntry("TopicName", opts.topicName)
+	kafkaNameTmplPair := localize.NewEntry("InstanceName", kafkaInstance.GetName())
+	if err != nil {
+		if httpRes == nil {
+			return err
+		}
+		if httpRes.StatusCode == 404 {
+			return errors.New(opts.localizer.MustLocalize("kafka.topic.common.error.topicNotFoundError", topicNameTmplPair, kafkaNameTmplPair))
+		}
+	}
+
 	logger, err := opts.Logger()
 	if err != nil {
 		return err

pkg/connection/keycloak_connection.go

@@ -187,9 +187,19 @@ func (c *KeycloakConnection) API() *api.API {
 		}
 
 		kafkaStatus := kafkaInstance.GetStatus()
-		if kafkaStatus != "ready" {
-			err = fmt.Errorf(`Kafka instance "%v" is not ready yet`, kafkaInstance.GetName())
 
+		switch kafkaStatus {
+		case "provisioning", "accepted":
+			err = fmt.Errorf(`Kafka instance "%v" is not ready yet`, kafkaInstance.GetName())
+			return nil, nil, err
+		case "failed":
+			err = fmt.Errorf(`Kafka instance "%v" has failed`, kafkaInstance.GetName())
+			return nil, nil, err
+		case "deprovision":
+			err = fmt.Errorf(`Kafka instance "%v" is being deprovisioned`, kafkaInstance.GetName())
+			return nil, nil, err
+		case "deleting":
+			err = fmt.Errorf(`Kafka instance "%v" is being deleted`, kafkaInstance.GetName())
 			return nil, nil, err
 		}