DO NOT MERGE: Basically fix "Unexpected EOF" log errors

But everything is in the wrong place.
Not particularly practical to test these locally.

In principle, fixes:

CORE-8134
CORE-8059
CORE-7883
CORE-7324
CORE-6895

src/v/kafka/server/protocol_utils.cc

@@ -56,7 +56,7 @@ parse_v1_header(ss::input_stream<char>& src) {
 
     if (unlikely(client_id_size < 0)) {
         // header parsing error, force connection shutdown
-        throw std::runtime_error(
+        throw malformed_header_exception(
           fmt::format("Invalid client_id size {}", client_id_size));
     }
 

src/v/kafka/server/protocol_utils.h

@@ -26,4 +26,10 @@ ss::future<std::optional<request_header>> parse_header(ss::input_stream<char>&);
 
 ss::scattered_message<char> response_as_scattered(response_ptr response);
 
+class malformed_header_exception : public std::runtime_error {
+public:
+    explicit malformed_header_exception(const std::string& m)
+      : std::runtime_error(m) {}
+};
+
 } // namespace kafka

src/v/kafka/server/server.cc

@@ -49,6 +49,7 @@
 #include "kafka/server/handlers/sasl_handshake.h"
 #include "kafka/server/handlers/sync_group.h"
 #include "kafka/server/logger.h"
+#include "kafka/server/protocol_utils.h"
 #include "kafka/server/quota_manager.h"
 #include "kafka/server/request_context.h"
 #include "kafka/server/response.h"

src/v/net/connection.cc

@@ -10,6 +10,7 @@
 #include "net/connection.h"
 
 #include "base/seastarx.h"
+#include "kafka/server/protocol_utils.h"
 #include "net/exceptions.h"
 #include "net/types.h"
 #include "ssx/abort_source.h"
@@ -130,6 +131,16 @@ bool is_auth_error(std::exception_ptr e) {
     __builtin_unreachable();
 }
 
+bool is_message_error(std::exception_ptr e) {
+    try {
+        std::rethrow_exception(e);
+    } catch (const kafka::malformed_header_exception& e) {
+        return true;
+    } catch (...) {
+        return false;
+    }
+}
+
 connection::connection(
   boost::intrusive::list<connection>& hook,
   ss::sstring name,

src/v/net/connection.h

@@ -33,6 +33,7 @@ bool is_reconnect_error(const std::system_error& e);
 std::optional<ss::sstring> is_disconnect_exception(std::exception_ptr);
 
 bool is_auth_error(std::exception_ptr);
+bool is_message_error(std::exception_ptr);
 
 class connection : public boost::intrusive::list_base_hook<> {
 public: