KAFKA-14737: Move kafka.utils.json to server-common

build.gradle

@@ -1569,6 +1569,7 @@ project(':server-common') {
     implementation libs.slf4jApi
     implementation libs.metrics
     implementation libs.joptSimple
+    implementation libs.jacksonDatabind
     implementation libs.pcollections
 
     testImplementation project(':clients')

checkstyle/import-control-server-common.xml

@@ -89,6 +89,8 @@
             <allow class="org.apache.kafka.clients.ClientResponse" />
             <allow class="org.apache.kafka.clients.KafkaClient" />
             <allow class="org.apache.kafka.clients.RequestCompletionHandler" />
+            <allow pkg="com.fasterxml.jackson" />
+            <allow pkg="org.apache.kafka.server.util.json" />
 
             <subpackage name="timer">
                 <allow class="org.apache.kafka.server.util.MockTime" />

server-common/src/main/java/org/apache/kafka/server/util/Json.java

@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.server.util;
+
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.MissingNode;
+import org.apache.kafka.server.util.json.JsonValue;
+
+import java.io.IOException;
+import java.util.Optional;
+
+/**
+ * Provides methods for parsing JSON with Jackson and encoding to JSON with a simple and naive custom implementation.
+ */
+public final class Json {
+    private static ObjectMapper mapper = new ObjectMapper();
+
+    /**
+     * Parse a JSON string into a JsonValue if possible. `None` is returned if `input` is not valid JSON.
+     */
+    public static Optional<JsonValue> parseFull(String input) {
+        try {
+            return Optional.ofNullable(tryParseFull(input));
+        } catch (JsonProcessingException e) {
+            return Optional.empty();
+        }
+    }
+
+    /**
+     * Parse a JSON string into a generic type T, or throw JsonProcessingException in the case of
+     * exception.
+     */
+    public static <T> T parseStringAs(String input, Class<T> clazz) throws JsonProcessingException {
+        return mapper.readValue(input, clazz);
+    }
+
+    /**
+     * Parse a JSON byte array into a JsonValue if possible. `None` is returned if `input` is not valid JSON.
+     */
+    public static Optional<JsonValue> parseBytes(byte[] input) throws IOException {
+        try {
+            return Optional.ofNullable(mapper.readTree(input)).map(JsonValue::apply);
+        } catch (JsonProcessingException e) {
+            return Optional.empty();
+        }
+    }
+
+    public static JsonValue tryParseBytes(byte[] input) throws IOException {
+        return JsonValue.apply(mapper.readTree(input));
+    }
+
+    /**
+     * Parse a JSON byte array into a generic type T, or throws a JsonProcessingException in the case of exception.
+     */
+    public static <T> T parseBytesAs(byte[] input, Class<T> clazz) throws IOException {
+        return mapper.readValue(input, clazz);
+    }
+
+    /**
+     * Parse a JSON string into a JsonValue if possible.
+     * @param input a JSON string to parse
+     * @return the actual json value.
+     * @throws JsonProcessingException if failed to parse
+     */
+    public static JsonValue tryParseFull(String input) throws JsonProcessingException {
+        if (input == null || input.isEmpty()) {
+            throw new JsonParseException(MissingNode.getInstance().traverse(), "The input string shouldn't be empty");
+        } else {
+            return JsonValue.apply(mapper.readTree(input));
+        }
+    }
+
+    /**
+     * Encode an object into a JSON string. This method accepts any type supported by Jackson's ObjectMapper in
+     * the default configuration. That is, Java collections are supported, but Scala collections are not (to avoid
+     * a jackson-scala dependency).
+     */
+    public static String encodeAsString(Object obj) throws JsonProcessingException {
+        return mapper.writeValueAsString(obj);
+    }
+
+    /**
+     * Encode an object into a JSON value in bytes. This method accepts any type supported by Jackson's ObjectMapper in
+     * the default configuration. That is, Java collections are supported, but Scala collections are not (to avoid
+     * a jackson-scala dependency).
+     */
+    public static byte[] encodeAsBytes(Object obj) throws JsonProcessingException {
+        return mapper.writeValueAsBytes(obj);
+    }
+}

server-common/src/main/java/org/apache/kafka/server/util/json/DecodeJson.java

@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.server.util.json;
+
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+public interface DecodeJson<T> {
+    /**
+     * Decode the JSON node provided into an instance of `T`.
+     *
+     * @throws JsonMappingException if `node` cannot be decoded into `T`.
+     */
+    T decode(JsonNode node) throws JsonMappingException;
+
+    static JsonMappingException throwJsonMappingException(String expectedType, JsonNode node) {
+        return new JsonMappingException(null, String.format("Expected `%s` value, received %s", expectedType, node));
+    }
+
+    final class DecodeBoolean implements DecodeJson<Boolean> {
+        @Override
+        public Boolean decode(JsonNode node) throws JsonMappingException {
+            if (node.isBoolean()) {
+                return node.booleanValue();
+            }
+            throw throwJsonMappingException(Boolean.class.getSimpleName(), node);
+        }
+    }
+
+    final class DecodeDouble implements DecodeJson<Double> {
+        @Override
+        public Double decode(JsonNode node) throws JsonMappingException {
+            if (node.isDouble() || node.isLong() || node.isInt()) {
+                return node.doubleValue();
+            }
+            throw throwJsonMappingException(Double.class.getSimpleName(), node);
+        }
+    }
+
+    final class DecodeInteger implements DecodeJson<Integer> {
+        @Override
+        public Integer decode(JsonNode node) throws JsonMappingException {
+            if (node.isInt()) {
+                return node.intValue();
+            }
+            throw throwJsonMappingException(Integer.class.getSimpleName(), node);
+        }
+    }
+
+    final class DecodeLong implements DecodeJson<Long> {
+        @Override
+        public Long decode(JsonNode node) throws JsonMappingException {
+            if (node.isLong() || node.isInt()) {
+                return node.longValue();
+            }
+            throw throwJsonMappingException(Long.class.getSimpleName(), node);
+        }
+    }
+
+    final class DecodeString implements DecodeJson<String> {
+        @Override
+        public String decode(JsonNode node) throws JsonMappingException {
+            if (node.isTextual()) {
+                return node.textValue();
+            }
+            throw throwJsonMappingException(String.class.getSimpleName(), node);
+        }
+    }
+
+    static <E> DecodeJson<Optional<E>> decodeOptional(DecodeJson<E> decodeJson) {
+        return node -> {
+            if (node.isNull()) return Optional.empty();
+            return Optional.of(decodeJson.decode(node));
+        };
+    }
+
+    static <E> DecodeJson<List<E>> decodeList(DecodeJson<E> decodeJson) throws JsonMappingException {
+        return node -> {
+            if (node.isArray()) {
+                List<E> result = new ArrayList<>();
+                Iterator<JsonNode> elements = node.elements();
+                while (elements.hasNext()) {
+                    result.add(decodeJson.decode(elements.next()));
+                }
+                return result;
+            }
+            throw throwJsonMappingException("JSON array", node);
+        };
+    }
+
+    static <V> DecodeJson<Map<String, V>> decodeMap(DecodeJson<V> decodeJson) throws JsonMappingException {
+        return node -> {
+            if (node.isObject()) {
+                Map<String, V> result = new HashMap<>();
+                Iterator<Map.Entry<String, JsonNode>> elements = node.fields();
+                while (elements.hasNext()) {
+                    Map.Entry<String, JsonNode> next = elements.next();
+                    result.put(next.getKey(), decodeJson.decode(next.getValue()));
+                }
+                return result;
+            }
+            throw throwJsonMappingException("JSON object", node);
+        };
+    }
+}

server-common/src/main/java/org/apache/kafka/server/util/json/JsonArray.java

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.server.util.json;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+
+import java.util.Iterator;
+import java.util.Spliterator;
+import java.util.Spliterators;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+public class JsonArray implements JsonValue {
+    protected final ArrayNode node;
+
+    JsonArray(ArrayNode node) {
+        this.node = node;
+    }
+
+    @Override
+    public JsonNode node() {
+        return node;
+    }
+
+    public Iterator<JsonValue> iterator() {
+        Stream<JsonNode> nodeStream = StreamSupport.stream(
+                Spliterators.spliteratorUnknownSize(node.elements(), Spliterator.ORDERED),
+                false);
+        Stream<JsonValue> results = nodeStream.map(node -> JsonValue.apply(node));
+        return results.collect(Collectors.toList()).iterator();
+    }
+
+    @Override
+    public int hashCode() {
+        return node().hashCode();
+    }
+
+    @Override
+    public boolean equals(Object a) {
+        if (a instanceof JsonArray) {
+            return node().equals(((JsonArray) a).node());
+        }
+        return false;
+    }
+
+    @Override
+    public String toString() {
+        return node().toString();
+    }
+}