AVRO-4189: [java] Simplify the setting of the serializable classes

lang/java/avro/pom.xml

@@ -90,9 +90,6 @@
         <artifactId>maven-surefire-plugin</artifactId>
         <configuration>
           <parallel>none</parallel>
-          <systemProperties>
-            <org.apache.avro.SERIALIZABLE_CLASSES>java.math.BigDecimal,java.math.BigInteger,java.net.URI,java.net.URL,java.io.File,java.lang.Integer,org.apache.avro.reflect.TestReflect$R10</org.apache.avro.SERIALIZABLE_CLASSES>
-          </systemProperties>
         </configuration>
         <executions>
           <execution>

lang/java/avro/src/it/pom.xml

@@ -90,11 +90,26 @@
         <artifactId>maven-surefire-plugin</artifactId>
         <version>@maven-surefire-plugin.version@</version>
         <configuration>
-          <systemProperties>
-            <org.apache.avro.SERIALIZABLE_CLASSES>java.math.BigDecimal,java.math.BigInteger,java.net.URI,java.net.URL,java.io.File,java.lang.Integer,org.apache.avro.reflect.TestReflect$R10</org.apache.avro.SERIALIZABLE_CLASSES>
-          </systemProperties>
           <useModulePath>false</useModulePath>
           <failIfNoTests>true</failIfNoTests>
+          <systemPropertyVariables>
+
+            <!-- Repeating the related system properties here because this pom does not inherit the configuration. -->
+            <org.apache.avro.SERIALIZABLE_CLASSES>
+              java.net.URI,java.net.URL,
+              java.io.File,
+              java.util.HashMap,
+              java.util.List,
+              java.util.Collection,
+              java.util.Map,
+              java.util.Set,
+              java.util.concurrent.ConcurrentHashMap,
+              java.util.LinkedHashMap,
+              java.util.TreeMap
+            </org.apache.avro.SERIALIZABLE_CLASSES>
+            <org.apache.avro.SERIALIZABLE_PACKAGES>org.apache.avro</org.apache.avro.SERIALIZABLE_PACKAGES>
+
+          </systemPropertyVariables>
         </configuration>
       </plugin>
     </plugins>

lang/java/avro/src/main/java/org/apache/avro/specific/SpecificDatumReader.java

@@ -22,15 +22,13 @@
 import org.apache.avro.AvroRuntimeException;
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.io.ResolvingDecoder;
+import org.apache.avro.util.ClassSecurityValidator.SystemPropertiesPredicate;
 import org.apache.avro.util.ClassUtils;
 
 import java.io.IOException;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.stream.Stream;
+import org.apache.avro.util.ClassSecurityValidator;
 
 /**
  * {@link org.apache.avro.io.DatumReader DatumReader} for generated Java
@@ -39,47 +37,20 @@
 public class SpecificDatumReader<T> extends GenericDatumReader<T> {
 
   /**
-   * @deprecated prefer to use {@link #SERIALIZABLE_CLASSES} instead.
+   * @deprecated Use {@link SystemPropertiesPredicate} instead.
+   * @see ClassSecurityValidator
    */
   @Deprecated
-  public static final String[] SERIALIZABLE_PACKAGES;
-
-  public static final String[] SERIALIZABLE_CLASSES;
-
-  static {
-    // no serializable classes by default
-    SERIALIZABLE_CLASSES = streamPropertyEntries(System.getProperty("org.apache.avro.SERIALIZABLE_CLASSES"))
-        .toArray(String[]::new);
-
-    // no serializable packages by default
-    SERIALIZABLE_PACKAGES = streamPropertyEntries(System.getProperty("org.apache.avro.SERIALIZABLE_PACKAGES"))
-        // Add a '.' suffix to ensure we'll be matching package names instead of
-        // arbitrary prefixes, except for the wildcard "*", which allows all
-        // packages (this is only safe in fully controlled environments!).
-        .map(entry -> "*".equals(entry) ? entry : entry + ".").toArray(String[]::new);
-  }
+  public static final String[] SERIALIZABLE_PACKAGES = SystemPropertiesPredicate.SERIALIZABLE_PACKAGES
+      .toArray(new String[0]);
 
   /**
-   * Parse a comma separated list into non-empty entries. Leading and trailing
-   * whitespace is stripped.
-   *
-   * @param commaSeparatedEntries the comma separated list of entries
-   * @return a stream of the entries
+   * @deprecated Use {@link SystemPropertiesPredicate} instead.
+   * @see ClassSecurityValidator
    */
