kafka: implement Metadata v13 (KIP-1102); bump kcl to v0.18 in tests

src/v/kafka/client/errors.h

@@ -113,6 +113,7 @@ inline bool is_retriable_error(kafka::error_code ec) {
     case error_code::duplicate_resource:
     case error_code::unacceptable_credential:
     case error_code::transactional_id_not_found:
+    case error_code::rebootstrap_required:
         return false;
     }
 }

src/v/kafka/protocol/errors.cc

@@ -201,6 +201,8 @@ std::string_view error_code_to_str(error_code error) {
         return "unknown_topic_id";
     case error_code::transactional_id_not_found:
         return "transactional_id_not_found";
+    case error_code::rebootstrap_required:
+        return "rebootstrap_required";
     default:
         return "unknown_error_code";
     }
@@ -319,6 +321,7 @@ bool is_retriable(error_code error) {
     case error_code::duplicate_resource:
     case error_code::unacceptable_credential:
     case error_code::transactional_id_not_found:
+    case error_code::rebootstrap_required:
         break;
     }
     return false;

src/v/kafka/protocol/errors.h

@@ -240,6 +240,9 @@ enum class error_code : int16_t {
     unknown_topic_id = 100,
     // The transactional_id could not be found for describe tx request.
     transactional_id_not_found = 105,
+    // Client metadata is stale; the client should rebootstrap to obtain new
+    // metadata. Introduced by KIP-1102.
+    rebootstrap_required = 129,
 };
 
 std::string_view error_code_to_str(error_code error);

src/v/kafka/protocol/metadata.h

@@ -51,7 +51,7 @@ struct metadata_request {
 
     metadata_request copy() const {
         static_assert(
-          api_type::max_valid == api_version(12),
+          api_type::max_valid == api_version(13),
           "Please update the metadata_request::copy method when updating the "
           "Metadata API");
         return {

src/v/kafka/protocol/schemata/metadata_request.json

@@ -18,7 +18,7 @@
   "type": "request",
   "listeners": ["zkBroker", "broker"],
   "name": "MetadataRequest",
-  "validVersions": "0-12",
+  "validVersions": "0-13",
   "flexibleVersions": "9+",
   "fields": [
     // In version 0, an empty array indicates "request metadata for all topics."  In version 1 and
@@ -39,6 +39,8 @@
     // Version 11 deprecates IncludeClusterAuthorizedOperations field. This is now exposed
     // by the DescribeCluster API (KIP-700).
     // Version 12 supports topic Id.
+    //
+    // Version 13 has no request-side changes (top-level ErrorCode added to the response only, KIP-1102).
     { "name": "Topics", "type": "[]MetadataRequestTopic", "versions": "0+", "nullableVersions": "1+",
       "about": "The topics to fetch metadata for.", "fields": [
       { "name": "TopicId", "type": "uuid", "versions": "10+", "ignorable": true, "about": "The topic id." },

src/v/kafka/protocol/schemata/metadata_response.json

@@ -42,7 +42,9 @@
   // Version 11 deprecates ClusterAuthorizedOperations. This is now exposed
   // by the DescribeCluster API (KIP-700).
   // Version 12 supports topicId.
-  "validVersions": "0-12",
+  //
+  // Version 13 supports top-level error code in the response (KIP-1102).
+  "validVersions": "0-13",
   "flexibleVersions": "9+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "3+", "ignorable": true,
@@ -93,6 +95,8 @@
         "about": "32-bit bitfield to represent authorized operations for this topic." }
     ]},
     { "name": "ClusterAuthorizedOperations", "type": "int32", "versions": "8-10", "default": "-2147483648",
-      "about": "32-bit bitfield to represent authorized operations for this cluster." }
+      "about": "32-bit bitfield to represent authorized operations for this cluster." },
+    { "name": "ErrorCode", "type": "int16", "versions": "13+", "ignorable": true,
+      "about": "The top-level error code, or 0 if there was no error." }
   ]
 }

src/v/kafka/server/handlers/metadata.h

@@ -27,7 +27,7 @@ memory_estimate_fn metadata_memory_estimator;
 using metadata_handler = single_stage_handler<
   metadata_api,
   0,
-  12,
+  13,
   metadata_memory_estimator,
   default_scheduling_group_provider,
   latency_hist::yes>;

src/v/pandaproxy/error.cc

@@ -277,6 +277,7 @@ std::error_condition make_error_condition(std::error_code ec) {
         case kec::group_subscribed_to_topic:
         case kec::unstable_offset_commit:
         case kec::no_reassignment_in_progress:
+        case kec::rebootstrap_required:
             return rec::kafka_error;
         case kec::network_exception:
         case kec::coordinator_load_in_progress:

tests/docker/ducktape-deps/kcl

@@ -1,4 +1,4 @@
 #!/usr/bin/env bash
 set -e
-go install github.com/twmb/kcl@v0.16.0
+go install github.com/twmb/kcl@v0.18.0
 mv /root/go/bin/kcl /usr/local/bin/