diff --git a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroParquetInputFormat.java b/parquet-avro/src/main/java/org/apache/parquet/avro/AvroParquetInputFormat.java
index ce625f20ed..28c9480595 100644
--- a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroParquetInputFormat.java
+++ b/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);
   }
 }
diff --git a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroParquetOutputFormat.java b/parquet-avro/src/main/java/org/apache/parquet/avro/AvroParquetOutputFormat.java
index de80be5aa4..10453aeb41 100644
--- a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroParquetOutputFormat.java
+++ b/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);
   }
 }
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java
index 288e1d241e..784a7d1695 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java
+++ b/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,7 +172,7 @@ 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) {
@@ -181,11 +180,11 @@ public static boolean isTaskSideMetaData(Configuration configuration) {
   }
 
   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,7 +339,7 @@ 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);
   }
 
   /**
@@ -348,7 +347,7 @@ protected boolean isSplitable(JobContext context, Path filename) {
    */
   @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>();
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputCommitter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputCommitter.java
index b41c039e15..3d9fca90c0 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputCommitter.java
+++ b/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);
   }
 
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java
index 0e8823001d..6cc19acca7 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java
+++ b/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,7 +421,7 @@ 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) {
@@ -430,7 +429,7 @@ public static boolean getStatisticsEnabled(Configuration conf) {
   }
 
   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)
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordReader.java
index c0e52fc5c6..c93f406a8e 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordReader.java
+++ b/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)
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/CodecConfig.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/CodecConfig.java
index 3ec31a712a..b3712379df 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/CodecConfig.java
+++ b/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();
     }
   }
 
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/example/ExampleOutputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/example/ExampleOutputFormat.java
index ddb2d8023f..97144a860f 100755
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/example/ExampleOutputFormat.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/example/ExampleOutputFormat.java
@@ -21,7 +21,6 @@
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.parquet.example.data.Group;
 import org.apache.parquet.hadoop.ParquetOutputFormat;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.apache.parquet.schema.MessageType;
 
 /**
@@ -41,7 +40,7 @@ public class ExampleOutputFormat extends ParquetOutputFormat<Group> {
    * @param schema the schema of the data
    */
   public static void setSchema(Job job, MessageType schema) {
-    GroupWriteSupport.setSchema(schema, ContextUtil.getConfiguration(job));
+    GroupWriteSupport.setSchema(schema, job.getConfiguration());
   }
 
   /**
@@ -51,7 +50,7 @@ public static void setSchema(Job job, MessageType schema) {
    * @return the schema
    */
   public static MessageType getSchema(Job job) {
-    return GroupWriteSupport.getSchema(ContextUtil.getConfiguration(job));
+    return GroupWriteSupport.getSchema(job.getConfiguration());
   }
 
   public ExampleOutputFormat() {
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/mapred/MapredParquetOutputCommitter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/mapred/MapredParquetOutputCommitter.java
index 509c59dc3a..ce7e844161 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/mapred/MapredParquetOutputCommitter.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/mapred/MapredParquetOutputCommitter.java
@@ -27,7 +27,6 @@
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.JobContext;
 import org.apache.parquet.hadoop.ParquetOutputCommitter;
-import org.apache.parquet.hadoop.util.ContextUtil;
 
 /**
  * Adapter for supporting ParquetOutputCommitter in mapred API
@@ -37,7 +36,7 @@ public class MapredParquetOutputCommitter extends FileOutputCommitter {
   @Override
   public void commitJob(JobContext jobContext) throws IOException {
     super.commitJob(jobContext);
-    Configuration conf = ContextUtil.getConfiguration(jobContext);
+    Configuration conf = jobContext.getConfiguration();
     Path outputPath = FileOutputFormat.getOutputPath(new JobConf(conf));
     ParquetOutputCommitter.writeMetaDataFile(conf, outputPath);
   }
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/ContextUtil.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/ContextUtil.java
index 9c80571043..8c64114d5d 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/ContextUtil.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/ContextUtil.java
@@ -18,270 +18,85 @@
  */
 package org.apache.parquet.hadoop.util;
 
-import java.lang.reflect.Constructor;
-import java.lang.reflect.Field;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.util.HashMap;
-import java.util.Map;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapreduce.Counter;
-import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.JobContext;
 import org.apache.hadoop.mapreduce.JobID;
-import org.apache.hadoop.mapreduce.MapContext;
-import org.apache.hadoop.mapreduce.OutputCommitter;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.RecordWriter;
-import org.apache.hadoop.mapreduce.StatusReporter;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.apache.hadoop.mapreduce.TaskAttemptID;
-
-/*
- * This is based on ContextFactory.java from hadoop-2.0.x sources.
- */
+import org.apache.hadoop.mapreduce.counters.GenericCounter;
+import org.apache.hadoop.mapreduce.task.JobContextImpl;
+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
 
 /**
- * Utility methods to allow applications to deal with inconsistencies between
- * MapReduce Context Objects API between hadoop-0.20 and later versions.
+ * Utility methods for creating and accessing Hadoop MapReduce contexts.
+ * Runtime exceptions from Hadoop constructors and methods propagate directly to callers.
+ *
+ * @deprecated Use {@link JobContextImpl}, {@link TaskAttemptContextImpl}, and {@link GenericCounter}
+ * constructors and Hadoop context and counter methods directly. All supported contexts expose counter
+ * methods; check for a non-null context instead of using {@link #hasCounterMethod(TaskAttemptContext)}.
  */
+@Deprecated
 public class ContextUtil {
 
-  private static final boolean useV21;
-
-  private static final Constructor<?> JOB_CONTEXT_CONSTRUCTOR;
-  private static final Constructor<?> TASK_CONTEXT_CONSTRUCTOR;
-  private static final Constructor<?> MAP_CONTEXT_CONSTRUCTOR;
-  private static final Constructor<?> MAP_CONTEXT_IMPL_CONSTRUCTOR;
-  private static final Constructor<?> GENERIC_COUNTER_CONSTRUCTOR;
-
-  private static final Field READER_FIELD;
-  private static final Field WRITER_FIELD;
-  private static final Field OUTER_MAP_FIELD;
-  private static final Field WRAPPED_CONTEXT_FIELD;
-
-  private static final Method GET_CONFIGURATION_METHOD;
-  private static final Method INCREMENT_COUNTER_METHOD;
-
-  private static final Map<Class, Method> COUNTER_METHODS_BY_CLASS = new HashMap<Class, Method>();
-
-  static {
-    boolean v21 = true;
-    final String PACKAGE = "org.apache.hadoop.mapreduce";
-    try {
-      Class.forName(PACKAGE + ".task.JobContextImpl");
-    } catch (ClassNotFoundException cnfe) {
-      v21 = false;
-    }
-    useV21 = v21;
-    Class<?> jobContextCls;
-    Class<?> taskContextCls;
-    Class<?> taskIOContextCls;
-    Class<?> mapCls;
-    Class<?> mapContextCls;
-    Class<?> innerMapContextCls;
-    Class<?> genericCounterCls;
-    try {
-      if (v21) {
-        jobContextCls = Class.forName(PACKAGE + ".task.JobContextImpl");
-        taskContextCls = Class.forName(PACKAGE + ".task.TaskAttemptContextImpl");
-        taskIOContextCls = Class.forName(PACKAGE + ".task.TaskInputOutputContextImpl");
-        mapContextCls = Class.forName(PACKAGE + ".task.MapContextImpl");
-        mapCls = Class.forName(PACKAGE + ".lib.map.WrappedMapper");
-        innerMapContextCls = Class.forName(PACKAGE + ".lib.map.WrappedMapper$Context");
-        genericCounterCls = Class.forName(PACKAGE + ".counters.GenericCounter");
-      } else {
-        jobContextCls = Class.forName(PACKAGE + ".JobContext");
-        taskContextCls = Class.forName(PACKAGE + ".TaskAttemptContext");
-        taskIOContextCls = Class.forName(PACKAGE + ".TaskInputOutputContext");
-        mapContextCls = Class.forName(PACKAGE + ".MapContext");
-        mapCls = Class.forName(PACKAGE + ".Mapper");
-        innerMapContextCls = Class.forName(PACKAGE + ".Mapper$Context");
-        genericCounterCls = Class.forName("org.apache.hadoop.mapred.Counters$Counter");
-      }
-    } catch (ClassNotFoundException e) {
-      throw new IllegalArgumentException("Can't find class", e);
-    }
-    try {
-      JOB_CONTEXT_CONSTRUCTOR = jobContextCls.getConstructor(Configuration.class, JobID.class);
-      JOB_CONTEXT_CONSTRUCTOR.setAccessible(true);
-      TASK_CONTEXT_CONSTRUCTOR = taskContextCls.getConstructor(Configuration.class, TaskAttemptID.class);
-      TASK_CONTEXT_CONSTRUCTOR.setAccessible(true);
-      GENERIC_COUNTER_CONSTRUCTOR =
-          genericCounterCls.getDeclaredConstructor(String.class, String.class, Long.TYPE);
-      GENERIC_COUNTER_CONSTRUCTOR.setAccessible(true);
-
-      if (useV21) {
-        MAP_CONTEXT_CONSTRUCTOR = innerMapContextCls.getConstructor(mapCls, MapContext.class);
-        MAP_CONTEXT_IMPL_CONSTRUCTOR = mapContextCls.getDeclaredConstructor(
-            Configuration.class,
-            TaskAttemptID.class,
-            RecordReader.class,
-            RecordWriter.class,
-            OutputCommitter.class,
-            StatusReporter.class,
-            InputSplit.class);
-        MAP_CONTEXT_IMPL_CONSTRUCTOR.setAccessible(true);
-        WRAPPED_CONTEXT_FIELD = innerMapContextCls.getDeclaredField("mapContext");
-        WRAPPED_CONTEXT_FIELD.setAccessible(true);
-        try {
-          Class<?> taskAttemptContextClass = Class.forName(PACKAGE + ".TaskAttemptContext");
-          Method getCounterMethodForTaskAttemptContext =
-              taskAttemptContextClass.getMethod("getCounter", String.class, String.class);
-
-          COUNTER_METHODS_BY_CLASS.put(taskAttemptContextClass, getCounterMethodForTaskAttemptContext);
-        } catch (ClassNotFoundException e) {
-          Class<?> taskInputOutputContextClass = Class.forName(PACKAGE + ".TaskInputOutputContext");
-          Method getCounterMethodForTaskInputOutputContextClass =
-              taskInputOutputContextClass.getMethod("getCounter", String.class, String.class);
-
-          COUNTER_METHODS_BY_CLASS.put(
-              taskInputOutputContextClass, getCounterMethodForTaskInputOutputContextClass);
-        }
-      } else {
-        MAP_CONTEXT_CONSTRUCTOR = innerMapContextCls.getConstructor(
-            mapCls,
-            Configuration.class,
-            TaskAttemptID.class,
-            RecordReader.class,
-            RecordWriter.class,
-            OutputCommitter.class,
-            StatusReporter.class,
-            InputSplit.class);
-        MAP_CONTEXT_IMPL_CONSTRUCTOR = null;
-        WRAPPED_CONTEXT_FIELD = null;
-
-        COUNTER_METHODS_BY_CLASS.put(
-            taskIOContextCls, taskIOContextCls.getMethod("getCounter", String.class, String.class));
-      }
-      MAP_CONTEXT_CONSTRUCTOR.setAccessible(true);
-      READER_FIELD = mapContextCls.getDeclaredField("reader");
-      READER_FIELD.setAccessible(true);
-      WRITER_FIELD = taskIOContextCls.getDeclaredField("output");
-      WRITER_FIELD.setAccessible(true);
-      OUTER_MAP_FIELD = innerMapContextCls.getDeclaredField("this$0");
-      OUTER_MAP_FIELD.setAccessible(true);
-      GET_CONFIGURATION_METHOD = Class.forName(PACKAGE + ".JobContext").getMethod("getConfiguration");
-      INCREMENT_COUNTER_METHOD = Class.forName(PACKAGE + ".Counter").getMethod("increment", Long.TYPE);
-    } catch (SecurityException e) {
-      throw new IllegalArgumentException("Can't run constructor ", e);
-    } catch (NoSuchMethodException e) {
-      throw new IllegalArgumentException("Can't find constructor ", e);
-    } catch (NoSuchFieldException e) {
-      throw new IllegalArgumentException("Can't find field ", e);
-    } catch (ClassNotFoundException e) {
-      throw new IllegalArgumentException("Can't find class", e);
-    }
-  }
-
   /**
-   * Creates JobContext from a JobConf and jobId using the correct constructor
-   * for based on Hadoop version. <code>jobId</code> could be null.
+   * Creates a JobContext from a configuration and job ID. {@code jobId} may be null.
    *
    * @param conf  a configuration
    * @param jobId a job id
    * @return a job context
    */
   public static JobContext newJobContext(Configuration conf, JobID jobId) {
-    try {
-      return (JobContext) JOB_CONTEXT_CONSTRUCTOR.newInstance(conf, jobId);
-    } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
-      throw new IllegalArgumentException("Can't instantiate JobContext", e);
-    }
+    return new JobContextImpl(conf, jobId);
   }
 
   /**
-   * Creates TaskAttemptContext from a JobConf and jobId using the correct
-   * constructor for based on Hadoop version.
+   * Creates a TaskAttemptContext from a configuration and task attempt ID.
    *
    * @param conf          a configuration
    * @param taskAttemptId a task attempt id
    * @return a task attempt context
    */
   public static TaskAttemptContext newTaskAttemptContext(Configuration conf, TaskAttemptID taskAttemptId) {
-    try {
-      return (TaskAttemptContext) TASK_CONTEXT_CONSTRUCTOR.newInstance(conf, taskAttemptId);
-    } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
-      throw new IllegalArgumentException("Can't instantiate TaskAttemptContext", e);
-    }
+    return new TaskAttemptContextImpl(conf, taskAttemptId);
   }
 
   /**
    * @param name        a string name
    * @param displayName a string display name
    * @param value       an initial value
-   * @return with Hadoop 2 : <code>new GenericCounter(args)</code>,<br>
-   * with Hadoop 1 : <code>new Counter(args)</code>
+   * @return a counter with the given name, display name and initial value
    */
   public static Counter newGenericCounter(String name, String displayName, long value) {
-    try {
-      return (Counter) GENERIC_COUNTER_CONSTRUCTOR.newInstance(name, displayName, value);
-    } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
-      throw new IllegalArgumentException("Can't instantiate Counter", e);
-    }
+    return new GenericCounter(name, displayName, value);
   }
 
   /**
-   * Invoke getConfiguration() method on JobContext. Works with both
-   * Hadoop 1 and 2.
+   * Returns the configuration of a JobContext.
    *
    * @param context a job context
    * @return the context's configuration
    */
   public static Configuration getConfiguration(JobContext context) {
-    try {
-      return (Configuration) GET_CONFIGURATION_METHOD.invoke(context);
-    } catch (IllegalAccessException | InvocationTargetException e) {
-      throw new IllegalArgumentException("Can't invoke method", e);
-    }
+    return context.getConfiguration();
   }
 
   public static Counter getCounter(TaskAttemptContext context, String groupName, String counterName) {
-    Method counterMethod = findCounterMethod(context);
-    return (Counter) invoke(counterMethod, context, groupName, counterName);
-  }
-
-  public static boolean hasCounterMethod(TaskAttemptContext context) {
-    return findCounterMethod(context) != null;
-  }
-
-  private static Method findCounterMethod(TaskAttemptContext context) {
-    if (context != null) {
-      if (COUNTER_METHODS_BY_CLASS.containsKey(context.getClass())) {
-        return COUNTER_METHODS_BY_CLASS.get(context.getClass());
-      }
-
-      try {
-        Method method = context.getClass().getMethod("getCounter", String.class, String.class);
-        if (method.getReturnType().isAssignableFrom(Counter.class)) {
-          COUNTER_METHODS_BY_CLASS.put(context.getClass(), method);
-          return method;
-        }
-      } catch (NoSuchMethodException e) {
-        return null;
-      }
-    }
-
-    return null;
+    return context.getCounter(groupName, counterName);
   }
 
   /**
-   * Invokes a method and rethrows any exception as runtime exceptions.
+   * Returns whether a context is available for counter access. All supported
+   * TaskAttemptContext implementations expose counter methods.
    *
-   * @param method a method
-   * @param obj    an object to run method on
-   * @param args   an array of arguments to the method
-   * @return the result of the method call
+   * @param context a task attempt context, or null
+   * @return true if the context is non-null
    */
