AVRO-4242: [Java] Fix NPE in DataFileStream and DataFileReader when schema metadata is missing

Malformed Avro container files without the 'avro.schema' metadata entry
caused a NullPointerException in both DataFileStream and DataFileReader12
when the null value was passed directly to Schema.Parser.parse(). Replace
inline parsing with null-safe helper methods that throw a descriptive
IOException instead.

Author: iemejia

lang/java/avro/src/main/java/org/apache/avro/file/DataFileReader12.java

@@ -17,21 +17,22 @@
  */
 package org.apache.avro.file;
 
-import java.io.IOException;
 import java.io.Closeable;
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
 
+import org.apache.avro.AvroRuntimeException;
 import org.apache.avro.InvalidAvroMagicException;
 import org.apache.avro.Schema;
 import org.apache.avro.UnknownAvroCodecException;
+import org.apache.avro.io.BinaryDecoder;
 import org.apache.avro.io.DatumReader;
 import org.apache.avro.io.DecoderFactory;
-import org.apache.avro.io.BinaryDecoder;
 
 /** Read files written by Avro version 1.2. */
 public class DataFileReader12<D> implements FileReader<D>, Closeable {
@@ -88,7 +89,7 @@ public DataFileReader12(SeekableInput sin, DatumReader<D> reader) throws IOExcep
     if (codec != null && !codec.equals(NULL_CODEC)) {
       throw new UnknownAvroCodecException("Unknown codec: " + codec);
     }
-    this.schema = new Schema.Parser().parse(getMetaString(SCHEMA));
+    this.schema = parseSchema();
     this.reader = reader;
 
     reader.setSchema(schema);
@@ -115,6 +116,18 @@ public synchronized long getMetaLong(String key) {
     return Long.parseLong(getMetaString(key));
   }
 
+  private Schema parseSchema() throws IOException {
+    String schemaJson = getMetaString(SCHEMA);
+    if (schemaJson == null) {
+      throw new IOException("Missing required metadata: " + SCHEMA);
+    }
+    try {
+      return new Schema.Parser().parse(schemaJson);
+    } catch (AvroRuntimeException e) {
+      throw new IOException("Invalid schema in metadata: " + SCHEMA, e);
+    }
+  }
+
   /** Return the schema used in this file. */
   @Override
   public Schema getSchema() {

lang/java/avro/src/main/java/org/apache/avro/file/DataFileStream.java

@@ -140,8 +140,7 @@ void initialize(InputStream in, byte[] magic) throws IOException {
 
     // finalize the header
     header.metaKeyList = Collections.unmodifiableList(header.metaKeyList);
-    header.schema = new Schema.Parser(NameValidator.NO_VALIDATION).setValidateDefaults(false)
-        .parse(getMetaString(DataFileConstants.SCHEMA));
+    header.schema = parseHeaderSchema();
     this.codec = resolveCodec();
     reader.setSchema(header.schema);
   }
@@ -198,6 +197,18 @@ public long getMetaLong(String key) {
     return Long.parseLong(getMetaString(key));
   }
 
+  private Schema parseHeaderSchema() throws IOException {
+    String schemaJson = getMetaString(DataFileConstants.SCHEMA);
+    if (schemaJson == null) {
+      throw new IOException("Missing required metadata: " + DataFileConstants.SCHEMA);
+    }
+    try {
+      return new Schema.Parser(NameValidator.NO_VALIDATION).setValidateDefaults(false).parse(schemaJson);
+    } catch (AvroRuntimeException e) {
+      throw new IOException("Invalid schema in metadata: " + DataFileConstants.SCHEMA, e);
+    }
+  }
+
   /**
    * Returns an iterator over entries in this file. Note that this iterator is
    * shared with other users of the file: it does not contain a separate pointer

lang/java/avro/src/test/java/org/apache/avro/TestDataFileReader.java

@@ -24,19 +24,25 @@
 import java.io.FileInputStream;
 import java.io.FileWriter;
 import java.io.IOException;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.lang.management.ManagementFactory;
 import java.lang.management.OperatingSystemMXBean;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import com.sun.management.UnixOperatingSystemMXBean;
 import org.apache.avro.file.DataFileReader;
+import org.apache.avro.file.DataFileConstants;
 import org.apache.avro.file.DataFileStream;
 import org.apache.avro.file.DataFileWriter;
 import org.apache.avro.file.FileReader;
+import org.apache.avro.file.SeekableByteArrayInput;
 import org.apache.avro.file.SeekableFileInput;
 import org.apache.avro.file.SeekableInput;
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.EncoderFactory;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
@@ -238,4 +244,36 @@ void invalidMagicBytes() throws IOException {
           () -> DataFileReader.openReader(fileInput, new GenericDatumReader<>()));
     }
   }
+
+  @Test
+  void missingSchemaMetadataDoesNotThrowNullPointerException() throws IOException {
+    byte[] malformedFile = buildContainerHeaderWithoutSchema();
+
+    IOException streamException = assertThrows(IOException.class,
+        () -> new DataFileStream<>(new ByteArrayInputStream(malformedFile), new GenericDatumReader<>()));
+    assertNotNull(streamException.getMessage());
+    assertTrue(streamException.getMessage().contains(DataFileConstants.SCHEMA));
+
+    IOException readerException = assertThrows(IOException.class,
+        () -> new DataFileReader<>(new SeekableByteArrayInput(malformedFile), new GenericDatumReader<>()));
+    assertNotNull(readerException.getMessage());
+    assertTrue(readerException.getMessage().contains(DataFileConstants.SCHEMA));
+  }
+
+  private static byte[] buildContainerHeaderWithoutSchema() throws IOException {
+    ByteArrayOutputStream output = new ByteArrayOutputStream();
+    output.write(DataFileConstants.MAGIC);
+
+    BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(output, null);
+    encoder.writeMapStart();
+    encoder.setItemCount(1);
+    encoder.startItem();
+    encoder.writeString(DataFileConstants.CODEC);
+    encoder.writeBytes("null".getBytes());
+    encoder.writeMapEnd();
+    encoder.writeFixed(new byte[DataFileConstants.SYNC_SIZE]);
+    encoder.flush();
+
+    return output.toByteArray();
+  }
 }