[GH-3914] Simplify and deprecate ContextUtil

parquet-avro/src/main/java/org/apache/parquet/avro/AvroParquetInputFormat.java

@@ -21,7 +21,6 @@
 import org.apache.avro.Schema;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.parquet.hadoop.ParquetInputFormat;
-import org.apache.parquet.hadoop.util.ContextUtil;
 
 /**
  * A Hadoop {@link org.apache.hadoop.mapreduce.InputFormat} for Parquet files.
@@ -53,7 +52,7 @@ public AvroParquetInputFormat() {
    * @see org.apache.parquet.avro.AvroParquetOutputFormat#setSchema(org.apache.hadoop.mapreduce.Job, org.apache.avro.Schema)
    */
   public static void setRequestedProjection(Job job, Schema requestedProjection) {
-    AvroReadSupport.setRequestedProjection(ContextUtil.getConfiguration(job), requestedProjection);
+    AvroReadSupport.setRequestedProjection(job.getConfiguration(), requestedProjection);
   }
 
   /**
@@ -69,7 +68,7 @@ public static void setRequestedProjection(Job job, Schema requestedProjection) {
    * @see org.apache.parquet.avro.AvroParquetOutputFormat#setSchema(org.apache.hadoop.mapreduce.Job, org.apache.avro.Schema)
    */
   public static void setAvroReadSchema(Job job, Schema avroReadSchema) {
-    AvroReadSupport.setAvroReadSchema(ContextUtil.getConfiguration(job), avroReadSchema);
+    AvroReadSupport.setAvroReadSchema(job.getConfiguration(), avroReadSchema);
   }
 
   /**
@@ -81,6 +80,6 @@ public static void setAvroReadSchema(Job job, Schema avroReadSchema) {
    * @param supplierClass an avro data supplier class
    */
   public static void setAvroDataSupplier(Job job, Class<? extends AvroDataSupplier> supplierClass) {
-    AvroReadSupport.setAvroDataSupplier(ContextUtil.getConfiguration(job), supplierClass);
+    AvroReadSupport.setAvroDataSupplier(job.getConfiguration(), supplierClass);
   }
 }

parquet-avro/src/main/java/org/apache/parquet/avro/AvroParquetOutputFormat.java

@@ -21,7 +21,6 @@
 import org.apache.avro.Schema;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.parquet.hadoop.ParquetOutputFormat;
