HDFS-16644. java.io.IOException Invalid token in javax.security.sasl.qop (#5962)

(cherry picked from commit 9d93d08a388badb3a7b4211f9ff4ae3b783869d7)
(cherry picked from commit cc0580c6ac42573977074476f52009e2debe6ab9)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/SaslDataTransferServer.java

@@ -30,6 +30,7 @@ import java.io.InputStream;
 import java.io.OutputStream;
 import java.util.List;
 import java.util.Map;
+import java.util.TreeMap;
 
 import javax.security.auth.callback.Callback;
 import javax.security.auth.callback.CallbackHandler;
@@ -378,14 +379,16 @@ public class SaslDataTransferServer {
       SaslMessageWithHandshake message = readSaslMessageWithHandshakeSecret(in);
       byte[] secret = message.getSecret();
       String bpid = message.getBpid();
+      Map<String, String> dynamicSaslProps = new TreeMap<>(saslProps);
       if (secret != null || bpid != null) {
         // sanity check, if one is null, the other must also not be null
         assert(secret != null && bpid != null);
         String qop = new String(secret, Charsets.UTF_8);
         saslProps.put(Sasl.QOP, qop);
+        dynamicSaslProps.put(Sasl.QOP, qop);
       }
       SaslParticipant sasl = SaslParticipant.createServerSaslParticipant(
-          saslProps, callbackHandler);
+          dynamicSaslProps, callbackHandler);
 
       byte[] remoteResponse = message.getPayload();
       byte[] localResponse = sasl.evaluateChallengeOrResponse(remoteResponse);
@@ -398,7 +401,7 @@ public class SaslDataTransferServer {
       localResponse = sasl.evaluateChallengeOrResponse(remoteResponse);
 
       // SASL handshake is complete
-      checkSaslComplete(sasl, saslProps);
+      checkSaslComplete(sasl, dynamicSaslProps);
 
       CipherOption cipherOption = null;
       negotiatedQOP = sasl.getNegotiatedQop();