-  private static Object invoke(Method method, Object obj, Object... args) {
-    try {
-      return method.invoke(obj, args);
-    } catch (IllegalAccessException | InvocationTargetException e) {
-      throw new IllegalArgumentException("Can't invoke method " + method.getName(), e);
-    }
+  public static boolean hasCounterMethod(TaskAttemptContext context) {
+    return context != null;
   }
 
   public static void incrementCounter(Counter counter, long increment) {
-    invoke(INCREMENT_COUNTER_METHOD, counter, increment);
+    counter.increment(increment);
   }
 }
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/counters/mapreduce/MapReduceCounterAdapter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/counters/mapreduce/MapReduceCounterAdapter.java
index 4334d2f14d..ca7b78baf8 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/counters/mapreduce/MapReduceCounterAdapter.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/counters/mapreduce/MapReduceCounterAdapter.java
@@ -19,7 +19,6 @@
 package org.apache.parquet.hadoop.util.counters.mapreduce;
 
 import org.apache.hadoop.mapreduce.Counter;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.apache.parquet.hadoop.util.counters.ICounter;
 
 /**
@@ -34,7 +33,7 @@ public MapReduceCounterAdapter(Counter adaptee) {
 
   @Override
   public void increment(long val) {
-    ContextUtil.incrementCounter(adaptee, val);
+    adaptee.increment(val);
   }
 
   @Override
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/counters/mapreduce/MapReduceCounterLoader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/counters/mapreduce/MapReduceCounterLoader.java
index a9f71ecf8d..eba15c662a 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/counters/mapreduce/MapReduceCounterLoader.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/counters/mapreduce/MapReduceCounterLoader.java
@@ -19,7 +19,6 @@
 package org.apache.parquet.hadoop.util.counters.mapreduce;
 
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.apache.parquet.hadoop.util.counters.BenchmarkCounter;
 import org.apache.parquet.hadoop.util.counters.CounterLoader;
 import org.apache.parquet.hadoop.util.counters.ICounter;
@@ -37,8 +36,8 @@ public MapReduceCounterLoader(TaskAttemptContext context) {
 
   @Override
   public ICounter getCounterByNameAndFlag(String groupName, String counterName, String counterFlag) {
-    if (ContextUtil.getConfiguration(context).getBoolean(counterFlag, true)) {
-      return new MapReduceCounterAdapter(ContextUtil.getCounter(context, groupName, counterName));
+    if (context.getConfiguration().getBoolean(counterFlag, true)) {
+      return new MapReduceCounterAdapter(context.getCounter(groupName, counterName));
     } else {
       return new BenchmarkCounter.NullCounter();
     }
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/DeprecatedInputFormatTest.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/DeprecatedInputFormatTest.java
index 0a1878d436..76e1b1ebc2 100644
--- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/DeprecatedInputFormatTest.java
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/DeprecatedInputFormatTest.java
@@ -61,7 +61,6 @@
 import org.apache.parquet.hadoop.mapred.Container;
 import org.apache.parquet.hadoop.mapred.DeprecatedParquetInputFormat;
 import org.apache.parquet.hadoop.metadata.CompressionCodecName;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.apache.parquet.schema.MessageTypeParser;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -320,7 +319,7 @@ public static class ReadMapper extends Mapper<LongWritable, Text, Void, Group> {
     private SimpleGroupFactory factory;
 
     protected void setup(Context context) throws IOException, InterruptedException {
-      factory = new SimpleGroupFactory(GroupWriteSupport.getSchema(ContextUtil.getConfiguration(context)));
+      factory = new SimpleGroupFactory(GroupWriteSupport.getSchema(context.getConfiguration()));
     }
 
     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatColumnProjection.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatColumnProjection.java
index d4fb01022e..d6bd142391 100644
--- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatColumnProjection.java
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatColumnProjection.java
@@ -43,7 +43,6 @@
 import org.apache.parquet.example.data.simple.SimpleGroupFactory;
 import org.apache.parquet.hadoop.example.ExampleInputFormat;
 import org.apache.parquet.hadoop.example.ExampleOutputFormat;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.apache.parquet.io.api.Binary;
 import org.apache.parquet.schema.MessageType;
 import org.apache.parquet.schema.Types;
@@ -96,7 +95,7 @@ public static void setBytesReadCounter(Counter bytesRead) {
     @Override
     protected void map(Void key, Group value, Context context) throws IOException, InterruptedException {
       // Do nothing. The test uses Hadoop FS counters for verification.
-      setBytesReadCounter(ContextUtil.getCounter(context, "parquet", "bytesread"));
+      setBytesReadCounter(context.getCounter("parquet", "bytesread"));
     }
   }
 
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java
index 4191703131..60456f5667 100644
--- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java
@@ -57,6 +57,7 @@
 import org.apache.hadoop.mapreduce.RecordReader;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
 import org.apache.parquet.ParquetReadOptions;
 import org.apache.parquet.Version;
 import org.apache.parquet.bytes.BytesInput;
@@ -93,7 +94,6 @@
 import org.apache.parquet.hadoop.metadata.GlobalMetaData;
 import org.apache.parquet.hadoop.metadata.ParquetMetadata;
 import org.apache.parquet.hadoop.metadata.StrictKeyValueMetadataMergeStrategy;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.apache.parquet.hadoop.util.HadoopInputFile;
 import org.apache.parquet.hadoop.util.HadoopOutputFile;
 import org.apache.parquet.hadoop.util.HiddenFileFilter;
@@ -387,7 +387,7 @@ public void testWriteReadWithRecordReader(boolean vectoredRead) throws Exception
     ParquetInputFormat input = new ParquetInputFormat();
     configuration.set(READ_SUPPORT_CLASS, GroupReadSupport.class.getName());
     TaskAttemptID taskAttemptID = TaskAttemptID.forName("attempt_0_1_m_1_1");
-    TaskAttemptContext taskContext = ContextUtil.newTaskAttemptContext(configuration, taskAttemptID);
+    TaskAttemptContext taskContext = new TaskAttemptContextImpl(configuration, taskAttemptID);
     RecordReader<Void, ArrayWritable> reader = input.createRecordReader(split, taskContext);
     assertThat(reader).isInstanceOf(ParquetRecordReader.class);
     // RowGroup.file_offset is checked here
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/codec/CodecConfigTest.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/codec/CodecConfigTest.java
index 77d6ccd223..1946453607 100644
--- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/codec/CodecConfigTest.java
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/codec/CodecConfigTest.java
@@ -28,9 +28,9 @@
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.apache.hadoop.mapreduce.TaskAttemptID;
 import org.apache.hadoop.mapreduce.TaskID;
+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
 import org.apache.parquet.hadoop.ParquetOutputFormat;
 import org.apache.parquet.hadoop.metadata.CompressionCodecName;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.junit.jupiter.api.Test;
 
 public class CodecConfigTest {
@@ -55,8 +55,8 @@ public void shouldUseParquetFlagToSetCodec(String codecNameStr, CompressionCodec
     Job job = new Job();
     Configuration conf = job.getConfiguration();
     conf.set(ParquetOutputFormat.COMPRESSION, codecNameStr);
-    TaskAttemptContext task = ContextUtil.newTaskAttemptContext(
-        conf, new TaskAttemptID(new TaskID(new JobID("test", 1), false, 1), 1));
+    TaskAttemptContext task =
+        new TaskAttemptContextImpl(conf, new TaskAttemptID(new TaskID(new JobID("test", 1), false, 1), 1));
     assertThat(CodecConfig.from(task).getCodec()).isEqualTo(expectedCodec);
 
     // Test mapred API
@@ -72,8 +72,8 @@ public void shouldUseHadoopFlagToSetCodec(String codecClassStr, CompressionCodec
     Configuration conf = job.getConfiguration();
     conf.setBoolean("mapred.output.compress", true);
     conf.set("mapred.output.compression.codec", codecClassStr);
-    TaskAttemptContext task = ContextUtil.newTaskAttemptContext(
-        conf, new TaskAttemptID(new TaskID(new JobID("test", 1), false, 1), 1));
+    TaskAttemptContext task =
+        new TaskAttemptContextImpl(conf, new TaskAttemptID(new TaskID(new JobID("test", 1), false, 1), 1));
     assertThat(CodecConfig.from(task).getCodec()).isEqualTo(expectedCodec);
 
     // Test mapred API
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/example/TestInputOutputFormat.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/example/TestInputOutputFormat.java
index 203e7ab91a..9587f1f0a7 100644
--- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/example/TestInputOutputFormat.java
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/example/TestInputOutputFormat.java
@@ -55,7 +55,6 @@
 import org.apache.parquet.hadoop.api.InitContext;
 import org.apache.parquet.hadoop.api.ReadSupport;
 import org.apache.parquet.hadoop.metadata.CompressionCodecName;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.apache.parquet.schema.MessageTypeParser;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -140,7 +139,7 @@ public static class ReadMapper extends Mapper<LongWritable, Text, Void, Group> {
 
     protected void setup(org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, Void, Group>.Context context)
         throws java.io.IOException, InterruptedException {
-      factory = new SimpleGroupFactory(GroupWriteSupport.getSchema(ContextUtil.getConfiguration(context)));
+      factory = new SimpleGroupFactory(GroupWriteSupport.getSchema(context.getConfiguration()));
     }
 
     protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Void, Group>.Context context)
diff --git a/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoParquetInputFormat.java b/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoParquetInputFormat.java
index f108df981d..a9444297f4 100644
--- a/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoParquetInputFormat.java
+++ b/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoParquetInputFormat.java
@@ -21,7 +21,6 @@
 import com.google.protobuf.MessageOrBuilder;
 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.
@@ -32,6 +31,6 @@ public ProtoParquetInputFormat() {
   }
 
   public static void setRequestedProjection(Job job, String requestedProjection) {
-    ProtoReadSupport.setRequestedProjection(ContextUtil.getConfiguration(job), requestedProjection);
+    ProtoReadSupport.setRequestedProjection(job.getConfiguration(), requestedProjection);
   }
 }
diff --git a/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoParquetOutputFormat.java b/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoParquetOutputFormat.java
index c79c8b62b7..3950dfc475 100644
--- a/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoParquetOutputFormat.java
+++ b/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoParquetOutputFormat.java
@@ -22,7 +22,6 @@
 import com.google.protobuf.MessageOrBuilder;
 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 Protocol Buffer Parquet files.
@@ -38,7 +37,7 @@
 public class ProtoParquetOutputFormat<T extends MessageOrBuilder> extends ParquetOutputFormat<T> {
 
   public static void setProtobufClass(Job job, Class<? extends Message> protoClass) {
-    ProtoWriteSupport.setSchema(ContextUtil.getConfiguration(job), protoClass);
+    ProtoWriteSupport.setSchema(job.getConfiguration(), protoClass);
   }
 
   public ProtoParquetOutputFormat(Class<? extends Message> msg) {
diff --git a/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ParquetThriftBytesOutputFormat.java b/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ParquetThriftBytesOutputFormat.java
index 8b137e3e29..611d200d75 100644
--- a/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ParquetThriftBytesOutputFormat.java
+++ b/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ParquetThriftBytesOutputFormat.java
@@ -22,7 +22,6 @@
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.parquet.hadoop.ParquetOutputFormat;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.apache.parquet.thrift.FieldIgnoredHandler;
 import org.apache.thrift.TBase;
 import org.apache.thrift.protocol.TProtocol;
@@ -34,15 +33,15 @@
 public class ParquetThriftBytesOutputFormat extends ParquetOutputFormat<BytesWritable> {
 
   public static void setThriftClass(Job job, Class<? extends TBase<?, ?>> thriftClass) {
-    TBaseWriteSupport.setThriftClass(ContextUtil.getConfiguration(job), thriftClass);
+    TBaseWriteSupport.setThriftClass(job.getConfiguration(), thriftClass);
   }
 
   public static Class<? extends TBase<?, ?>> getThriftClass(Job job) {
-    return TBaseWriteSupport.getThriftClass(ContextUtil.getConfiguration(job));
+    return TBaseWriteSupport.getThriftClass(job.getConfiguration());
   }
 
   public static <U extends TProtocol> void setTProtocolClass(Job job, Class<U> tProtocolClass) {
-    ThriftBytesWriteSupport.setTProtocolClass(ContextUtil.getConfiguration(job), tProtocolClass);
+    ThriftBytesWriteSupport.setTProtocolClass(job.getConfiguration(), tProtocolClass);
   }
 
   /**
diff --git a/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ParquetThriftOutputFormat.java b/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ParquetThriftOutputFormat.java
index 98d557feb5..0e953120b3 100644
--- a/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ParquetThriftOutputFormat.java
+++ b/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ParquetThriftOutputFormat.java
@@ -20,7 +20,6 @@
 
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.parquet.hadoop.ParquetOutputFormat;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.apache.thrift.TBase;
 
 /**
@@ -29,11 +28,11 @@
 public class ParquetThriftOutputFormat<T extends TBase<?, ?>> extends ParquetOutputFormat<T> {
 
   public static void setThriftClass(Job job, Class<? extends TBase<?, ?>> thriftClass) {
-    ThriftWriteSupport.setThriftClass(ContextUtil.getConfiguration(job), thriftClass);
+    ThriftWriteSupport.setThriftClass(job.getConfiguration(), thriftClass);
   }
 
   public static Class<? extends TBase<?, ?>> getThriftClass(Job job) {
-    return ThriftWriteSupport.getThriftClass(ContextUtil.getConfiguration(job));
+    return ThriftWriteSupport.getThriftClass(job.getConfiguration());
   }
 
   public ParquetThriftOutputFormat() {
diff --git a/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestParquetToThriftReadWriteAndProjection.java b/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestParquetToThriftReadWriteAndProjection.java
index 5e546072e1..04417c10f3 100644
--- a/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestParquetToThriftReadWriteAndProjection.java
+++ b/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestParquetToThriftReadWriteAndProjection.java
@@ -42,8 +42,9 @@
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.apache.hadoop.mapreduce.TaskAttemptID;
 import org.apache.hadoop.mapreduce.TaskID;
+import org.apache.hadoop.mapreduce.task.JobContextImpl;
+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
 import org.apache.parquet.hadoop.api.ReadSupport;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.apache.parquet.thrift.test.RequiredListFixture;
 import org.apache.parquet.thrift.test.RequiredMapFixture;
 import org.apache.parquet.thrift.test.RequiredPrimitiveFixture;
@@ -350,7 +351,7 @@ private void shouldDoProjectionWithThriftColumnFilter(
     final TProtocolFactory protocolFactory = new TCompactProtocol.Factory();
     final TaskAttemptID taskId = new TaskAttemptID("local", 0, true, 0, 0);
     final ThriftToParquetFileWriter w = new ThriftToParquetFileWriter(
-        parquetFile, ContextUtil.newTaskAttemptContext(conf, taskId), protocolFactory, thriftClass);
+        parquetFile, new TaskAttemptContextImpl(conf, taskId), protocolFactory, thriftClass);
     final ByteArrayOutputStream baos = new ByteArrayOutputStream();
     final TProtocol protocol = protocolFactory.getProtocol(new TIOStreamTransport(baos));
 
@@ -363,12 +364,11 @@ private void shouldDoProjectionWithThriftColumnFilter(
     job.setInputFormatClass(ParquetThriftInputFormat.class);
     ParquetThriftInputFormat.setInputPaths(job, parquetFile);
     final JobID jobID = new JobID("local", 1);
-    List<InputSplit> splits =
-        parquetThriftInputFormat.getSplits(ContextUtil.newJobContext(ContextUtil.getConfiguration(job), jobID));
+    List<InputSplit> splits = parquetThriftInputFormat.getSplits(new JobContextImpl(job.getConfiguration(), jobID));
     T readValue = null;
     for (InputSplit split : splits) {
-      TaskAttemptContext taskAttemptContext = ContextUtil.newTaskAttemptContext(
-          ContextUtil.getConfiguration(job), new TaskAttemptID(new TaskID(jobID, true, 1), 0));
+      TaskAttemptContext taskAttemptContext = new TaskAttemptContextImpl(
+          job.getConfiguration(), new TaskAttemptID(new TaskID(jobID, true, 1), 0));
       try (final RecordReader<Void, T> reader =
           parquetThriftInputFormat.createRecordReader(split, taskAttemptContext)) {
         reader.initialize(split, taskAttemptContext);
diff --git a/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestThriftToParquetFileWriter.java b/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestThriftToParquetFileWriter.java
index f24c2c3ff9..f79daa7ab5 100644
--- a/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestThriftToParquetFileWriter.java
+++ b/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestThriftToParquetFileWriter.java
@@ -38,6 +38,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
 import org.apache.parquet.column.statistics.BinaryStatistics;
 import org.apache.parquet.column.statistics.BooleanStatistics;
 import org.apache.parquet.column.statistics.DoubleStatistics;
@@ -51,7 +52,6 @@
 import org.apache.parquet.hadoop.metadata.BlockMetaData;
 import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
 import org.apache.parquet.hadoop.metadata.ParquetMetadata;
-import org.apache.parquet.hadoop.util.ContextUtil;
 import org.apache.parquet.io.api.Binary;
 import org.apache.parquet.schema.MessageType;
 import org.apache.parquet.thrift.ParquetWriteProtocol;
@@ -375,8 +375,7 @@ private ParquetReader<Group> createRecordReader(Path parquetFilePath) throws IOE
     TProtocolFactory protocolFactory = new TCompactProtocol.Factory();
     TaskAttemptID taskId = new TaskAttemptID("local", 0, true, 0, 0);
     ThriftToParquetFileWriter w = new ThriftToParquetFileWriter(
-        fileToCreate, ContextUtil.newTaskAttemptContext(conf, taskId), protocolFactory, (Class<
-                ? extends TBase<?, ?>>)
+        fileToCreate, new TaskAttemptContextImpl(conf, taskId), protocolFactory, (Class<? extends TBase<?, ?>>)
             tObjs[0].getClass());
 
     for (T tObj : tObjs) {