[GitHub] [spark] zhouyejoe commented on a diff in pull request #40843: [SPARK-43179][SHUFFLE] Allowing apps to control whether their metadata gets saved in the db by the External Shuffle Service

["zhouyejoe (via GitHub)" <gi...@apache.org>]

zhouyejoe commented on code in PR #40843:
URL: https://github.com/apache/spark/pull/40843#discussion_r1171719972


##########
common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java:
##########
@@ -407,10 +415,32 @@ private static byte[] dbAppKey(AppId appExecId) throws IOException {
   public void initializeApplication(ApplicationInitializationContext context) {
     String appId = context.getApplicationId().toString();
     try {
-      ByteBuffer shuffleSecret = context.getApplicationDataForService();
+      ByteBuffer appServiceData = context.getApplicationDataForService();
+      AppId fullId = new AppId(appId);

Review Comment:
   Nit: This can be moved to line 444.



##########
common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java:
##########
@@ -407,10 +415,32 @@ private static byte[] dbAppKey(AppId appExecId) throws IOException {
   public void initializeApplication(ApplicationInitializationContext context) {
     String appId = context.getApplicationId().toString();
     try {
-      ByteBuffer shuffleSecret = context.getApplicationDataForService();
+      ByteBuffer appServiceData = context.getApplicationDataForService();
+      AppId fullId = new AppId(appId);
+      String payload = JavaUtils.bytesToString(appServiceData);
+      String shuffleSecret;
+      boolean updateDb = true;

Review Comment:
   Nit: This boolean is not necessary. We can call "AppsWithRecoveryDisabled.isRecoveryEnabledForApp(appId)" on line 443 instead. And line 429 can also be removed.



##########
common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java:
##########
@@ -407,10 +415,32 @@ private static byte[] dbAppKey(AppId appExecId) throws IOException {
   public void initializeApplication(ApplicationInitializationContext context) {
     String appId = context.getApplicationId().toString();
     try {
-      ByteBuffer shuffleSecret = context.getApplicationDataForService();
+      ByteBuffer appServiceData = context.getApplicationDataForService();
+      AppId fullId = new AppId(appId);
+      String payload = JavaUtils.bytesToString(appServiceData);
+      String shuffleSecret;
+      boolean updateDb = true;
+      Map<String, Object> metaInfo;
+      try {
+        metaInfo = mapper.readValue(payload,
+            new TypeReference<Map<String, Object>>() {});
+        Object metadataStorageVal = metaInfo.get(SPARK_SHUFFLE_SERVER_RECOVERY_DISABLED);
+        if (metadataStorageVal != null && (Boolean) metadataStorageVal) {
+          updateDb = false;
+          AppsWithRecoveryDisabled.disableRecoveryOfApp(appId);
+          logger.info("Not saving metadata of application {}", appId);

Review Comment:
   Nit, Wording: Disabling metadata persistency for application {}.



##########
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ExecutorRunnable.scala:
##########
@@ -128,6 +121,33 @@ private[yarn] class ExecutorRunnable(
     }
   }
 
+  private[yarn] def configureServiceData(ctx: ContainerLaunchContext): Unit = {
+    val secretString = securityMgr.getSecretKey()
+
+    val serviceName = sparkConf.get(SHUFFLE_SERVICE_NAME)
+    if (!sparkConf.get(SHUFFLE_SERVER_RECOVERY_DISABLED)) {
+      val secretBytes =
+        if (secretString != null) {
+          // This conversion must match how the YarnShuffleService decodes our secret
+          JavaUtils.stringToBytes(secretString)
+        } else {
+          // Authentication is not enabled, so just provide dummy metadata
+          ByteBuffer.allocate(0)
+        }
+      ctx.setServiceData(Collections.singletonMap(serviceName, secretBytes))
+    } else {
+      val payload = new mutable.HashMap[String, Object]()
+      payload.put(SHUFFLE_SERVER_RECOVERY_DISABLED.key, java.lang.Boolean.TRUE)
+      if (secretString != null) {
+        payload.put(ExecutorRunnable.SECRET_KEY, secretString)

Review Comment:
   Do we need to keep the same behavior here, when the secretString is null, we still put ByteBuffer.allocate(0) there in the payload?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org