Method DeleteTopic

backend/pkg/console/delete_topic.go:24–72

DeleteTopic deletes a Kafka Topic (if possible and not disabled).

(ctx context.Context, topicName string)

// DeleteTopic deletes a Kafka Topic (if possible and not disabled).
func (s *Service) DeleteTopic(ctx context.Context, topicName string) *rest.Error {
	cl, _, err := s.kafkaClientFactory.GetKafkaClient(ctx)
	if err != nil {
		return errorToRestError(err)
	}

	req := kmsg.NewDeleteTopicsRequest()
	req.TopicNames = []string{topicName}
	req.Topics = []kmsg.DeleteTopicsRequestTopic{
		{
			Topic: new(topicName),
		},
	}
	req.TimeoutMillis = 30 * 1000 // 30s

	res, err := req.RequestWith(ctx, cl)
	if err != nil {
		return &rest.Error{
			Err:          err,
			Status:       http.StatusServiceUnavailable,
			Message:      fmt.Sprintf("Failed to execute delete topic command: %v", err.Error()),
			InternalLogs: []slog.Attr{slog.String("topic_name", topicName)},
			IsSilent:     false,
		}
	}

	if len(res.Topics) != 1 {
		return &rest.Error{
			Err:          errors.New("topics array in response is empty"),
			Status:       http.StatusServiceUnavailable,
			Message:      "Unexpected Kafka response: No topics set in the response",
			InternalLogs: []slog.Attr{slog.String("topic_name", topicName)},
			IsSilent:     false,
		}
	}

	topicRes := res.Topics[0]
	if err := newKafkaErrorWithDynamicMessage(topicRes.ErrorCode, topicRes.ErrorMessage); err != nil {
		return &rest.Error{
			Err:          err,
			Status:       http.StatusServiceUnavailable,
			Message:      fmt.Sprintf("Failed to delete Kafka topic: %v", err.Error()),
			InternalLogs: []slog.Attr{slog.String("topic_name", topicName)},
			IsSilent:     false,
		}
	}

	return nil
}

// DeleteTopics proxies the Kafka request/response between the Console service and Kafka.
func (s *Service) DeleteTopics(ctx context.Context, req *kmsg.DeleteTopicsRequest) (*kmsg.DeleteTopicsResponse, error) {