Added handling members with group.instance.id set i.e. static members

src/v/kafka/client/test/fixture.h

@@ -20,7 +20,8 @@ namespace kc = kafka::client;
 class kafka_client_fixture : public redpanda_thread_fixture {
 public:
     void restart() {
-        app.shutdown();
+        shutdown();
+        app_signal = std::make_unique<::stop_signal>();
         ss::smp::invoke_on_all([this] {
             auto& config = config::shard_local_cfg();
             config.get("disable_metrics").set_value(false);
@@ -29,8 +30,7 @@ class kafka_client_fixture : public redpanda_thread_fixture {
         app.check_environment();
         app.configure_admin_server();
         app.wire_up_services();
-        ::stop_signal app_signal;
-        app.start(app_signal);
+        app.start(*app_signal);
     }
 
     kc::client make_client() { return kc::client{proxy_client_config()}; }

src/v/kafka/protocol/leave_group.h

@@ -37,6 +37,8 @@ struct leave_group_request final {
 
     // set during request processing after mapping group to ntp
     model::ntp ntp;
+    // additional context, set in decode
+    api_version version;
 
     void encode(response_writer& writer, api_version version) {
         data.encode(writer, version);

src/v/kafka/protocol/schemata/generator.py

@@ -81,6 +81,7 @@
             },
         },
         "MemberId": ("kafka::member_id", "string"),
+        "GroupInstanceId": ("kafka::group_instance_id", "string"),
     },
     "AddPartitionsToTxnRequestData": {
         "Topics": {