[CORE 8273] Adjust license warning text to include trial license links

src/v/cluster/BUILD

@@ -616,6 +616,9 @@ redpanda_cc_library(
         "types.h",
         "version.h",
     ],
+    implementation_deps = [
+        "//src/v/features:enterprise_feature_messages",
+    ],
     include_prefix = "cluster",
     visibility = ["//visibility:public"],
     deps = [

src/v/cluster/feature_manager.cc

@@ -23,6 +23,7 @@
 #include "config/node_config.h"
 #include "config/types.h"
 #include "config/validators.h"
+#include "features/enterprise_feature_messages.h"
 #include "features/feature_state.h"
 #include "features/feature_table.h"
 #include "model/timeout_clock.h"
@@ -304,12 +305,9 @@ ss::future<> feature_manager::maybe_log_license_check_info() {
         if (_feature_table.local().should_sanction()) {
             vlog(
               clusterlog.warn,
-              "A Redpanda Enterprise Edition license is required to use "
-              "enterprise features: ([{}]). Enter an active license key (for "
-              "example, rpk cluster license set <key>). To request a license, "
-              "see https://redpanda.com/license-request. For more information, "
-              "see https://docs.redpanda.com/current/get-started/licenses.",
-              fmt::join(enterprise_features.enabled(), ", "));
+              "{}",
+              features::enterprise_error_message::license_nag(
+                enterprise_features.enabled()));
         }
     }
 }
@@ -363,14 +361,9 @@ void feature_manager::verify_enterprise_license() {
       fallback_license ? " (detected fallback license)" : "");
 
     if (enterprise_features.any() && license_missing_or_expired) {
-        throw std::runtime_error{fmt::format(
-          "A Redpanda Enterprise Edition license is required to use enterprise "
-          "features: ([{}]). To add your license, downgrade this broker to the "
-          "pre-upgrade version, and enter the active license key (for example, "
-          "rpk cluster license set). To request a license, see "
-          "https://redpanda.com/license-request. For more information, see "
-          "https://docs.redpanda.com/current/get-started/licenses.",
-          fmt::join(enterprise_features.enabled(), ", "))};
+        throw std::runtime_error{
+          features::enterprise_error_message::upgrade_failure(
+            enterprise_features.enabled())};
     }
 
     _verified_enterprise_license.signal();

src/v/cluster/partition_balancer_backend.cc

@@ -23,6 +23,7 @@
 #include "cluster/types.h"
 #include "config/configuration.h"
 #include "config/property.h"
+#include "features/enterprise_feature_messages.h"
 #include "features/enterprise_features.h"
 #include "features/feature_table.h"
 #include "model/metadata.h"
@@ -370,9 +371,9 @@ ss::future<> partition_balancer_backend::do_tick() {
     if (is_sanctioned) {
         vlog(
           clusterlog.warn,
-          "A Redpanda Enterprise Edition license is required to use enterprise "
-          "feature \"partition_autobalancing_mode\" with value \"{}\". "
-          "Behavior is being restricted to \"{}\".",
+          "{}",
+          features::enterprise_error_message::
+            partition_autobalancing_continuous(),
           _mode(false).first,
           mode);
     }

src/v/cluster/scheduling/leader_balancer.cc

@@ -25,6 +25,7 @@
 #include "config/configuration.h"
 #include "config/node_config.h"
 #include "container/fragmented_vector.h"
+#include "features/enterprise_feature_messages.h"
 #include "model/metadata.h"
 #include "model/namespace.h"
 #include "raft/rpc_client_protocol.h"
@@ -465,12 +466,7 @@ bool leader_balancer::leadership_pinning_enabled() const {
             vlog(
               clusterlog.warn,
               "{}",
-              "A Redpanda Enterprise Edition license is required to use the "
-              "enterprise feature \"leadership pinning\". This feature is "
-              "disabled. The values of the cluster property \"{}\" and the "
-              "topic property \"redpanda.leaders.preference\" are being "
-              "ignored.",
-              config::shard_local_cfg().default_leaders_preference.name());
+              features::enterprise_error_message::default_leaders_preference());
         }
         return false;
     }

src/v/cluster/security_frontend.cc

@@ -21,6 +21,7 @@
 #include "cluster/logger.h"
 #include "cluster/partition_leaders_table.h"
 #include "cluster/types.h"
+#include "features/enterprise_feature_messages.h"
 #include "model/errc.h"
 #include "model/metadata.h"
 #include "model/namespace.h"
