PARQUET-212: Add Configuration to the ThriftRecordConverter ctor.

Previously, this added Configurable to ThriftRecordConverter, but that
caused problems with semver and test failures because a separate
initialize method had to be called. This is brittle because callers
might not know to call initialize and the class is part of the API
because subclassing is allowed. To avoid the issue, this replaces the
Configurable interface and initialize method with a new constructor that
takes a Configuration.

Author: rdblue

parquet-scrooge/src/main/java/org/apache/parquet/scrooge/ScroogeRecordConverter.java

@@ -18,6 +18,7 @@
  */
 package org.apache.parquet.scrooge;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.thrift.TException;
 import org.apache.thrift.protocol.TProtocol;
 
@@ -31,16 +32,24 @@
 
 public class ScroogeRecordConverter<T extends ThriftStruct> extends ThriftRecordConverter<T> {
 
-
+  /**
+   * This is for compatibility only.
+   * @deprecated will be removed in 2.x
+   */
+  @Deprecated
   public ScroogeRecordConverter(final Class<T> thriftClass, MessageType parquetSchema, StructType thriftType) {
+    this(thriftClass, parquetSchema, thriftType, null);
+  }
+
+  public ScroogeRecordConverter(final Class<T> thriftClass, MessageType parquetSchema, StructType thriftType, Configuration conf) {
     super(new ThriftReader<T>() {
       @SuppressWarnings("unchecked")
       ThriftStructCodec<T> codec = (ThriftStructCodec<T>) getCodec(thriftClass);
       @Override
       public T readOneRecord(TProtocol protocol) throws TException {
           return codec.decode(protocol);
       }
-    }, thriftClass.getSimpleName(), parquetSchema, thriftType);
+    }, thriftClass.getSimpleName(), parquetSchema, thriftType, conf);
   }
 
   private static ThriftStructCodec<?> getCodec(Class<?> klass) {

parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftReadSupport.java

@@ -19,6 +19,7 @@
 package org.apache.parquet.hadoop.thrift;
 
 import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
 import java.util.Map;
 import java.util.Set;
 
@@ -225,23 +226,55 @@ public RecordMaterializer<T> prepareForRead(Configuration configuration,
     ThriftMetaData thriftMetaData = ThriftMetaData.fromExtraMetaData(keyValueMetaData);
     try {
       initThriftClass(thriftMetaData, configuration);
+    } catch (ClassNotFoundException e) {
+      throw new RuntimeException("Cannot find Thrift object class for metadata: " + thriftMetaData, e);
+    }
+
+    // if there was not metadata in the file, get it from requested class
+    if (thriftMetaData == null) {
+      thriftMetaData = ThriftMetaData.fromThriftClass(thriftClass);
+    }
 
-      // if there was not metadata in the file, get it from requested class
-      if (thriftMetaData == null) {
-        thriftMetaData = ThriftMetaData.fromThriftClass(thriftClass);
+    String converterClassName = configuration.get(RECORD_CONVERTER_CLASS_KEY, RECORD_CONVERTER_DEFAULT);
+    return getRecordConverterInstance(converterClassName, thriftClass,
+        readContext.getRequestedSchema(), thriftMetaData.getDescriptor(),
+        configuration);
+  }
+
+  @SuppressWarnings("unchecked")
+  private static <T> ThriftRecordConverter<T> getRecordConverterInstance(
+      String converterClassName, Class<T> thriftClass,
+      MessageType requestedSchema, StructType descriptor, Configuration conf) {
+    Class<ThriftRecordConverter<T>> converterClass;
+    try {
+      converterClass = (Class<ThriftRecordConverter<T>>) Class.forName(converterClassName);
+    } catch (ClassNotFoundException e) {
+      throw new RuntimeException("Cannot find Thrift converter class: " + converterClassName, e);
+    }
+
+    try {
+      // first try the new version that accepts a Configuration
+      try {
+        Constructor<ThriftRecordConverter<T>> constructor =
+            converterClass.getConstructor(Class.class, MessageType.class, StructType.class, Configuration.class);
+        return constructor.newInstance(thriftClass, requestedSchema, descriptor, conf);
+      } catch (IllegalAccessException e) {
+        // try the other constructor pattern
+      } catch (NoSuchMethodException e) {
+        // try to find the other constructor pattern
       }
 
-      String converterClassName = configuration.get(RECORD_CONVERTER_CLASS_KEY, RECORD_CONVERTER_DEFAULT);
-      @SuppressWarnings("unchecked")
-      Class<ThriftRecordConverter<T>> converterClass = (Class<ThriftRecordConverter<T>>) Class.forName(converterClassName);
       Constructor<ThriftRecordConverter<T>> constructor =
           converterClass.getConstructor(Class.class, MessageType.class, StructType.class);
-      ThriftRecordConverter<T> converter = constructor.newInstance(thriftClass, readContext.getRequestedSchema(), thriftMetaData.getDescriptor());
-      converter.setConf(configuration);
-      converter.initialize();
-      return converter;
-    } catch (Exception t) {
-      throw new RuntimeException("Unable to create Thrift Converter for Thrift metadata " + thriftMetaData, t);
+      return constructor.newInstance(thriftClass, requestedSchema, descriptor);
+    } catch (InstantiationException e) {
+      throw new RuntimeException("Failed to construct Thrift converter class: " + converterClassName, e);
+    } catch (InvocationTargetException e) {
+      throw new RuntimeException("Failed to construct Thrift converter class: " + converterClassName, e);
+    } catch (IllegalAccessException e) {
+      throw new RuntimeException("Cannot access constructor for Thrift converter class: " + converterClassName, e);
+    } catch (NoSuchMethodException e) {
+      throw new RuntimeException("Cannot find constructor for Thrift converter class: " + converterClassName, e);
     }
   }
 }

parquet-thrift/src/main/java/org/apache/parquet/thrift/TBaseRecordConverter.java

@@ -18,6 +18,7 @@
  */
 package org.apache.parquet.thrift;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.thrift.TBase;
 import org.apache.thrift.TException;
 import org.apache.thrift.protocol.TProtocol;
@@ -28,7 +29,16 @@
 
 public class TBaseRecordConverter<T extends TBase<?,?>> extends ThriftRecordConverter<T> {
 
+  /**
+   * This is for compatibility only.
+   * @deprecated will be removed in 2.x
+   */
+  @Deprecated
   public TBaseRecordConverter(final Class<T> thriftClass, MessageType requestedParquetSchema, StructType thriftType) {
+    this(thriftClass, requestedParquetSchema, thriftType, null);
+  }
+
+  public TBaseRecordConverter(final Class<T> thriftClass, MessageType requestedParquetSchema, StructType thriftType, Configuration conf) {
     super(new ThriftReader<T>() {
       @Override
       public T readOneRecord(TProtocol protocol) throws TException {
@@ -42,7 +52,7 @@ public T readOneRecord(TProtocol protocol) throws TException {
             throw new ParquetDecodingException("Thrift class or constructor not public " + thriftClass, e);
           }
       }
-    }, thriftClass.getSimpleName(), requestedParquetSchema, thriftType);
+    }, thriftClass.getSimpleName(), requestedParquetSchema, thriftType, conf);
   }
 
 }

parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java

@@ -65,7 +65,7 @@
  *
  * @param <T>
  */
-public class ThriftRecordConverter<T> extends RecordMaterializer<T> implements Configurable {
+public class ThriftRecordConverter<T> extends RecordMaterializer<T> {
 
   private static final Log LOG = Log.getLog(ThriftRecordConverter.class);
 
@@ -861,49 +861,41 @@ public void end() {
   }
   private final ThriftReader<T> thriftReader;
   private final ParquetReadProtocol protocol;
-  private final MessageType requestedParquetSchema;
-  private final String name;
   private GroupConverter structConverter;
   private List<TProtocol> rootEvents = new ArrayList<TProtocol>();
   private boolean missingRequiredFieldsInProjection = false;
-  private Configuration conf = null;
   private boolean ignoreNullElements = IGNORE_NULL_LIST_ELEMENTS_DEFAULT;
 
+  /**
+   * This is for compatibility only.
+   * @deprecated will be removed in 2.x
+   */
+  @Deprecated
+  public ThriftRecordConverter(ThriftReader<T> thriftReader, String name, MessageType requestedParquetSchema, ThriftType.StructType thriftType) {
+    this(thriftReader, name, requestedParquetSchema, thriftType, null);
+  }
+
   /**
    *
    * @param thriftReader the class responsible for instantiating the final object and read from the protocol
    * @param name the name of that type ( the thrift class simple name)
    * @param requestedParquetSchema the schema for the incoming columnar events
    * @param thriftType the thrift type descriptor
+   * @param conf a Configuration
    */
-  public ThriftRecordConverter(ThriftReader<T> thriftReader, String name, MessageType requestedParquetSchema, ThriftType.StructType thriftType) {
+  public ThriftRecordConverter(ThriftReader<T> thriftReader, String name, MessageType requestedParquetSchema, ThriftType.StructType thriftType, Configuration conf) {
     super();
     this.thriftReader = thriftReader;
-    this.name = name;
-    this.requestedParquetSchema = requestedParquetSchema;
     this.protocol = new ParquetReadProtocol();
     this.thriftType = thriftType;
-  }
-
-  public void initialize() {
-    MessageType fullSchema = ThriftSchemaConverter.convertWithoutProjection(thriftType);
-    missingRequiredFieldsInProjection = hasMissingRequiredFieldInGroupType(requestedParquetSchema, fullSchema);
-    this.structConverter = new StructConverter(rootEvents, requestedParquetSchema, new ThriftField(name, (short)0, Requirement.REQUIRED, thriftType));
-  }
-
-  @Override
-  public void setConf(Configuration configuration) {
-    this.conf = configuration;
     if (conf != null) {
       this.ignoreNullElements = conf.getBoolean(
           IGNORE_NULL_LIST_ELEMENTS,
           IGNORE_NULL_LIST_ELEMENTS_DEFAULT);
     }
-  }
-
-  @Override
-  public Configuration getConf() {
-    return conf;
+    MessageType fullSchema = ThriftSchemaConverter.convertWithoutProjection(thriftType);
+    missingRequiredFieldsInProjection = hasMissingRequiredFieldInGroupType(requestedParquetSchema, fullSchema);
+    this.structConverter = new StructConverter(rootEvents, requestedParquetSchema, new ThriftField(name, (short)0, Requirement.REQUIRED, thriftType));
   }
 
   private boolean hasMissingRequiredFieldInGroupType(GroupType requested, GroupType fullSchema) {

parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestArrayCompatibility.java

@@ -331,7 +331,7 @@ public void write(RecordConsumer rc) {
       fail("Should fail: locations are optional and not ignored");
     } catch (RuntimeException e) {
       // e is a RuntimeException wrapping the decoding exception
-      assertTrue(e.getCause().getMessage().contains("locations"));
+      assertTrue(e.getCause().getCause().getMessage().contains("locations"));
     }
 
     assertReaderContains(readerIgnoreNulls(test, ListOfLocations.class), expected);
@@ -615,7 +615,7 @@ public void write(RecordConsumer rc) {
       fail("Should fail: locations are optional and not ignored");
     } catch (RuntimeException e) {
       // e is a RuntimeException wrapping the decoding exception
-      assertTrue(e.getCause().getMessage().contains("locations"));
+      assertTrue(e.getCause().getCause().getMessage().contains("locations"));
     }
 
     assertReaderContains(readerIgnoreNulls(test, ListOfLocations.class), expected);