MINOR: Use NullNode instead of empty TextNode for missing request/response fields (apache#22295)

See apache#22075 (comment) ,
use `NullNode.getInstance()` instead of `new TextNode("")` as the
fallback for missing `request`/`response` fields.

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

Author: majialoong

core/src/test/scala/unit/kafka/network/RequestConvertToJsonTest.scala

@@ -19,7 +19,7 @@ package kafka.network
 
 import java.net.InetAddress
 import java.nio.ByteBuffer
-import com.fasterxml.jackson.databind.node.{BooleanNode, DoubleNode, JsonNodeFactory, LongNode, ObjectNode, TextNode}
+import com.fasterxml.jackson.databind.node.{BooleanNode, DoubleNode, JsonNodeFactory, LongNode, NullNode, ObjectNode, TextNode}
 import org.apache.kafka.common.memory.MemoryPool
 import org.apache.kafka.common.message._
 import org.apache.kafka.common.network.{ClientInformation, ListenerName, NetworkSend}
@@ -58,7 +58,7 @@ class RequestConvertToJsonTest {
     val expectedNode = new ObjectNode(JsonNodeFactory.instance)
     expectedNode.set("isForwarded", if (req.isForwarded) BooleanNode.TRUE else BooleanNode.FALSE)
     expectedNode.set("requestHeader", RequestConvertToJson.requestHeaderNode(req.header))
-    expectedNode.set("request", req.requestLog.orElse(new TextNode("")))
+    expectedNode.set("request", req.requestLog.orElse(NullNode.getInstance()))
 
     val actualNode = RequestConvertToJson.requestDesc(req.header, req.requestLog, req.isForwarded)
 
@@ -84,7 +84,7 @@ class RequestConvertToJsonTest {
     val messageConversionsTimeMs = 9
 
     val expectedNode = RequestConvertToJson.requestDesc(req.header, req.requestLog, req.isForwarded).asInstanceOf[ObjectNode]
-    expectedNode.set("response", res.responseLog.getOrElse(new TextNode("")))
+    expectedNode.set("response", res.responseLog.getOrElse(NullNode.getInstance()))
     expectedNode.set("connection", new TextNode(req.context.connectionId))
     expectedNode.set("totalTimeMs", new DoubleNode(totalTimeMs))
     expectedNode.set("requestQueueTimeMs", new DoubleNode(requestQueueTimeMs))

server/src/main/java/org/apache/kafka/network/RequestConvertToJson.java

@@ -384,6 +384,7 @@
 import com.fasterxml.jackson.databind.node.DoubleNode;
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.databind.node.LongNode;
+import com.fasterxml.jackson.databind.node.NullNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.fasterxml.jackson.databind.node.TextNode;
 
@@ -766,7 +767,7 @@ public static JsonNode requestDesc(RequestHeader header, Optional<JsonNode> requ
         ObjectNode node = JsonNodeFactory.instance.objectNode();
         node.set("isForwarded", isForwarded ? BooleanNode.TRUE : BooleanNode.FALSE);
         node.set("requestHeader", requestHeaderNode(header));
-        node.set("request", requestNode.orElse(new TextNode("")));
+        node.set("request", requestNode.orElse(NullNode.getInstance()));
         return node;
     }
 
@@ -784,7 +785,7 @@ public static JsonNode requestDescMetrics(RequestHeader header, Optional<JsonNod
                                               double responseSendTimeMs, long temporaryMemoryBytes,
                                               double messageConversionsTimeMs) {
         ObjectNode node = (ObjectNode) requestDesc(header, requestNode, isForwarded);
-        node.set("response", responseNode.orElse(new TextNode("")));
+        node.set("response", responseNode.orElse(NullNode.getInstance()));
         node.set("connection", new TextNode(context.connectionId));
         node.set("totalTimeMs", new DoubleNode(totalTimeMs));
         node.set("requestQueueTimeMs", new DoubleNode(requestQueueTimeMs));