-  private static Stream<String> streamPropertyEntries(String commaSeparatedEntries) {
-    if (commaSeparatedEntries == null) {
-      return Stream.empty();
-    }
-    return Stream.of(commaSeparatedEntries.split(",")).map(String::strip).filter(s -> !s.isEmpty());
-  }
-
-  // The primitive "class names" based on Class.isPrimitive()
-  private static final Set<String> PRIMITIVES = new HashSet<>(Arrays.asList(Boolean.TYPE.getName(),
-      Character.TYPE.getName(), Byte.TYPE.getName(), Short.TYPE.getName(), Integer.TYPE.getName(), Long.TYPE.getName(),
-      Float.TYPE.getName(), Double.TYPE.getName(), Void.TYPE.getName()));
-
-  private final List<String> trustedPackages = new ArrayList<>();
-  private final List<String> trustedClasses = new ArrayList<>();
+  @Deprecated
+  public static final String[] SERIALIZABLE_CLASSES = SystemPropertiesPredicate.SERIALIZABLE_CLASSES
+      .toArray(new String[0]);
 
   public SpecificDatumReader() {
     this(null, null, SpecificData.get());
@@ -106,15 +77,11 @@ public SpecificDatumReader(Schema writer, Schema reader) {
    */
   public SpecificDatumReader(Schema writer, Schema reader, SpecificData data) {
     super(writer, reader, data);
-    trustedPackages.addAll(Arrays.asList(SERIALIZABLE_PACKAGES));
-    trustedClasses.addAll(Arrays.asList(SERIALIZABLE_CLASSES));
   }
 
   /** Construct given a {@link SpecificData}. */
   public SpecificDatumReader(SpecificData data) {
     super(data);
-    trustedPackages.addAll(Arrays.asList(SERIALIZABLE_PACKAGES));
-    trustedClasses.addAll(Arrays.asList(SERIALIZABLE_CLASSES));
   }
 
   /** Return the contained {@link SpecificData}. */
@@ -156,51 +123,29 @@ private Class getPropAsClass(Schema schema, String prop) {
     if (name == null)
       return null;
     try {
-      checkSecurity(name);
       Class clazz = ClassUtils.forName(getData().getClassLoader(), name);
       return clazz;
     } catch (ClassNotFoundException e) {
       throw new AvroRuntimeException(e);
     }
   }
 
-  private boolean trustAllPackages() {
-    return (trustedPackages.size() == 1 && "*".equals(trustedPackages.get(0)));
-  }
-
-  private void checkSecurity(String className) throws ClassNotFoundException {
-    if (trustAllPackages() || PRIMITIVES.contains(className)) {
-      return;
-    }
-
-    for (String trustedClass : getTrustedClasses()) {
-      if (className.equals(trustedClass)) {
-        return;
-      }
-    }
-
-    for (String trustedPackage : getTrustedPackages()) {
-      if (className.startsWith(trustedPackage)) {
-        return;
-      }
-    }
-
-    throw new SecurityException("Forbidden " + className + "! This class is not trusted to be included in Avro "
-        + "schemas using java-class. Please set the system property org.apache.avro.SERIALIZABLE_CLASSES to the comma "
-        + "separated list of classes you trust. You can also set the system property "
-        + "org.apache.avro.SERIALIZABLE_PACKAGES to the comma separated list of the packages you trust.");
-  }
-
   /**
-   * @deprecated Use getTrustedClasses() instead
+   * @deprecated Use {@link SystemPropertiesPredicate} instead.
+   * @see ClassSecurityValidator
    */
   @Deprecated
   public final List<String> getTrustedPackages() {
-    return trustedPackages;
+    return Arrays.asList(SERIALIZABLE_PACKAGES);
   }
 
+  /**
+   * @deprecated Use {@link SystemPropertiesPredicate} instead.
+   * @see ClassSecurityValidator
+   */
+  @Deprecated
   public final List<String> getTrustedClasses() {
-    return trustedClasses;
+    return Arrays.asList(SERIALIZABLE_CLASSES);
   }
 
   @Override