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

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"
@@ -427,6 +428,16 @@ ss::future<> server::apply(ss::lw_shared_ptr<net::connection> conn) {
     }
 }
 
+bool server::is_message_error(std::exception_ptr e) const {
+    try {
+        std::rethrow_exception(e);
+    } catch (const kafka::malformed_header_exception& e) {
+        return true;
+    } catch (...) {
+        return false;
+    }
+}
+
 template<>
 ss::future<response_ptr> heartbeat_handler::handle(
   request_context ctx, [[maybe_unused]] ss::smp_service_group g) {

src/v/kafka/server/server.h

@@ -91,6 +91,8 @@ class server final
     // until the end of the server (container/parent)
     ss::future<> apply(ss::lw_shared_ptr<net::connection>) final;
 
+    bool is_message_error(std::exception_ptr) const final;
+
     ss::smp_service_group smp_group() const { return _smp_group; }
 
     /**

src/v/net/server.cc

@@ -151,10 +151,17 @@ void server::print_exceptional_future(
               ctx,
               address,
               ex);
+        } else if (is_message_error(ex)) {
+            vlog(
+              _log.warn,
+              "Message Error[{}] remote address: {} - {}",
+              ctx,
+              address,
+              ex);
         } else {
-            // Authentication exceptions are logged at WARN, not ERROR, because
-            // they generally point to a misbehaving client rather than a fault
-            // in the server.
+            // Authentication and Message exceptions are logged at WARN, not
+            // ERROR, because they generally point to a misbehaving client
+            // rather than a fault in the server.
             vlog(
               _log.error,
               "Error[{}] remote address: {} - {}",

src/v/net/server.h

@@ -146,6 +146,15 @@ class server {
     ss::abort_source& abort_source() { return _as; }
     bool abort_requested() const { return _as.abort_requested(); }
 
+    /**
+       Called by print_exceptional_future.
+       When this function returns 'true', the exception will be logged at WARN
+       level rather than the usual ERROR. This provides a customization point
+       for derived classes to avoid producing error logs for certain exception
+       types which may be application specific.
+     */
+    virtual bool is_message_error(std::exception_ptr) const { return false; };
+
 private:
     struct listener {
         ss::sstring name;