[kafka] branch trunk updated: KAFKA-14794; Decode base64 JSON string (#13363)

[jsancio]

This is an automated email from the ASF dual-hosted git repository.

jsancio pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 00c7a018b8b KAFKA-14794; Decode base64 JSON string (#13363)
00c7a018b8b is described below

commit 00c7a018b8b1f8676290cd3bc41cbf6a2527bb89
Author: José Armando García Sancio <jsan...@users.noreply.github.com>
AuthorDate: Wed Mar 8 18:40:29 2023 -0800

    KAFKA-14794; Decode base64 JSON string (#13363)
    
    A binary value (array of bytes) can be a BinaryNode or a TextNode. When it 
is a BinaryNode, the method binaryValue() always returns non-null. When it is a 
TextNode, the method binaryValue() will return non-null if the value is a 
base64 string. For all other JSON nodes binaryValue() returns null.
    
    Reviewers: Colin Patrick McCabe <cmcc...@apache.org>
---
 .../apache/kafka/common/protocol/MessageUtil.java  | 10 +++--
 .../kafka/common/protocol/MessageUtilTest.java     | 44 +++++++++++++++++++---
 2 files changed, 45 insertions(+), 9 deletions(-)

diff --git 
a/clients/src/main/java/org/apache/kafka/common/protocol/MessageUtil.java 
b/clients/src/main/java/org/apache/kafka/common/protocol/MessageUtil.java
index b366ebd8ead..95bc01945ec 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/MessageUtil.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/MessageUtil.java
@@ -22,6 +22,7 @@ import org.apache.kafka.common.protocol.types.RawTaggedField;
 import org.apache.kafka.common.utils.Utils;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Iterator;
@@ -161,14 +162,15 @@ public final class MessageUtil {
     }
 
     public static byte[] jsonNodeToBinary(JsonNode node, String about) {
-        if (!node.isBinary()) {
-            throw new RuntimeException(about + ": expected Base64-encoded 
binary data.");
-        }
         try {
             byte[] value = node.binaryValue();
+            if (value == null) {
+                throw new IllegalArgumentException(about + ": expected 
Base64-encoded binary data.");
+            }
+
             return value;
         } catch (IOException e) {
-            throw new RuntimeException(about + ": unable to retrieve 
Base64-encoded binary data", e);
+            throw new UncheckedIOException(about + ": unable to retrieve 
Base64-encoded binary data", e);
         }
     }
 
diff --git 
a/clients/src/test/java/org/apache/kafka/common/protocol/MessageUtilTest.java 
b/clients/src/test/java/org/apache/kafka/common/protocol/MessageUtilTest.java
index 5195f551157..05640f26b73 100755
--- 
a/clients/src/test/java/org/apache/kafka/common/protocol/MessageUtilTest.java
+++ 
b/clients/src/test/java/org/apache/kafka/common/protocol/MessageUtilTest.java
@@ -17,18 +17,25 @@
 
 package org.apache.kafka.common.protocol;
 
-import org.apache.kafka.common.protocol.types.RawTaggedField;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.Timeout;
-
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.BinaryNode;
+import com.fasterxml.jackson.databind.node.IntNode;
+import com.fasterxml.jackson.databind.node.TextNode;
+import java.io.IOException;
+import java.io.StringWriter;
+import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Collections;
-
+import org.apache.kafka.common.protocol.types.RawTaggedField;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @Timeout(120)
@@ -83,4 +90,31 @@ public final class MessageUtilTest {
         assertEquals(MessageUtil.UNSIGNED_SHORT_MAX, 0xFFFF);
         assertEquals(MessageUtil.UNSIGNED_INT_MAX, 0xFFFFFFFFL);
     }
+
+    @Test
+    public void testBinaryNode() throws IOException {
+        byte[] expected = new byte[] {5, 2, 9, 4, 1, 8, 7, 0, 3, 6};
+        StringWriter writer = new StringWriter();
+        ObjectMapper mapper = new ObjectMapper();
+
+        mapper.writeTree(mapper.createGenerator(writer), new 
BinaryNode(expected));
+
+        JsonNode textNode = mapper.readTree(writer.toString());
+
+        assertTrue(textNode.isTextual(), String.format("Expected a JSON string 
but was: %s", textNode.toString()));
+        byte[] actual = MessageUtil.jsonNodeToBinary(textNode, "Test base64 
JSON string");
+        assertArrayEquals(expected, actual);
+    }
+
+    @Test
+    public void testInvalidBinaryNode() {
+        assertThrows(
+            IllegalArgumentException.class,
+            () -> MessageUtil.jsonNodeToBinary(new IntNode(42), "Test int to 
binary")
+        );
+        assertThrows(
+            UncheckedIOException.class,
+            () -> MessageUtil.jsonNodeToBinary(new TextNode("This is not 
base64!"), "Test non-base64 to binary")
+        );
+    }
 }