@@ -235,8 +236,8 @@ ss::future<std::vector<errc>> security_frontend::do_create_acls(
         err = errc::feature_disabled;
         vlog(
           clusterlog.warn,
-          "An enterprise license is required to create an ACL with a role "
-          "binding");
+          "{}",
+          features::enterprise_error_message::acl_with_rbac());
     } else {
         try {
             create_acls_cmd_data data;

src/v/cluster/shard_balancer.cc

@@ -14,6 +14,7 @@
 #include "cluster/cluster_utils.h"
 #include "cluster/logger.h"
 #include "config/node_config.h"
+#include "features/enterprise_feature_messages.h"
 #include "features/enterprise_features.h"
 #include "random/generators.h"
 #include "ssx/async_algorithm.h"
@@ -469,9 +470,8 @@ void shard_balancer::maybe_assign(
         if (is_sanctioned) {
             vlog(
               clusterlog.warn,
-              "A Redpanda Enterprise Edition license is required to use "
-              "enterprise feature \"core_balancing_continuous\". "
-              "This property is being ignored.");
+              "{}",
+              features::enterprise_error_message::core_balancing_continuous());
         }
 
         if (

src/v/cluster/topics_frontend.cc

@@ -34,6 +34,7 @@
 #include "config/configuration.h"
 #include "config/leaders_preference.h"
 #include "data_migration_types.h"
+#include "features/enterprise_feature_messages.h"
 #include "features/feature_table.h"
 #include "fwd.h"
 #include "model/errc.h"
@@ -428,8 +429,8 @@ ss::future<std::vector<topic_result>> topics_frontend::update_topic_properties(
                 && is_user_topic(update.tp_ns)) {
                   if (auto f = get_enterprise_features(_metadata_cache, update);
                       !f.empty()) {
-                      auto msg = ssx::sformat(
-                        "An enterprise license is required to enable {}.", f);
+                      auto msg
+                        = features::enterprise_error_message::topic_property(f);
                       vlog(clusterlog.warn, "{}", msg);
                       return ss::make_ready_future<topic_result>(topic_result(
                         update.tp_ns,
@@ -609,8 +610,7 @@ topic_result topics_frontend::validate_topic_configuration(
       && is_user_topic(assignable_config.cfg.tp_ns)) {
         if (auto f = get_enterprise_features(assignable_config.cfg);
             !f.empty()) {
-            auto msg = ssx::sformat(
-              "An enterprise license is required to enable {}.", f);
+            auto msg = features::enterprise_error_message::topic_property(f);
             vlog(clusterlog.warn, "{}", msg);
             return make_result(errc::topic_invalid_config, std::move(msg));
         }
@@ -1635,9 +1635,7 @@ ss::future<topic_result> topics_frontend::do_create_partition(
 
     if (_features.local().should_sanction() && is_user_topic(tp_cfg->tp_ns)) {
         if (auto f = get_enterprise_features(*tp_cfg); !f.empty()) {
-            auto msg = ssx::sformat(
-              "An enterprise license is required to create partitions with {}.",
-              f);
+            auto msg = features::enterprise_error_message::create_partition(f);
             vlog(clusterlog.warn, "{}", msg);
             co_return make_error_result(
               p_cfg.tp_ns, errc::topic_invalid_config, std::move(msg));

src/v/config/BUILD

@@ -48,6 +48,7 @@ redpanda_cc_library(
     deps = [
         "//src/v/base",
         "//src/v/container:intrusive",
+        "//src/v/features:enterprise_feature_messages",
         "//src/v/json",
         "//src/v/model",
         "//src/v/pandaproxy:config",

src/v/config/property.h

@@ -15,6 +15,7 @@
 #include "config/base_property.h"
 #include "config/rjson_serialization.h"
 #include "container/intrusive_list_helpers.h"
+#include "features/enterprise_feature_messages.h"
 #include "json/stringbuffer.h"
 #include "json/writer.h"
 #include "pandaproxy/schema_registry/schema_id_validation.h"
@@ -1084,8 +1085,8 @@ class enterprise : public P {
         if (do_check_restricted(v)) {
             return std::make_optional<validation_error>(
               P::name().data(),
-              ssx::sformat(
-                "'{}' is restricted to enterprise licensed clusters", v));
+              features::enterprise_error_message::cluster_property(
+                P::name().data(), v));
         }
         return std::nullopt;
     }

src/v/features/BUILD

@@ -1,5 +1,20 @@
 load("//bazel:build.bzl", "redpanda_cc_library")
 
+redpanda_cc_library(
+    name = "enterprise_feature_messages",
+    hdrs = [
+        "enterprise_feature_messages.h",
+    ],
+    include_prefix = "features",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//src/v/base",
+        "//src/v/ssx:sformat",
+        "@fmt",
+        "@seastar",
+    ],
+)
+
 redpanda_cc_library(
     name = "features",
     srcs = [

src/v/features/enterprise_feature_messages.h

@@ -0,0 +1,120 @@
+/*
+ * Copyright 2024 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#pragma once
+
+#include "base/seastarx.h"
+#include "ssx/sformat.h"
+
+#include <seastar/core/sstring.hh>
+
+#include <fmt/core.h>
+
+#include <string>
+#include <string_view>
+
+namespace features::enterprise_error_message {
+
+constexpr std::string_view required
+  = "A Redpanda Enterprise Edition license is required";
+
+constexpr std::string_view request_or_trial
+  = "To request an Enterprise license, please visit "
+    "https://redpanda.com/upgrade. To try Redpanda Enterprise for 30 days, "
+    "visit https://redpanda.com/try-enterprise.";
+
+constexpr std::string_view get_started
+  = "For more information, see "
+    "<https://docs.redpanda.com/current/get-started/licenses>.";
+
+inline ss::sstring license_nag(const auto& features) {
+    return ssx::sformat(
+      "{} to use the currently enabled feature(s): ({}). Please provide a "
+      "valid license key via rpk using 'rpk cluster license set <key>', or "
+      "via Redpanda Console. {} {}",
+      required,
+      fmt::join(features, ", "),
+      request_or_trial,
+      get_started);
+}
+
+inline constexpr std::string upgrade_failure(const auto& features) {
+    return fmt::format(
+      "{} to use the currently enabled feature(s): ({}). To apply your "
+      "license, downgrade this broker to the pre-upgrade version and provide a "
+      "valid license key via rpk using 'rpk cluster license set <key>', or via "
+      "Redpanda Console. {} {}",
+      required,
+      fmt::join(features, ", "),
+      request_or_trial,
+      get_started);
+}
+
+inline ss::sstring cluster_property(std::string_view name, auto value) {
+    return ssx::sformat(
+      "Rejected {}: '{}' '{}' is restricted to clusters with an Enterprise "
+      "license. {}",
+      name,
+      value,
+      name,
+      request_or_trial);
+}
+
+inline ss::sstring partition_autobalancing_continuous() {
+    return ssx::sformat(
+      R"({} to use feature "partition_autobalancing_mode" with value "continuous".  Behavior is restricted to "node_add". {})",
+      required,
+      request_or_trial);
+}
+
+inline ss::sstring core_balancing_continuous() {
+    return ssx::sformat(
+      R"({} to use enterprise feature "core_balancing_continuous".  This property is being ignored. {})",
+      required,
+      request_or_trial);
+}
+
+inline ss::sstring default_leaders_preference() {
+    return ssx::sformat(
+      R"({} to use the enterprise feature "leadership pinning".  This feature is disabled.  The values of the cluster property "default_leaders_preference" and the topic property "redpanda.leaders.preference" are being ignored. {})",
+      required,
+      request_or_trial);
+}
+
+inline ss::sstring topic_property(auto features) {
+    return ssx::sformat(
+      "{} to enable ({}). {}.",
+      required,
+      fmt::join(features, ", "),
+      request_or_trial);
+}
+
+inline ss::sstring create_partition(auto features) {
+    return ssx::sformat(
+      "{} to create partitions with ({}). {}",
+      required,
+      fmt::join(features, ", "),
+      request_or_trial);
+}
+
+inline ss::sstring acl_with_rbac() {
+    return ssx::sformat(
+      "{} to create an ACL with a role binding. {}",
+      required,
+      request_or_trial);
+}
+
+inline ss::sstring audit_log_fetch() {
+    return ssx::sformat(
+      "{} to consume the audit log topic. {}", required, request_or_trial);
+}
+
+} // namespace features::enterprise_error_message

src/v/kafka/server/BUILD

@@ -153,6 +153,9 @@ redpanda_cc_library(
         "usage_aggregator.h",
         "usage_manager.h",
     ],
+    implementation_deps = [
+        "//src/v/features:enterprise_feature_messages",
+    ],
     include_prefix = "kafka/server",
     visibility = ["//visibility:public"],
     deps = [

src/v/kafka/server/handlers/fetch.cc

@@ -16,6 +16,7 @@
 #include "cluster/shard_table.h"
 #include "config/configuration.h"
 #include "container/fragmented_vector.h"
+#include "features/enterprise_feature_messages.h"
 #include "kafka/data/partition_proxy.h"
 #include "kafka/data/replicated_partition.h"
 #include "kafka/protocol/batch_consumer.h"
@@ -1364,8 +1365,8 @@ class simple_fetch_planner final : public fetch_planner::impl {
                     klog,
                     ss::log_level::warn,
                     rate,
-                    "An enterprise license is required to consume the "
-                    "audit log topic");
+                    "{}",
+                    features::enterprise_error_message::audit_log_fetch());
                   resp_it->set(make_partition_response_error(
                     fp.topic_partition.get_partition(),
                     error_code::unknown_server_error));

src/v/kafka/server/server.cc

@@ -17,6 +17,7 @@
 #include "config/broker_authn_endpoint.h"
 #include "config/configuration.h"
 #include "config/node_config.h"
+#include "features/enterprise_feature_messages.h"
 #include "features/feature_table.h"
 #include "kafka/protocol/errors.h"
 #include "kafka/protocol/schemata/list_groups_response.h"
@@ -1578,13 +1579,13 @@ ss::future<response_ptr> create_acls_handler::handle(
         ss::visit(
           result,
           [&response, &results](size_t i) {
-              auto ec = map_topic_error_code(results[i]);
-              response.data.results.push_back(
-                creatable_acl_result{.error_code = ec});
               if (results[i] == cluster::errc::feature_disabled) {
-                  response.data.results.back().error_message.emplace(
-                    "An enterprise license is required to create an ACL with a "
-                    "role binding");
+                  response.data.results.emplace_back(
+                    error_code::invalid_config,
+                    features::enterprise_error_message::acl_with_rbac());
+              } else {
+                  response.data.results.emplace_back(
+                    map_topic_error_code(results[i]));
               }
           },
           [&response](creatable_acl_result r) {