Check for trailing content when parsing JSON

core/src/main/java/org/apache/iceberg/MetadataUpdateParser.java

@@ -216,7 +216,7 @@ public static void toJson(MetadataUpdate metadataUpdate, JsonGenerator generator
    */
   public static MetadataUpdate fromJson(String json) {
     try {
-      return fromJson(JsonUtil.mapper().readValue(json, JsonNode.class));
+      return fromJson(JsonUtil.parseJson(json));
     } catch (IOException e) {
       throw new UncheckedIOException("Failed to read JSON string: " + json, e);
     }

core/src/main/java/org/apache/iceberg/PartitionSpecParser.java

@@ -104,7 +104,7 @@ public static PartitionSpec fromJson(Schema schema, String json) {
     return SPEC_CACHE.get(Pair.of(schema.asStruct(), json),
         schemaJsonPair -> {
           try {
-            return fromJson(schema, JsonUtil.mapper().readValue(json, JsonNode.class));
+            return fromJson(schema, JsonUtil.parseJson(json));
           } catch (IOException e) {
             throw new RuntimeIOException(e);
           }
@@ -149,7 +149,7 @@ static PartitionSpec fromJsonFields(Schema schema, int specId, JsonNode json) {
 
   static PartitionSpec fromJsonFields(Schema schema, int specId, String json) {
     try {
-      return fromJsonFields(schema, specId, JsonUtil.mapper().readValue(json, JsonNode.class));
+      return fromJsonFields(schema, specId, JsonUtil.parseJson(json));
     } catch (IOException e) {
       throw new RuntimeIOException(e, "Failed to parse partition spec fields: %s", json);
     }

core/src/main/java/org/apache/iceberg/SchemaParser.java

@@ -274,7 +274,7 @@ public static Schema fromJson(JsonNode json) {
   public static Schema fromJson(String json) {
     return SCHEMA_CACHE.get(json, jsonKey -> {
       try {
-        return fromJson(JsonUtil.mapper().readValue(jsonKey, JsonNode.class));
+        return fromJson(JsonUtil.parseJson(json));
       } catch (IOException e) {
         throw new RuntimeIOException(e);
       }

core/src/main/java/org/apache/iceberg/SnapshotParser.java

@@ -171,7 +171,7 @@ static Snapshot fromJson(FileIO io, JsonNode node) {
 
   public static Snapshot fromJson(FileIO io, String json) {
     try {
-      return fromJson(io, JsonUtil.mapper().readValue(json, JsonNode.class));
+      return fromJson(io, JsonUtil.parseJson(json));
     } catch (IOException e) {
       throw new RuntimeIOException(e, "Failed to read version from json: %s", json);
     }

core/src/main/java/org/apache/iceberg/SnapshotRefParser.java

@@ -74,7 +74,7 @@ public static SnapshotRef fromJson(String json) {
     Preconditions.checkArgument(json != null && !json.isEmpty(), "Cannot parse snapshot ref from invalid JSON: %s",
         json);
     try {
-      return fromJson(JsonUtil.mapper().readValue(json, JsonNode.class));
+      return fromJson(JsonUtil.parseJson(json));
     } catch (IOException e) {
       throw new UncheckedIOException("Failed to parse snapshot ref: " + json, e);
     }

core/src/main/java/org/apache/iceberg/SortOrderParser.java

@@ -143,7 +143,7 @@ public static SortOrder fromJson(Schema schema, JsonNode json) {
 
   public static UnboundSortOrder fromJson(String json) {
     try {
-      return fromJson(JsonUtil.mapper().readValue(json, JsonNode.class));
+      return fromJson(JsonUtil.parseJson(json));
     } catch (IOException e) {
       throw new UncheckedIOException(e);
     }

core/src/main/java/org/apache/iceberg/TableMetadataParser.java

@@ -261,7 +261,7 @@ public static TableMetadata read(FileIO io, String path) {
   public static TableMetadata read(FileIO io, InputFile file) {
     Codec codec = Codec.fromFileName(file.location());
     try (InputStream is = codec == Codec.GZIP ? new GZIPInputStream(file.newStream()) : file.newStream()) {
-      return fromJson(io, file, JsonUtil.mapper().readValue(is, JsonNode.class));
+      return fromJson(io, file, JsonUtil.parseJson(is));
     } catch (IOException e) {
       throw new RuntimeIOException(e, "Failed to read file: %s", file);
     }
@@ -290,7 +290,7 @@ public static TableMetadata fromJson(FileIO io, String json) {
    */
   public static TableMetadata fromJson(FileIO io, String metadataLocation, String json) {
     try {
-      JsonNode node = JsonUtil.mapper().readValue(json, JsonNode.class);
+      JsonNode node = JsonUtil.parseJson(json);
       return fromJson(io, metadataLocation, node);
     } catch (IOException e) {
       throw new UncheckedIOException("Failed to read JSON string: " + json, e);

core/src/main/java/org/apache/iceberg/catalog/TableIdentifierParser.java

@@ -83,7 +83,7 @@ public static TableIdentifier fromJson(String json) {
     Preconditions.checkArgument(!json.isEmpty(),
         "Cannot parse table identifier from invalid JSON: ''");
     try {
-      return fromJson(JsonUtil.mapper().readValue(json, JsonNode.class));
+      return fromJson(JsonUtil.parseJson(json));
     } catch (IOException e) {
       throw new UncheckedIOException(String.format("Cannot parse table identifier from invalid JSON: %s", json), e);
     }

core/src/main/java/org/apache/iceberg/mapping/NameMappingParser.java

@@ -100,7 +100,7 @@ private static void toJson(MappedField field, JsonGenerator generator) throws IO
 
   public static NameMapping fromJson(String json) {
     try {
-      return fromJson(JsonUtil.mapper().readValue(json, JsonNode.class));
+      return fromJson(JsonUtil.parseJson(json));
     } catch (IOException e) {
       throw new RuntimeIOException(e, "Failed to convert version from json: %s", json);
     }

core/src/main/java/org/apache/iceberg/puffin/FileMetadataParser.java

@@ -62,7 +62,7 @@ public static String toJson(FileMetadata fileMetadata, boolean pretty) {
 
   public static FileMetadata fromJson(String json) {
     try {
-      return fromJson(JsonUtil.mapper().readValue(json, JsonNode.class));
+      return fromJson(JsonUtil.parseJson(json));
     } catch (IOException e) {
       throw new UncheckedIOException(e);
     }

core/src/main/java/org/apache/iceberg/rest/auth/OAuth2Util.java

@@ -248,7 +248,7 @@ public static void tokenResponseToJson(OAuthTokenResponse response, JsonGenerato
 
   public static OAuthTokenResponse tokenResponseFromJson(String json) {
     try {
-      return tokenResponseFromJson(JsonUtil.mapper().readValue(json, JsonNode.class));
+      return tokenResponseFromJson(JsonUtil.parseJson(json));
     } catch (IOException e) {
       throw new RuntimeIOException(e);
     }

core/src/main/java/org/apache/iceberg/rest/requests/UpdateRequirementParser.java

@@ -150,7 +150,7 @@ public static void toJson(UpdateRequirement updateRequirement, JsonGenerator gen
    */
   public static UpdateRequirement fromJson(String json) {
     try {
-      return fromJson(JsonUtil.mapper().readValue(json, JsonNode.class));
+      return fromJson(JsonUtil.parseJson(json));
     } catch (IOException e) {
       throw new UncheckedIOException("Failed to read JSON string: " + json, e);
     }

core/src/main/java/org/apache/iceberg/rest/responses/ErrorResponseParser.java

@@ -87,7 +87,7 @@ public static void toJson(ErrorResponse errorResponse, JsonGenerator generator)
    */
   public static ErrorResponse fromJson(String json) {
     try {
-      return fromJson(JsonUtil.mapper().readValue(json, JsonNode.class));
+      return fromJson(JsonUtil.parseJson(json));
     } catch (IOException e) {
       throw new UncheckedIOException("Failed to read JSON string: " + json, e);
     }

core/src/main/java/org/apache/iceberg/util/JsonUtil.java

@@ -21,10 +21,12 @@
 
 import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import java.io.IOException;
+import java.io.InputStream;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -50,6 +52,24 @@ public static ObjectMapper mapper() {
     return MAPPER;
   }
 
+  public static JsonNode parseJson(InputStream input) throws IOException {
+    Preconditions.checkNotNull(input, "input is null");
+    try (JsonParser parser = MAPPER.createParser(input)) {
+      JsonNode parsed = MAPPER.readValue(parser, JsonNode.class);
+      Preconditions.checkArgument(parser.nextToken() == null, "Found characters after the expected end of input");
+      return parsed;
+    }
+  }
+
+  public static JsonNode parseJson(String json) throws IOException {
+    Preconditions.checkNotNull(json, "input is null");
+    try (JsonParser parser = MAPPER.createParser(json)) {
+      JsonNode parsed = MAPPER.readValue(parser, JsonNode.class);
+      Preconditions.checkArgument(parser.nextToken() == null, "Found characters after the expected end of input");
+      return parsed;
+    }
+  }
+
   public static int getInt(String property, JsonNode node) {
     Preconditions.checkArgument(node.has(property), "Cannot parse missing int %s", property);
     JsonNode pNode = node.get(property);

core/src/test/java/org/apache/iceberg/TestSnapshotJson.java

@@ -21,9 +21,11 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.List;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.assertj.core.api.Assertions;
 import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
@@ -148,4 +150,16 @@ public void testJsonConversionWithManifestList() throws IOException {
     Assert.assertNull("Summary should be null", snapshot.summary());
     Assert.assertEquals("Schema ID should match", expected.schemaId(), snapshot.schemaId());
   }
+
+  @Test
+  public void testParseJsonWithTrailingContent() {
+    String validJson = SnapshotParser.toJson(new BaseSnapshot(ops.io(), System.currentTimeMillis(), 1,
+        "file:/tmp/manifest1.avro", "file:/tmp/manifest2.avro"));
+    Assertions.assertThatThrownBy(() -> SnapshotParser.fromJson(ops.io(), validJson + "{}"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Found characters after the expected end of input");
+    Assertions.assertThatThrownBy(() -> SnapshotParser.fromJson(ops.io(), validJson + " some more content"))
+        .isInstanceOf(UncheckedIOException.class)
+        .hasMessage("Failed to read version from json: " + validJson + " some more content");
+  }
 }

core/src/test/java/org/apache/iceberg/TestSnapshotRefParser.java

@@ -19,6 +19,8 @@
 
 package org.apache.iceberg;
 
+import java.io.UncheckedIOException;
+import org.assertj.core.api.Assertions;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -79,6 +81,20 @@ public void testTagFromJsonAllFields() {
     Assert.assertEquals("Should be able to deserialize tag with all fields", ref, SnapshotRefParser.fromJson(json));
   }
 
+  @Test
+  public void testTagFromJsonTrailingContent() {
+    String validJson = "{\"snapshot-id\":1,\"type\":\"tag\"}";
+    Assertions.assertThatThrownBy(() -> SnapshotRefParser.fromJson(validJson + "{}"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Found characters after the expected end of input");
+    Assertions.assertThatThrownBy(() -> SnapshotRefParser.fromJson(validJson + " null"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Found characters after the expected end of input");
+    Assertions.assertThatThrownBy(() -> SnapshotRefParser.fromJson(validJson + " none"))
+        .isInstanceOf(UncheckedIOException.class)
+        .hasMessage("Failed to parse snapshot ref: " + validJson + " none");
+  }
+
   @Test
   public void testBranchFromJsonDefault() {
     String json = "{\"snapshot-id\":1,\"type\":\"branch\"}";

core/src/test/java/org/apache/iceberg/puffin/TestFileMetadataParser.java

@@ -31,8 +31,8 @@ public class TestFileMetadataParser {
   @Test
   public void testInvalidJson() {
     assertThatThrownBy(() -> FileMetadataParser.fromJson((String) null))
-        .isInstanceOf(IllegalArgumentException.class)
-        .hasMessageContaining("argument \"content\" is null");
+        .isInstanceOf(NullPointerException.class)
+        .hasMessageContaining("input is null");
 
     assertThatThrownBy(() -> FileMetadataParser.fromJson(""))
         .isInstanceOf(UncheckedIOException.class)
@@ -45,6 +45,16 @@ public void testInvalidJson() {
     assertThatThrownBy(() -> FileMetadataParser.fromJson("{\"blobs\": []"))
         .isInstanceOf(UncheckedIOException.class)
         .hasMessageContaining("Unexpected end-of-input: expected close marker for Object");
+
+    assertThatThrownBy(() -> FileMetadataParser.fromJson("{\"blobs\": []} {}"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Found characters after the expected end of input");
+
+    assertThatThrownBy(() -> FileMetadataParser.fromJson("{\"blobs\": []} whatever"))
+        .isInstanceOf(UncheckedIOException.class)
+        .hasMessageStartingWith(
+            "com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'whatever': was expecting " +
+                "(JSON String, Number, Array, Object or token 'null', 'true' or 'false')");
   }
 
   @Test

core/src/test/java/org/apache/iceberg/rest/responses/TestOAuthTokenResponse.java

@@ -21,6 +21,7 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
 import java.util.Collections;
 import java.util.Set;
 import org.apache.iceberg.AssertHelpers;
@@ -72,15 +73,15 @@ public void testHasOnlyKnownFields() {
     Set<String> fieldsFromSpec = Sets.newHashSet();
     Collections.addAll(fieldsFromSpec, allFieldsFromSpec());
     try {
-      JsonNode node = JsonUtil.mapper().readValue(serialize(createExampleInstance()), JsonNode.class);
+      JsonNode node = JsonUtil.parseJson(serialize(createExampleInstance()));
       for (String field : fieldsFromSpec) {
         Assert.assertTrue("Should have field: " + field, node.has(field));
       }
 
       for (String field : ((Iterable<? extends String>) node::fieldNames)) {
         Assert.assertTrue("Should not have field: " + field, fieldsFromSpec.contains(field));
       }
-    } catch (JsonProcessingException e) {
+    } catch (IOException e) {
       throw new RuntimeException(e);
     }
   }

core/src/test/java/org/apache/iceberg/util/TestJsonUtil.java

@@ -0,0 +1,104 @@
+/*
+ * 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.iceberg.util;
+
+import com.fasterxml.jackson.core.JsonParseException;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+
+public class TestJsonUtil {
+  @Test
+  public void testParseJsonString() throws IOException {
+    // minimal
+    Assertions.assertThat(JsonUtil.parseJson("{}"))
+        .hasToString("{}");
+
+    Assertions.assertThat(JsonUtil.parseJson("{ \"attribute\": 123  }"))
+        .hasToString("{\"attribute\":123}");
+
+    // leading whitespace
+    Assertions.assertThat(JsonUtil.parseJson("   {}"))
+        .hasToString("{}");
+
+    // trailing whitespace
+    Assertions.assertThat(JsonUtil.parseJson("{}  "))
+        .hasToString("{}");
+  }
+
+  @Test
+  public void testRejectTrailingStringContent() {
+    Assertions.assertThatThrownBy(() -> JsonUtil.parseJson("{} a"))
+        .isInstanceOf(JsonParseException.class)
+        .hasMessageStartingWith(
+            "Unrecognized token 'a': was expecting (JSON String, Number, Array, Object or token 'null', " +
+                "'true' or 'false')");
+
+    Assertions.assertThatThrownBy(() -> JsonUtil.parseJson("{} null"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageStartingWith("Found characters after the expected end of input");
+
+    Assertions.assertThatThrownBy(() -> JsonUtil.parseJson("{}{}"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageStartingWith("Found characters after the expected end of input");
+  }
+
+  @Test
+  public void testParseJsonStream() throws IOException {
+    // minimal
+    Assertions.assertThat(JsonUtil.parseJson(byteStream("{}")))
+        .hasToString("{}");
+
+    Assertions.assertThat(JsonUtil.parseJson(byteStream("{ \"attribute\": 123  }")))
+        .hasToString("{\"attribute\":123}");
+
+    // leading whitespace
+    Assertions.assertThat(JsonUtil.parseJson(byteStream("   {}")))
+        .hasToString("{}");
+
+    // trailing whitespace
+    Assertions.assertThat(JsonUtil.parseJson(byteStream("{}  ")))
+        .hasToString("{}");
+  }
+
+  @Test
+  public void testRejectTrailingStreamContent() {
+    Assertions.assertThatThrownBy(() -> JsonUtil.parseJson(byteStream("{} a")))
+        .isInstanceOf(JsonParseException.class)
+        .hasMessageStartingWith(
+            "Unrecognized token 'a': was expecting (JSON String, Number, Array, Object or token 'null', " +
+                "'true' or 'false')");
+
+    Assertions.assertThatThrownBy(() -> JsonUtil.parseJson(byteStream("{} null")))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageStartingWith("Found characters after the expected end of input");
+
+    Assertions.assertThatThrownBy(() -> JsonUtil.parseJson(byteStream("{}{}")))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageStartingWith("Found characters after the expected end of input");
+  }
+
+  private InputStream byteStream(String input) {
+    return new ByteArrayInputStream(input.getBytes(StandardCharsets.UTF_8));
+  }
+}

mr/build.gradle

@@ -33,6 +33,8 @@ project(':iceberg-mr') {
     implementation project(':iceberg-hive-metastore')
     implementation project(':iceberg-orc')
     implementation project(':iceberg-parquet')
+    implementation "com.fasterxml.jackson.core:jackson-databind"
+    implementation "com.fasterxml.jackson.core:jackson-core"
 
     compileOnly("org.apache.hadoop:hadoop-client") {
       exclude group: 'org.apache.avro', module: 'avro'