-import org.apache.parquet.hadoop.util.ContextUtil;
 
 /**
  * A Hadoop {@link org.apache.hadoop.mapreduce.OutputFormat} for Parquet files.
@@ -39,7 +38,7 @@ public class AvroParquetOutputFormat<T> extends ParquetOutputFormat<T> {
    * @see org.apache.parquet.avro.AvroParquetInputFormat#setAvroReadSchema(org.apache.hadoop.mapreduce.Job, org.apache.avro.Schema)
    */
   public static void setSchema(Job job, Schema schema) {
-    AvroWriteSupport.setSchema(ContextUtil.getConfiguration(job), schema);
+    AvroWriteSupport.setSchema(job.getConfiguration(), schema);
   }
 
   public AvroParquetOutputFormat() {
@@ -55,6 +54,6 @@ public AvroParquetOutputFormat() {
    * @param supplierClass a supplier class
    */
   public static void setAvroDataSupplier(Job job, Class<? extends AvroDataSupplier> supplierClass) {
-    AvroWriteSupport.setAvroDataSupplier(ContextUtil.getConfiguration(job), supplierClass);
+    AvroWriteSupport.setAvroDataSupplier(job.getConfiguration(), supplierClass);
   }
 }

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java

@@ -64,7 +64,6 @@
 import org.apache.parquet.hadoop.metadata.GlobalMetaData;
 import org.apache.parquet.hadoop.metadata.ParquetMetadata;
 import org.apache.parquet.hadoop.util.ConfigurationUtil;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.apache.parquet.hadoop.util.HiddenFileFilter;
 import org.apache.parquet.hadoop.util.SerializationUtil;
 import org.apache.parquet.io.ParquetDecodingException;
@@ -173,19 +172,19 @@ public class ParquetInputFormat<T> extends FileInputFormat<Void, T> {
   public static final boolean HADOOP_VECTORED_IO_DEFAULT = true;
 
   public static void setTaskSideMetaData(Job job, boolean taskSideMetadata) {
-    ContextUtil.getConfiguration(job).setBoolean(TASK_SIDE_METADATA, taskSideMetadata);
+    job.getConfiguration().setBoolean(TASK_SIDE_METADATA, taskSideMetadata);
   }
 
   public static boolean isTaskSideMetaData(Configuration configuration) {
     return configuration.getBoolean(TASK_SIDE_METADATA, TRUE);
   }
 
   public static void setReadSupportClass(Job job, Class<?> readSupportClass) {
-    ContextUtil.getConfiguration(job).set(READ_SUPPORT_CLASS, readSupportClass.getName());
+    job.getConfiguration().set(READ_SUPPORT_CLASS, readSupportClass.getName());
   }
 
   public static void setUnboundRecordFilter(Job job, Class<? extends UnboundRecordFilter> filterClass) {
-    Configuration conf = ContextUtil.getConfiguration(job);
+    Configuration conf = job.getConfiguration();
     checkArgument(
         getFilterPredicate(conf) == null,
         "You cannot provide an UnboundRecordFilter after providing a FilterPredicate");
@@ -295,7 +294,7 @@ public <S extends ReadSupport<T>> ParquetInputFormat(Class<S> readSupportClass)
   @Override
   public RecordReader<Void, T> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext)
       throws IOException, InterruptedException {
-    Configuration conf = ContextUtil.getConfiguration(taskAttemptContext);
+    Configuration conf = taskAttemptContext.getConfiguration();
     ReadSupport<T> readSupport = getReadSupport(conf);
     return new ParquetRecordReader<T>(readSupport, getFilter(conf));
   }
@@ -340,15 +339,15 @@ static <T> ReadSupport<T> getReadSupportInstance(Class<? extends ReadSupport<T>>
 
   @Override
   protected boolean isSplitable(JobContext context, Path filename) {
-    return ContextUtil.getConfiguration(context).getBoolean(SPLIT_FILES, true);
+    return context.getConfiguration().getBoolean(SPLIT_FILES, true);
   }
 
   /**
    * {@inheritDoc}
    */
   @Override
   public List<InputSplit> getSplits(JobContext jobContext) throws IOException {
-    Configuration configuration = ContextUtil.getConfiguration(jobContext);
+    Configuration configuration = jobContext.getConfiguration();
     List<InputSplit> splits = new ArrayList<InputSplit>();
 
     if (isTaskSideMetaData(configuration)) {
@@ -396,7 +395,7 @@ public List<ParquetInputSplit> getSplits(Configuration configuration, List<Foote
    */
   @Override
   protected List<FileStatus> listStatus(JobContext jobContext) throws IOException {
-    return getAllFileRecursively(super.listStatus(jobContext), ContextUtil.getConfiguration(jobContext));
+    return getAllFileRecursively(super.listStatus(jobContext), jobContext.getConfiguration());
   }
 
   private static List<FileStatus> getAllFileRecursively(List<FileStatus> files, Configuration conf)
@@ -436,7 +435,7 @@ public List<Footer> getFooters(JobContext jobContext) throws IOException {
     if (statuses.isEmpty()) {
       return Collections.emptyList();
     }
-    Configuration config = ContextUtil.getConfiguration(jobContext);
+    Configuration config = jobContext.getConfiguration();
     // Use LinkedHashMap to preserve the insertion order and ultimately to return the list of
     // footers in the same order as the list of file statuses returned from listStatus()
     Map<FileStatusWrapper, Footer> footersMap = new LinkedHashMap<FileStatusWrapper, Footer>();

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputCommitter.java

@@ -28,7 +28,6 @@
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
 import org.apache.parquet.hadoop.ParquetOutputFormat.JobSummaryLevel;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,7 +43,7 @@ public ParquetOutputCommitter(Path outputPath, TaskAttemptContext context) throw
 
   public void commitJob(JobContext jobContext) throws IOException {
     super.commitJob(jobContext);
-    Configuration configuration = ContextUtil.getConfiguration(jobContext);
+    Configuration configuration = jobContext.getConfiguration();
     writeMetaDataFile(configuration, outputPath);
   }
 

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java

@@ -21,7 +21,6 @@
 import static org.apache.parquet.column.ParquetProperties.DEFAULT_ADAPTIVE_BLOOM_FILTER_ENABLED;
 import static org.apache.parquet.column.ParquetProperties.DEFAULT_BLOOM_FILTER_ENABLED;
 import static org.apache.parquet.hadoop.ParquetWriter.DEFAULT_BLOCK_SIZE;
-import static org.apache.parquet.hadoop.util.ContextUtil.getConfiguration;
 
 import java.io.IOException;
 import java.util.Objects;
@@ -190,7 +189,7 @@ public static JobSummaryLevel getJobSummaryLevel(Configuration conf) {
   }
 
   public static void setWriteSupportClass(Job job, Class<?> writeSupportClass) {
-    getConfiguration(job).set(WRITE_SUPPORT_CLASS, writeSupportClass.getName());
+    job.getConfiguration().set(WRITE_SUPPORT_CLASS, writeSupportClass.getName());
   }
 
   public static void setWriteSupportClass(JobConf job, Class<?> writeSupportClass) {
@@ -208,35 +207,35 @@ public static Class<?> getWriteSupportClass(Configuration configuration) {
   }
 
   public static void setBlockSize(Job job, int blockSize) {
-    getConfiguration(job).setInt(BLOCK_SIZE, blockSize);
+    job.getConfiguration().setInt(BLOCK_SIZE, blockSize);
   }
 
   public static void setPageSize(Job job, int pageSize) {
-    getConfiguration(job).setInt(PAGE_SIZE, pageSize);
+    job.getConfiguration().setInt(PAGE_SIZE, pageSize);
   }
 
   public static void setDictionaryPageSize(Job job, int pageSize) {
-    getConfiguration(job).setInt(DICTIONARY_PAGE_SIZE, pageSize);
+    job.getConfiguration().setInt(DICTIONARY_PAGE_SIZE, pageSize);
   }
 
   public static void setCompression(Job job, CompressionCodecName compression) {
-    getConfiguration(job).set(COMPRESSION, compression.name());
+    job.getConfiguration().set(COMPRESSION, compression.name());
   }
 
   public static void setColumnCompression(Job job, String columnPath, CompressionCodecName codec) {
-    getConfiguration(job).set(COMPRESSION + '#' + columnPath, codec.name());
+    job.getConfiguration().set(COMPRESSION + '#' + columnPath, codec.name());
   }
 
   public static void setColumnCompressionLevel(Job job, String columnPath, int level) {
-    getConfiguration(job).setInt(COLUMN_COMPRESSION_LEVEL_PREFIX + '#' + columnPath, level);
+    job.getConfiguration().setInt(COLUMN_COMPRESSION_LEVEL_PREFIX + '#' + columnPath, level);
   }
 
   public static void setEnableDictionary(Job job, boolean enableDictionary) {
-    getConfiguration(job).setBoolean(ENABLE_DICTIONARY, enableDictionary);
+    job.getConfiguration().setBoolean(ENABLE_DICTIONARY, enableDictionary);
   }
 
   public static boolean getEnableDictionary(JobContext jobContext) {
-    return getEnableDictionary(getConfiguration(jobContext));
+    return getEnableDictionary(jobContext.getConfiguration());
   }
 
   public static int getBloomFilterMaxBytes(Configuration conf) {
@@ -252,31 +251,31 @@ public static boolean getAdaptiveBloomFilterEnabled(Configuration conf) {
   }
 
   public static int getBlockSize(JobContext jobContext) {
-    return getBlockSize(getConfiguration(jobContext));
+    return getBlockSize(jobContext.getConfiguration());
   }
 
   public static int getPageSize(JobContext jobContext) {
-    return getPageSize(getConfiguration(jobContext));
+    return getPageSize(jobContext.getConfiguration());
   }
 
   public static int getDictionaryPageSize(JobContext jobContext) {
-    return getDictionaryPageSize(getConfiguration(jobContext));
+    return getDictionaryPageSize(jobContext.getConfiguration());
   }
 
   public static CompressionCodecName getCompression(JobContext jobContext) {
-    return getCompression(getConfiguration(jobContext));
+    return getCompression(jobContext.getConfiguration());
   }
 
   public static boolean isCompressionSet(JobContext jobContext) {
-    return isCompressionSet(getConfiguration(jobContext));
+    return isCompressionSet(jobContext.getConfiguration());
   }
 
   public static void setValidation(JobContext jobContext, boolean validating) {
-    setValidation(getConfiguration(jobContext), validating);
+    setValidation(jobContext.getConfiguration(), validating);
   }
 
   public static boolean getValidation(JobContext jobContext) {
-    return getValidation(getConfiguration(jobContext));
+    return getValidation(jobContext.getConfiguration());
   }
 
   public static boolean getEnableDictionary(Configuration configuration) {
@@ -350,7 +349,7 @@ private CompressionCodecName getCodec(TaskAttemptContext taskAttemptContext) {
   }
 
   public static void setMaxPaddingSize(JobContext jobContext, int maxPaddingSize) {
-    setMaxPaddingSize(getConfiguration(jobContext), maxPaddingSize);
+    setMaxPaddingSize(jobContext.getConfiguration(), maxPaddingSize);
   }
 
   public static void setMaxPaddingSize(Configuration conf, int maxPaddingSize) {
@@ -362,7 +361,7 @@ private static int getMaxPaddingSize(Configuration conf) {
   }
 
   public static void setColumnIndexTruncateLength(JobContext jobContext, int length) {
-    setColumnIndexTruncateLength(getConfiguration(jobContext), length);
+    setColumnIndexTruncateLength(jobContext.getConfiguration(), length);
   }
 
   public static void setColumnIndexTruncateLength(Configuration conf, int length) {
@@ -374,7 +373,7 @@ private static int getColumnIndexTruncateLength(Configuration conf) {
   }
 
   public static void setStatisticsTruncateLength(JobContext jobContext, int length) {
-    setStatisticsTruncateLength(getConfiguration(jobContext), length);
+    setStatisticsTruncateLength(jobContext.getConfiguration(), length);
   }
 
   private static void setStatisticsTruncateLength(Configuration conf, int length) {
@@ -386,7 +385,7 @@ private static int getStatisticsTruncateLength(Configuration conf) {
   }
 
   public static void setBlockRowCountLimit(JobContext jobContext, int rowCount) {
-    setBlockRowCountLimit(getConfiguration(jobContext), rowCount);
+    setBlockRowCountLimit(jobContext.getConfiguration(), rowCount);
   }
 
   public static void setBlockRowCountLimit(Configuration conf, int rowCount) {
@@ -398,7 +397,7 @@ static int getBlockRowCountLimit(Configuration conf) {
   }
 
   public static void setPageRowCountLimit(JobContext jobContext, int rowCount) {
-    setPageRowCountLimit(getConfiguration(jobContext), rowCount);
+    setPageRowCountLimit(jobContext.getConfiguration(), rowCount);
   }
 
   public static void setPageRowCountLimit(Configuration conf, int rowCount) {
@@ -410,7 +409,7 @@ private static int getPageRowCountLimit(Configuration conf) {
   }
 
   public static void setPageWriteChecksumEnabled(JobContext jobContext, boolean val) {
-    setPageWriteChecksumEnabled(getConfiguration(jobContext), val);
+    setPageWriteChecksumEnabled(jobContext.getConfiguration(), val);
   }
 
   public static void setPageWriteChecksumEnabled(Configuration conf, boolean val) {
@@ -422,15 +421,15 @@ public static boolean getPageWriteChecksumEnabled(Configuration conf) {
   }
 
   public static void setStatisticsEnabled(JobContext jobContext, boolean enabled) {
-    getConfiguration(jobContext).setBoolean(STATISTICS_ENABLED, enabled);
+    jobContext.getConfiguration().setBoolean(STATISTICS_ENABLED, enabled);
   }
 
   public static boolean getStatisticsEnabled(Configuration conf) {
     return conf.getBoolean(STATISTICS_ENABLED, ParquetProperties.DEFAULT_STATISTICS_ENABLED);
   }
 
   public static void setStatisticsEnabled(JobContext jobContext, String columnPath, boolean enabled) {
-    getConfiguration(jobContext).set(STATISTICS_ENABLED + "#" + columnPath, String.valueOf(enabled));
+    jobContext.getConfiguration().set(STATISTICS_ENABLED + "#" + columnPath, String.valueOf(enabled));
   }
 
   public static boolean getStatisticsEnabled(Configuration conf, String columnPath) {
@@ -490,7 +489,7 @@ public RecordWriter<Void, T> getRecordWriter(TaskAttemptContext taskAttemptConte
   public RecordWriter<Void, T> getRecordWriter(TaskAttemptContext taskAttemptContext, Mode mode)
       throws IOException, InterruptedException {
 
-    final Configuration conf = getConfiguration(taskAttemptContext);
+    final Configuration conf = taskAttemptContext.getConfiguration();
 
     CompressionCodecName codec = getCodec(taskAttemptContext);
     String extension = codec.getExtension() + ".parquet";
@@ -505,7 +504,7 @@ public RecordWriter<Void, T> getRecordWriter(TaskAttemptContext taskAttemptConte
 
   public RecordWriter<Void, T> getRecordWriter(TaskAttemptContext taskAttemptContext, Path file, Mode mode)
       throws IOException, InterruptedException {
-    return getRecordWriter(getConfiguration(taskAttemptContext), file, getCodec(taskAttemptContext), mode);
+    return getRecordWriter(taskAttemptContext.getConfiguration(), file, getCodec(taskAttemptContext), mode);
   }
 
   public RecordWriter<Void, T> getRecordWriter(Configuration conf, Path file, CompressionCodecName codec)

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordReader.java

@@ -43,12 +43,9 @@
 import org.apache.parquet.hadoop.metadata.BlockMetaData;
 import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
 import org.apache.parquet.hadoop.metadata.FileMetaData;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.apache.parquet.hadoop.util.HadoopInputFile;
 import org.apache.parquet.hadoop.util.counters.BenchmarkCounter;
 import org.apache.parquet.io.ParquetDecodingException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * Reads the records from a block of a Parquet file
@@ -59,7 +56,6 @@
  */
 public class ParquetRecordReader<T> extends RecordReader<Void, T> {
 
-  private static final Logger LOG = LoggerFactory.getLogger(ParquetRecordReader.class);
   private final InternalParquetRecordReader<T> internalReader;
 
   /**
@@ -125,15 +121,9 @@ public float getProgress() throws IOException, InterruptedException {
   @Override
   public void initialize(InputSplit inputSplit, TaskAttemptContext context) throws IOException, InterruptedException {
 
-    if (ContextUtil.hasCounterMethod(context)) {
-      BenchmarkCounter.initCounterFromContext(context);
-    } else {
-      LOG.error(String.format(
-          "Can not initialize counter because the class '%s' does not have a '.getCounterMethod'",
-          context.getClass().getCanonicalName()));
-    }
+    BenchmarkCounter.initCounterFromContext(context);
 
-    initializeInternalReader(toParquetSplit(inputSplit), ContextUtil.getConfiguration(context));
+    initializeInternalReader(toParquetSplit(inputSplit), context.getConfiguration());
   }
 
   public void initialize(InputSplit inputSplit, Configuration configuration, Reporter reporter)

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/CodecConfig.java

@@ -26,7 +26,6 @@
 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 import org.apache.parquet.hadoop.ParquetOutputFormat;
 import org.apache.parquet.hadoop.metadata.CompressionCodecName;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -141,7 +140,7 @@ public Class getHadoopOutputCompressorClass(Class defaultCodec) {
 
     @Override
     public Configuration getConfiguration() {
-      return ContextUtil.getConfiguration(context);
+      return context.getConfiguration();
     }
   }