Merge remote-tracking branch 'upstream/master' into spark-36447

Author: maryannxue

README.md

@@ -4,7 +4,7 @@ Spark is a unified analytics engine for large-scale data processing. It provides
 high-level APIs in Scala, Java, Python, and R, and an optimized engine that
 supports general computation graphs for data analysis. It also supports a
 rich set of higher-level tools including Spark SQL for SQL and DataFrames,
-MLlib for machine learning, GraphX for graph processing,
+pandas API on Spark for pandas workloads, MLlib for machine learning, GraphX for graph processing,
 and Structured Streaming for stream processing.
 
 <https://spark.apache.org/>

common/network-common/pom.xml

@@ -91,6 +91,10 @@
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-crypto</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.google.crypto.tink</groupId>
+      <artifactId>tink</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.roaringbitmap</groupId>
       <artifactId>RoaringBitmap</artifactId>

common/network-common/src/main/java/org/apache/spark/network/client/StreamCallbackWithID.java

@@ -17,6 +17,16 @@
 
 package org.apache.spark.network.client;
 
+import java.nio.ByteBuffer;
+
 public interface StreamCallbackWithID extends StreamCallback {
   String getID();
+
+  /**
+   * Response to return to client upon the completion of a stream. Currently only invoked in
+   * {@link org.apache.spark.network.server.TransportRequestHandler#processStreamUpload}
+   */
+  default ByteBuffer getCompletionResponse() {
+    return ByteBuffer.allocate(0);
+  }
 }

common/network-common/src/main/java/org/apache/spark/network/crypto/AuthClientBootstrap.java

@@ -105,15 +105,15 @@ private void doSparkAuth(TransportClient client, Channel channel)
 
     String secretKey = secretKeyHolder.getSecretKey(appId);
     try (AuthEngine engine = new AuthEngine(appId, secretKey, conf)) {
-      ClientChallenge challenge = engine.challenge();
+      AuthMessage challenge = engine.challenge();
       ByteBuf challengeData = Unpooled.buffer(challenge.encodedLength());
       challenge.encode(challengeData);
 
       ByteBuffer responseData =
           client.sendRpcSync(challengeData.nioBuffer(), conf.authRTTimeoutMs());
-      ServerResponse response = ServerResponse.decodeMessage(responseData);
+      AuthMessage response = AuthMessage.decodeMessage(responseData);
 
-      engine.validate(response);
+      engine.deriveSessionCipher(challenge, response);
       engine.sessionCipher().addToChannel(channel);
     }
   }