fix: fix response earlier release (#555)

Signed-off-by: Robin Han <[email protected]>
AutoMQ · Dec 13, 2023 · 59c6044 · 59c6044
1 parent 20fd237
commit 59c6044
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 3 deletions.
diff --git a/clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java b/clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java
@@ -380,6 +380,7 @@ public void setSend(NetworkSend send) {
 
     public NetworkSend maybeCompleteSend() {
         if (send != null && send.completed()) {
+            send.release();
             midWrite = false;
             transportLayer.removeInterestOps(SelectionKey.OP_WRITE);
             NetworkSend result = send;
@@ -432,7 +433,12 @@ public long write() throws IOException {
             return 0;
 
         midWrite = true;
-        return send.writeTo(transportLayer);
+        try {
+            return send.writeTo(transportLayer);
+        } catch (IOException e) {
+            send.release();
+            throw e;
+        }
     }
 
     /**

diff --git a/core/src/main/scala/kafka/log/streamaspect/ElasticLogFileRecords.java b/core/src/main/scala/kafka/log/streamaspect/ElasticLogFileRecords.java
@@ -345,6 +345,8 @@ public void release() {
             if (!freed) {
                 pack.release();
                 freed = true;
+            } else {
+                LOGGER.warn("PooledMemoryRecords[{}] has been freed", this, new RuntimeException());
             }
         }
 

diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala
@@ -1146,7 +1146,6 @@ private[kafka] class Processor(
         if (response == null) {
           throw new IllegalStateException(s"Send for ${send.destinationId} completed, but not in `inflightResponses`")
         }
-        send.release()
         updateRequestMetrics(response)
 
         // Invoke send completion callback
@@ -1184,7 +1183,6 @@ private[kafka] class Processor(
         }.remoteHost
         inflightResponses.entrySet().removeIf(e => {
           val remove = connectionId.equals(e.getValue.request.context.connectionId)
-          e.getKey.release()
           remove
         })
         // the channel has been closed by the selector but the quotas still need to be updated