dubbo: Fix heartbeat packet parsing error (#8103)

Description: 
The heartbeat packet may carry data, and it is treated as null data when processing the heartbeat packet, causing some data to remain in the buffer.

Risk Level: low 
Testing: Existing unit test
Docs Changes: N/A
Release Notes: N/A
Fixes #7970 

Signed-off-by: tianqian.zyf <tianqian.zyf@alibaba-inc.com>

source/extensions/filters/network/dubbo_proxy/decoder.cc

@@ -18,12 +18,16 @@ DecoderStateMachine::onDecodeStreamHeader(Buffer::Instance& buffer) {
     return {ProtocolState::WaitForData};
   }
 
-  // The heartbeat message has no body.
   auto context = ret.first;
   if (metadata->message_type() == MessageType::HeartbeatRequest ||
       metadata->message_type() == MessageType::HeartbeatResponse) {
+    if (buffer.length() < (context->header_size() + context->body_size())) {
+      ENVOY_LOG(debug, "dubbo decoder: need more data for {} protocol heartbeat", protocol_.name());
+      return {ProtocolState::WaitForData};
+    }
+
     ENVOY_LOG(debug, "dubbo decoder: this is the {} heartbeat message", protocol_.name());
-    buffer.drain(context->header_size());
+    buffer.drain(context->header_size() + context->body_size());
     delegate_.onHeartbeat(metadata);
     return {ProtocolState::Done};
   }

test/extensions/filters/network/dubbo_proxy/conn_manager_test.cc

@@ -306,7 +306,8 @@ class ConnectionManagerTest : public testing::Test {
     buffer.add(static_cast<void*>(&msg_type), 1);
     buffer.add(std::string{0x14});
     addInt64(buffer, request_id);                    // Request Id
-    buffer.add(std::string{0x00, 0x00, 0x00, 0x00}); // Body Length
+    buffer.add(std::string{0x00, 0x00, 0x00, 0x01}); // Body Length
+    buffer.add(std::string{0x01});                   // Body
   }
 
   NiceMock<Server::Configuration::MockFactoryContext> factory_context_;
@@ -377,6 +378,7 @@ TEST_F(ConnectionManagerTest, OnDataHandlesHeartbeatEvent) {
       }));
 
   EXPECT_EQ(conn_manager_->onData(buffer_, false), Network::FilterStatus::StopIteration);
+  EXPECT_EQ(0U, buffer_.length());
   filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList();
 
   EXPECT_EQ(0U, store_.counter("test.request").value());