[CELEBORN-1526] Fix MR plugin can not run on Hadoop 3.1.0

### What changes were proposed in this pull request?
To fix an NPE when using Celeborn on Hadoop 3.1.0

### Why are the changes needed?
Adapt to API change.

### Does this PR introduce _any_ user-facing change?
NO.

### How was this patch tested?
Cluster test.

Closes apache#2647 from FMX/b1526.

Authored-by: mingji <fengmingxiao.fmx@alibaba-inc.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>

client-mr/mr/src/main/java/org/apache/hadoop/mapreduce/task/reduce/CelebornShuffleConsumer.java

@@ -102,7 +102,7 @@ public void init(Context<K, V> context) {
   private ShuffleClientMetrics createMetrics(
       org.apache.hadoop.mapreduce.TaskAttemptID taskAttemptID, JobConf jobConf)
       throws NoSuchMethodException {
-    // for hadoop 3
+    // for hadoop 3.1+ see MAPREDUCE-6861
     try {
       return DynMethods.builder("create")
           .impl(
@@ -114,6 +114,16 @@ private ShuffleClientMetrics createMetrics(
     } catch (Exception e) {
       // ignore this exception because the createMetrics might use hadoop2
     }
+
+    // for hadoop 3.1 see MAPREDUCE-6526
+    try {
+      return DynMethods.builder("create")
+          .impl(ShuffleClientMetrics.class)
+          .buildStaticChecked()
+          .invoke(taskAttemptID, jobConf);
+    } catch (Exception e) {
+    }
+
     // for hadoop 2
     return DynConstructors.builder(ShuffleClientMetrics.class)
         .hiddenImpl(new Class[] {org.apache.hadoop.mapreduce.TaskAttemptID.class, JobConf.class})

client-mr/mr/src/main/java/org/apache/hadoop/mapreduce/task/reduce/CelebornShuffleFetcher.java

@@ -78,13 +78,17 @@ public void fetchAndMerge() {
         // If merge is on, block
         merger.waitForResource();
         // Do shuffle
-        metrics.threadBusy();
+        if (metrics != null) {
+          metrics.threadBusy();
+        }
         // read blocks
         fetchToLocalAndMerge();
       } catch (Exception e) {
         logger.error("Celeborn shuffle fetcher fetch data failed.", e);
       } finally {
-        metrics.threadFree();
+        if (metrics != null) {
+          metrics.threadFree();
+        }
       }
     }
   }
@@ -134,7 +138,9 @@ private void fetchToLocalAndMerge() throws IOException {
       reporter.progress();
     } else {
       celebornInputStream.close();
-      metrics.inputBytes(inputShuffleSize);
+      if (metrics != null) {
+        metrics.inputBytes(inputShuffleSize);
+      }
       logger.info("reduce task {} read {} bytes", reduceId, inputShuffleSize);
       stopped = true;
     }